wip v2 experimentation

hatchet_sdk/runtime/admin.py

@@ -0,0 +1,20 @@
+
+# import hatchet_sdk.v2.callable as sdk
+# import hatchet_sdk.clients.admin as client
+
+# from hatchet_sdk.contracts.workflows_pb2 import (
+#     CreateStepRateLimit,
+#     CreateWorkflowJobOpts,
+#     CreateWorkflowStepOpts,
+#     CreateWorkflowVersionOpts,
+#     DesiredWorkerLabels,
+#     StickyStrategy,
+#     WorkflowConcurrencyOpts,
+#     WorkflowKind,
+# )
+
+# async def put_workflow(callable: sdk.HatchetCallable, client: client.AdminClient):
+#     options = callable._.options
+
+#     kind: WorkflowKind = WorkflowKind.DURABLE if options.durable else WorkflowKind.FUNCTION
+

hatchet_sdk/runtime/registry.py

@@ -0,0 +1,19 @@
+from typing import Dict, List
+
+
+class ActionRegistry:
+
+    _registry: Dict[str, "HatchetCallable"] = dict()
+
+    def register(self, callable: "HatchetCallable") -> str:
+        key = "{namespace}:{name}".format(
+            namespace=callable._.namespace, name=callable._.name
+        )
+        self._registry[key] = callable
+        return key
+
+    def list(self) -> List[str]:
+        return list(self._registry.keys())
+
+
+global_registry = ActionRegistry()