1
0
mirror of https://codeberg.org/diginaut/digimarks.git synced 2026-02-04 08:10:26 +01:00

Better declaration of the httpx client

This commit is contained in:
2026-01-03 23:45:39 +01:00
parent fe734d6dd8
commit 9b03d51276

View File

@@ -4,7 +4,7 @@ import logging
from collections.abc import Sequence from collections.abc import Sequence
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from datetime import UTC, datetime from datetime import UTC, datetime
from typing import Annotated from typing import Annotated, AsyncGenerator, cast
import httpx import httpx
from fastapi import Depends, FastAPI, HTTPException, Query, Request from fastapi import Depends, FastAPI, HTTPException, Query, Request
@@ -54,22 +54,32 @@ engine = create_async_engine(f'sqlite+aiosqlite:///{settings.database_file}', co
async def get_session() -> AsyncSession: async def get_session() -> AsyncSession:
"""SQLAlchemy session factory.""" """SQLAlchemy session factory."""
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) async_session = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
async with async_session() as session: async with async_session() as session:
yield session yield session
# Shorter way of getting the DB session in an endpoint
SessionDep = Annotated[AsyncSession, Depends(get_session)] SessionDep = Annotated[AsyncSession, Depends(get_session)]
@asynccontextmanager @asynccontextmanager
async def lifespan(the_app: FastAPI): async def lifespan(the_app: FastAPI) -> AsyncGenerator[None, None]:
"""Upon start, initialise an AsyncClient and assign it to an attribute named requests_client on the app object.""" """Upon start, initialise an AsyncClient and assign it to an attribute named requests_client on the app object."""
the_app.requests_client = httpx.AsyncClient() async with httpx.AsyncClient() as requests_client:
yield the_app.state.requests_client = requests_client
await the_app.requests_client.aclose() yield
await the_app.state.requests_client.aclose()
async def get_requests_client(request: Request) -> httpx.AsyncClient:
"""Get the httpx client from the application object."""
return cast(httpx.AsyncClient, request.app.state.requests_client)
# Shorter way of getting the httpx client in an endpoint
RequestsDep = Annotated[AsyncSession, Depends(get_requests_client)]
app = FastAPI(lifespan=lifespan) app = FastAPI(lifespan=lifespan)
app.mount('/static', StaticFiles(directory=settings.static_dir), name='static') app.mount('/static', StaticFiles(directory=settings.static_dir), name='static')
app.mount('/content/favicons', StaticFiles(directory=settings.favicons_dir), name='favicons') app.mount('/content/favicons', StaticFiles(directory=settings.favicons_dir), name='favicons')