feat: implement async scheduler with lifespan management and update README

README.md

@@ -130,11 +130,13 @@ In `pl.read_excel()` “calamine” engine can be used for reading all major typ
 <p align="right">(<a href="#readme-top">back to top</a>)</p>
 
 ### Worker aware async scheduler
-The project uses the APScheduler library to schedule tasks in the background.
-The APScheduler library is a powerful and flexible in-process task scheduler with Cron-like capabilities.
-It allows you to schedule jobs to run at specific times or intervals, and it supports multiple job stores, triggers, and executors.
-The library is designed to be easy to use and highly configurable, making it suitable for a wide range of use cases.
-It was added to project in version 4.0.0a5 with Redis as event broker and SQLAlchemy as data store.
+The app starts an `AsyncScheduler` as part of FastAPI's lifespan. Each worker shares
+the persistent SQLAlchemy data store in the `scheduler` schema and uses Redis as the
+event broker, allowing APScheduler to coordinate scheduling across workers. The
+`tick-sql-25` schedule runs `app.services.scheduler.tick` every 25 seconds and is
+replaced on startup so its definition stays current. The project uses APScheduler
+4.0.0a6, a pre-release; see the
+[ASGI/FastAPI integration example](https://github.com/agronholm/apscheduler/blob/4.0.0a6/examples/web/asgi_fastapi.py).
 
 <p align="right">(<a href="#readme-top">back to top</a>)</p>
 

app/main.py

@@ -22,6 +22,7 @@ from app.middleware.profiler import ProfilingMiddleware
 from app.redis import get_redis
 from app.services.auth import AuthBearer
 from app.services.chat_agent import build_chat_agent
+from app.services.scheduler import scheduler_lifespan
 
 templates = Jinja2Templates(directory=Path(__file__).parent.parent / "templates")
 
@@ -45,7 +46,8 @@ async def lifespan(app: FastAPI):
         # await app.logger.ainfo(
         #     "Postgres pool created", idle_size=app.postgres_pool.get_idle_size()
         # )
-        yield
+        async with scheduler_lifespan():
+            yield
     except Exception as e:
         await app.logger.aerror("Error during app startup", error=repr(e))
         raise
@@ -96,16 +98,4 @@ def create_app() -> FastAPI:
 
 app = create_app()
 
-# --- Unused/experimental code and TODOs ---
-# from apscheduler import AsyncScheduler
-# from apscheduler.datastores.sqlalchemy import SQLAlchemyDataStore
-# from apscheduler.eventbrokers.redis import RedisEventBroker
-# from app.database import engine
-# from app.services.scheduler import SchedulerMiddleware
-# _scheduler_data_store = SQLAlchemyDataStore(engine, schema="scheduler")
-# _scheduler_event_broker = RedisEventBroker(client_or_url=global_settings.redis_url.unicode_string())
-# _scheduler_himself = AsyncScheduler(_scheduler_data_store, _scheduler_event_broker)
-# app.add_middleware(SchedulerMiddleware, scheduler=_scheduler_himself)
 # TODO: every non-GET method should reset cache
-# TODO: scheduler tasks needing DB should access connection pool via request
-# TODO: https://stackoverflow.com/questions/16053364/make-sure-only-one-worker-launches-the-apscheduler-event-in-a-pyramid-web-app-ru

app/services/scheduler.py

@@ -1,13 +1,16 @@
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
 from datetime import datetime
 
-from apscheduler import AsyncScheduler
+from apscheduler import AsyncScheduler, ConflictPolicy
+from apscheduler.datastores.sqlalchemy import SQLAlchemyDataStore
+from apscheduler.eventbrokers.redis import RedisEventBroker
 from apscheduler.triggers.interval import IntervalTrigger
-from attrs import define
 from rotoger import get_logger
 from sqlalchemy import text
-from starlette.types import ASGIApp, Receive, Scope, Send
 
-from app.database import AsyncSessionFactory
+from app.config import settings as global_settings
+from app.database import AsyncSessionFactory, engine
 
 logger = get_logger()
 
@@ -21,26 +24,19 @@ async def tick():
         return True
 
 
-@define
-class SchedulerMiddleware:
-    app: ASGIApp
-    scheduler: AsyncScheduler
+@asynccontextmanager
+async def scheduler_lifespan() -> AsyncIterator[None]:
+    scheduler = AsyncScheduler(
+        SQLAlchemyDataStore(engine, schema="scheduler"),
+        RedisEventBroker(client_or_url=global_settings.redis_url.unicode_string()),
+    )
 
-    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
-        """
-        Handles the incoming request and schedules tasks if the scope type is 'lifespan'.
-
-        Args:
-            scope (Scope): The ASGI scope dictionary containing request information.
-            receive (Receive): The ASGI receive callable.
-            send (Send): The ASGI send callable.
-        """
-        if scope["type"] == "lifespan":
-            async with self.scheduler:
-                await self.scheduler.add_schedule(
-                    tick, IntervalTrigger(seconds=25), id="tick-sql-25"
-                )
-                await self.scheduler.start_in_background()
-                await self.app(scope, receive, send)
-        else:
-            await self.app(scope, receive, send)
+    async with scheduler:
+        await scheduler.add_schedule(
+            tick,
+            IntervalTrigger(seconds=25),
+            id="tick-sql-25",
+            conflict_policy=ConflictPolicy.replace,
+        )
+        await scheduler.start_in_background()
+        yield

pyproject.toml

@@ -29,6 +29,7 @@ dependencies = [
     "websockets==17.3",
     "jinja2==3.1.6",
     "whenever==0.10.5",
+    "sniffio==1.3.1",
 ]
 
 [dependency-groups]

tests/test_scheduler.py

@@ -0,0 +1,62 @@
+import pytest
+from apscheduler import ConflictPolicy
+from apscheduler.triggers.interval import IntervalTrigger
+
+import app.services.scheduler as scheduler_service
+
+
+@pytest.mark.anyio
+async def test_scheduler_lifespan_starts_and_stops_scheduler(monkeypatch) -> None:
+    events = []
+    data_store = object()
+    event_broker = object()
+
+    class SchedulerStub:
+        async def __aenter__(self):
+            events.append("enter")
+            return self
+
+        async def __aexit__(self, *_args: object) -> None:
+            events.append("exit")
+
+        async def add_schedule(
+            self, func, trigger, *, id: str, conflict_policy: ConflictPolicy
+        ) -> None:
+            events.append(("schedule", func, trigger, id, conflict_policy))
+
+        async def start_in_background(self) -> None:
+            events.append("start")
+
+    scheduler = SchedulerStub()
+
+    def create_data_store(actual_engine, *, schema):
+        assert actual_engine is scheduler_service.engine
+        assert schema == "scheduler"
+        return data_store
+
+    def create_event_broker(*, client_or_url):
+        assert client_or_url == (
+            scheduler_service.global_settings.redis_url.unicode_string()
+        )
+        return event_broker
+
+    def create_scheduler(actual_data_store, actual_event_broker):
+        assert actual_data_store is data_store
+        assert actual_event_broker is event_broker
+        return scheduler
+
+    monkeypatch.setattr(scheduler_service, "SQLAlchemyDataStore", create_data_store)
+    monkeypatch.setattr(scheduler_service, "RedisEventBroker", create_event_broker)
+    monkeypatch.setattr(scheduler_service, "AsyncScheduler", create_scheduler)
+
+    async with scheduler_service.scheduler_lifespan():
+        assert events[0] == "enter"
+        event_type, func, trigger, schedule_id, conflict_policy = events[1]
+        assert event_type == "schedule"
+        assert func is scheduler_service.tick
+        assert isinstance(trigger, IntervalTrigger)
+        assert schedule_id == "tick-sql-25"
+        assert conflict_policy is ConflictPolicy.replace
+        assert events[2:] == ["start"]
+
+    assert events[-1] == "exit"

uv.lock

@@ -341,6 +341,7 @@ dependencies = [
     { name = "redis" },
     { name = "rich" },
     { name = "rotoger" },
+    { name = "sniffio" },
     { name = "sqlalchemy" },
     { name = "uvicorn", extra = ["standard"] },
     { name = "uvloop" },
@@ -385,6 +386,7 @@ requires-dist = [
     { name = "redis", specifier = "==8.1.0" },
     { name = "rich", specifier = "==15.0.0" },
     { name = "rotoger", specifier = "==0.3.0" },
+    { name = "sniffio", specifier = "==1.3.1" },
     { name = "sqlalchemy", specifier = "==2.1.6" },
     { name = "uvicorn", extras = ["standard"], specifier = "==0.54.0" },
     { name = "uvloop", specifier = "==0.23.0" },
@@ -1177,6 +1179,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274", size = 11050, upload-time = "2024-12-04T17:35:26.475Z" },
 ]
 
+[[package]]
+name = "sniffio"
+version = "1.3.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/a2/87/a6771e1546d97e7e041b6ae58d80074f81b7d5121207425c964ddf5cfdbd/sniffio-1.3.1.tar.gz", hash = "sha256:f4324edc670a0f49750a81b895f35c3adb843cca46f0530f79fc1babb23789dc", size = 20372, upload-time = "2024-02-25T23:20:04.057Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/e9/44/75a9c9421471a6c4805dbf2356f7c181a29c1879239abab1ea2cc8f38b40/sniffio-1.3.1-py3-none-any.whl", hash = "sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2", size = 10235, upload-time = "2024-02-25T23:20:01.196Z" },
+]
+
 [[package]]
 name = "sqlalchemy"
 version = "2.1.6"