Merge pull request #123 from lucafaggianelli/fix/async-loop

Fix/async loop

README.md

@@ -139,8 +139,14 @@ pip install -r requirements-dev.txt
 for development purposes, it's useful to run the example application:
 ```sh
 cd examples/
-export PYTHONPATH=$(pwd)/..
-uvicorn dummy.app:app --reload --reload-dir ..
+
+# Create a venv for the example app
+python -m venv .venv
+source .venv/bin/activate
+pip install -r requirements
+
+./run.sh
+# or ./run.ps1 on windows
 ```
 
 The React frontend is in the `frontend/` folder, enter the folder

examples/requirements.txt

@@ -0,0 +1,3 @@
+-e ..
+python-dateutil
+pandas

examples/run.ps1

@@ -1,2 +1,2 @@
-$env:PYTHONPATH="$(pwd)/.."
+$env:PYTHONPATH="$(pwd)"
 python src/app.py

examples/run.sh

@@ -1,2 +1,2 @@
-export PYTHONPATH=$(pwd)/..
+export PYTHONPATH=$(pwd)
 python src/app.py

examples/src/app.py

@@ -4,7 +4,7 @@
 
 from plombery import get_app  # noqa: F401
 
-from examples.src import sales_pipeline, sync_pipeline  # noqa: F401
+from src import sales_pipeline, sync_pipeline  # noqa: F401
 
 
 if __name__ == "__main__":

frontend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "plombery",
-  "version": "0.3.0-rc2",
+  "version": "0.3.0",
   "description": "",
   "license": "MIT",
   "author": {

src/plombery/__init__.py

@@ -1,5 +1,6 @@
-import logging
 from typing import List, Type
+import logging
+import os
 
 from apscheduler.schedulers.base import SchedulerAlreadyRunningError
 from pydantic import BaseModel
@@ -20,15 +21,15 @@
 _logger.addHandler(logging.StreamHandler())
 
 
+if os.getenv("DEBUG_APS"):
+    logging.basicConfig()
+    logging.getLogger("apscheduler").setLevel(logging.DEBUG)
+
+
 class _Plombery:
     def __init__(self) -> None:
         self._apply_settings()
 
-        try:
-            orchestrator.start()
-        except SchedulerAlreadyRunningError:
-            pass
-
     def _apply_settings(self):
         for notification in settings.notifications or []:
             self.add_notification_rule(notification)
@@ -51,6 +52,14 @@ async def __call__(self, scope, receive, send):
 _app = _Plombery()
 
 
+@app.on_event("startup")
+def on_fastapi_start():
+    try:
+        orchestrator.start()
+    except SchedulerAlreadyRunningError:
+        pass
+
+
 def get_app():
     return _app
 

src/plombery/_version.py

@@ -1,3 +1,3 @@
 # Keep it aligned with version in package.json
 
-__version__ = "0.3.0-rc2"
+__version__ = "0.3.0"

src/plombery/orchestrator/__init__.py

@@ -1,5 +1,5 @@
-from datetime import datetime, timedelta
 from typing import Any, Dict, Tuple
+from datetime import datetime, timedelta
 
 from apscheduler.executors.asyncio import AsyncIOExecutor
 from apscheduler.job import Job
@@ -37,6 +37,7 @@ def register_pipeline(self, pipeline: Pipeline):
 
             self.scheduler.add_job(
                 id=job_id,
+                name=job_id,
                 func=run,
                 trigger=trigger.schedule,
                 kwargs=dict(pipeline=pipeline, trigger=trigger),