Merge pull request #18 from northpowered/5-http-workers

5 http workers

Author: northpowered

example_app.py

@@ -1,36 +1,62 @@
-# Import `BoostApp` class
-from temporal_boost import BoostApp, BoostLoggerConfig, BoostHTTPRoute
+"""
+For development purposes
+"""
 
+# Import `BoostApp` class
+from temporal_boost import BoostApp, BoostLoggerConfig
 from temporalio import activity
 from temporalio import workflow
+from datetime import timedelta
+from dataclasses import dataclass
 
-
-from aiohttp import web
+from example_asgi_app import fastapi_app
 
 # Create `BoostApp` object
-app: BoostApp = BoostApp(
-    logger_config=BoostLoggerConfig(json=False),
-)
+app: BoostApp = BoostApp(logger_config=BoostLoggerConfig(json=False), use_pydantic=True)
+
+
+@dataclass
+class TestModel:
+    foo: str
+    bar: int
 
 
 # Describe your activities/workflows
 @activity.defn(name="test_boost_activity_1")
-async def test_boost_activity_1(foo: str, bar: str) -> str:
-    app.logger.info("This is built-in logger")
-    return f"1_{foo}{bar}"
+async def test_boost_activity_1(payload: TestModel) -> TestModel:
+    payload.foo = f"{payload.foo}+activity1"
+    payload.bar = payload.bar + 1
+    return payload
 
 
 @activity.defn(name="test_boost_activity_2")
-async def test_boost_activity_2(foo: str, bar: str) -> str:
-    return f"2_{foo}{bar}"
+async def test_boost_activity_2(payload: TestModel) -> TestModel:
+    payload.foo = f"{payload.foo}+activity2"
+    payload.bar = payload.bar + 1
+    return payload
 
 
-@workflow.defn(name="TestCronWorkflow", sandboxed=False)
-class TestCronWorkflow:
+@workflow.defn(sandboxed=False)
+class MyWorkflow:
     @workflow.run
-    async def run(self) -> None:
-        print("With is cron workflow")
-        return None
+    async def run(self, foo: str):
+        start_payload: TestModel = TestModel(foo="hello", bar=0)
+        print(type(start_payload))
+        result_1 = await workflow.execute_activity(
+            test_boost_activity_1,
+            start_payload,
+            task_queue="task_q_1",
+            start_to_close_timeout=timedelta(minutes=1),
+        )
+        print(type(result_1))
+        result_2 = await workflow.execute_activity(
+            test_boost_activity_2,
+            result_1,
+            task_queue="task_q_2",
+            start_to_close_timeout=timedelta(minutes=1),
+        )
+        print(type(result_2))
+        return result_2
 
 
 # Add async workers to your app
@@ -42,28 +68,12 @@ async def run(self) -> None:
     metrics_endpoint="0.0.0.0:9000",
 )
 app.add_worker("worker_2", "task_q_2", activities=[test_boost_activity_2])
-# Example of CRON worker
-app.add_worker(
-    "test_cron",
-    "task_q_3",
-    workflows=[TestCronWorkflow],
-    cron_schedule="* * * * *",
-    cron_runner=TestCronWorkflow.run,
-)
 
+app.add_worker("worker_3", "task_q_3", workflows=[MyWorkflow])
 
-async def aaa(request):
-    import json
-    q: dict = {"foo": "bar"}
-    return web.Response(text=json.dumps(q), content_type="application/json")
+# app.add_http_worker("test_http_worker_!", "0.0.0.0", 8000, routes=[])
 
-
-app.add_http_worker(
-    "test_http_1", host="127.0.0.1", port=8899,
-    routes=[
-        BoostHTTPRoute("/", aaa)
-    ]
-)
+app.add_asgi_worker("asgi_worker", fastapi_app, "0.0.0.0", 8000)
 
 # Run your app and start workers with CLI
 app.run()

example_asgi_app.py

@@ -0,0 +1,10 @@
+from fastapi import FastAPI, Response
+
+
+async def foo():
+    return Response(status_code=200)
+
+
+fastapi_app: FastAPI = FastAPI(docs_url="/doc")
+
+fastapi_app.add_api_route("/foo", foo)

example_starter.py

@@ -0,0 +1,24 @@
+"""
+For development purposes
+"""
+
+import asyncio
+from temporalio.client import Client
+
+
+async def main():
+
+    client = await Client.connect("localhost:7233")
+
+    # Run workflow
+    result = await client.execute_workflow(
+        "MyWorkflow",
+        "blabla",
+        id="pydantic_converter-workflow-id",
+        task_queue="task_q_3",
+    )
+    print(result)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())