diff --git a/README.md b/README.md
index a321257..74a9e4d 100644
--- a/README.md
+++ b/README.md
@@ -130,11 +130,13 @@ In `pl.read_excel()` “calamine” engine can be used for reading all major typ
(back to top)
### 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).
(back to top)
diff --git a/app/main.py b/app/main.py
index e3ab516..c5f6068 100644
--- a/app/main.py
+++ b/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
diff --git a/app/services/scheduler.py b/app/services/scheduler.py
index 6352fb6..22182ce 100644
--- a/app/services/scheduler.py
+++ b/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
diff --git a/pyproject.toml b/pyproject.toml
index 9b88b57..2910ee0 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -29,6 +29,7 @@ dependencies = [
"websockets==17.3",
"jinja2==3.1.6",
"whenever==0.10.5",
+ "sniffio==1.3.1",
]
[dependency-groups]
diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py
new file mode 100644
index 0000000..a21db8b
--- /dev/null
+++ b/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"
diff --git a/uv.lock b/uv.lock
index e44ae8f..7ef411d 100644
--- a/uv.lock
+++ b/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"