mirror of
https://github.com/grillazz/fastapi-sqlalchemy-asyncpg.git
synced 2026-01-17 11:40:39 +03:00
add scheduler middleware
This commit is contained in:
23
app/main.py
23
app/main.py
@@ -1,6 +1,6 @@
|
||||
import asyncpg
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from apscheduler.eventbrokers.redis import RedisEventBroker
|
||||
from apscheduler.datastores.sqlalchemy import SQLAlchemyDataStore
|
||||
from fastapi import FastAPI, Depends
|
||||
from fastapi_cache import FastAPICache
|
||||
from fastapi_cache.backends.redis import RedisBackend
|
||||
@@ -9,11 +9,17 @@ from app.api.nonsense import router as nonsense_router
|
||||
from app.api.shakespeare import router as shakespeare_router
|
||||
from app.api.stuff import router as stuff_router
|
||||
from app.config import settings as global_settings
|
||||
from app.database import engine
|
||||
from app.utils.logging import AppLogger
|
||||
from app.api.user import router as user_router
|
||||
from app.api.health import router as health_router
|
||||
from app.redis import get_redis, get_cache
|
||||
from app.services.auth import AuthBearer
|
||||
from app.services.scheduler import SchedulerMiddleware
|
||||
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from apscheduler import AsyncScheduler
|
||||
|
||||
logger = AppLogger().get_logger()
|
||||
|
||||
@@ -60,3 +66,16 @@ app.include_router(
|
||||
tags=["Health, Bearer"],
|
||||
dependencies=[Depends(AuthBearer())],
|
||||
)
|
||||
|
||||
_scheduler_data_store = SQLAlchemyDataStore(engine)
|
||||
_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 not GET meth should reset cache
|
||||
# TODO: every scheduler task which needs to act on database hsould have access to 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
|
||||
|
||||
@@ -17,7 +17,9 @@ def compile_sql_or_scalar(func):
|
||||
async def wrapper(cls, db_session, name, compile_sql=False, *args, **kwargs):
|
||||
stmt = await func(cls, db_session, name, *args, **kwargs)
|
||||
if compile_sql:
|
||||
return stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True})
|
||||
return stmt.compile(
|
||||
dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}
|
||||
)
|
||||
result = await db_session.execute(stmt)
|
||||
return result.scalars().first()
|
||||
|
||||
|
||||
@@ -27,12 +27,12 @@ class User(Base):
|
||||
@password.setter
|
||||
def password(self, password: SecretStr):
|
||||
_password_string = password.get_secret_value().encode("utf-8")
|
||||
self._password = bcrypt.hashpw(
|
||||
_password_string, bcrypt.gensalt()
|
||||
)
|
||||
self._password = bcrypt.hashpw(_password_string, bcrypt.gensalt())
|
||||
|
||||
def check_password(self, password: SecretStr):
|
||||
return bcrypt.checkpw(password.get_secret_value().encode("utf-8"), self._password)
|
||||
return bcrypt.checkpw(
|
||||
password.get_secret_value().encode("utf-8"), self._password
|
||||
)
|
||||
|
||||
@classmethod
|
||||
async def find(cls, database_session: AsyncSession, where_conditions: list[Any]):
|
||||
|
||||
38
app/services/scheduler.py
Normal file
38
app/services/scheduler.py
Normal file
@@ -0,0 +1,38 @@
|
||||
from datetime import datetime
|
||||
|
||||
from starlette.types import ASGIApp, Receive, Scope, Send
|
||||
|
||||
from apscheduler import AsyncScheduler
|
||||
from apscheduler.triggers.interval import IntervalTrigger
|
||||
|
||||
from app.utils.logging import AppLogger
|
||||
|
||||
logger = AppLogger().get_logger()
|
||||
|
||||
|
||||
def tick():
|
||||
|
||||
logger.info(f">>>> Be or not to be...{datetime.now()}")
|
||||
|
||||
|
||||
class SchedulerMiddleware:
|
||||
# TODO: need access to request to be able to get db conn pool
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
app: ASGIApp,
|
||||
scheduler: AsyncScheduler,
|
||||
) -> None:
|
||||
self.app = app
|
||||
self.scheduler = scheduler
|
||||
|
||||
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
|
||||
if scope["type"] == "lifespan":
|
||||
async with self.scheduler:
|
||||
await self.scheduler.add_schedule(
|
||||
tick, IntervalTrigger(seconds=25), id="tick"
|
||||
)
|
||||
await self.scheduler.start_in_background()
|
||||
await self.app(scope, receive, send)
|
||||
else:
|
||||
await self.app(scope, receive, send)
|
||||
@@ -20,5 +20,6 @@ class AppLogger(metaclass=SingletonMeta):
|
||||
class RichConsoleHandler(RichHandler):
|
||||
def __init__(self, width=200, style=None, **kwargs):
|
||||
super().__init__(
|
||||
console=Console(color_system="256", width=width, style=style, stderr=True), **kwargs
|
||||
console=Console(color_system="256", width=width, style=style, stderr=True),
|
||||
**kwargs
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user