PersonalPublic/fastapi-sqlalchemy-asyncpg
1
0
Fork 0
mirror of https://github.com/grillazz/fastapi-sqlalchemy-asyncpg.git synced 2025-08-26 16:40:40 +03:00
Code Issues Packages Projects Releases Wiki Activity

refactor SchedulerMiddleware

Browse Source
This commit is contained in:
Jakub Miazek 2024-11-22 14:10:39 +01:00
parent a7c70079b0
commit dd619e70ce
1 changed files with 14 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
app/services/scheduler.py
View File

@ -1,5 +1,7 @@
from datetime import datetime from datetime import datetime
from attrs import define
from sqlalchemy import text from sqlalchemy import text
from starlette.types import ASGIApp, Receive, Scope, Send from starlette.types import ASGIApp, Receive, Scope, Send
from apscheduler import AsyncScheduler from apscheduler import AsyncScheduler
@ -20,16 +22,20 @@ async def tick():
return True return True
@define
class SchedulerMiddleware: class SchedulerMiddleware:
def __init__( app: ASGIApp
self, scheduler: AsyncScheduler
app: ASGIApp,
scheduler: AsyncScheduler,
) -> None:
self.app = app
self.scheduler = scheduler
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 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": if scope["type"] == "lifespan":
async with self.scheduler: async with self.scheduler:
await self.scheduler.add_schedule( await self.scheduler.add_schedule(
@ -39,3 +45,4 @@ class SchedulerMiddleware:
await self.app(scope, receive, send) await self.app(scope, receive, send)
else: else:
await self.app(scope, receive, send) await self.app(scope, receive, send)