How to implement the use of global variable DB

[xhatt]

I am looking for a demo project that can happily use asynchronous sqlalchemy on FastAPI, and then find you for this project. The other views I saw online about the combination of FastAPI and sqlalchemy are usually written like this, and I have to explicitly pass db through parameters to the view function, which makes me feel very uncomfortable.

@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)

I saw your sample code and it doesn't seem like it needs to be explicitly passed to the view function, so I would like to ask how it is implemented internally? Sorry, I don't know English. This English paragraph is all machine translated Chinese.

[h0rn3t]

from fastapi_async_sqlalchemy import db
from sqlalchemy import func
from sqlalchemy.future import select
from sqlalchemy.orm.attributes import flag_modified
from sqlmodel import SQLModel, col, or_
from sqlmodel.sql.expression import Select

from app.core.constants import DEFAULT_PAGINATION_PAGE_NUMBER, DEFAULT_PAGINATION_PAGE_SIZE
from app.schemas.response import PageResponse

ModelType = TypeVar("ModelType", bound=SQLModel)

class BaseCRUD(Generic[ModelType]):
    model: type[ModelType]

    @classmethod
    async def create(
        cls,
        params: dict[str, Any],
        commit: bool = True,
        autoflush: bool = False,
    ) -> ModelType:
        obj = cls.model(**params)
        db.session.add(obj)

        if commit:
            await db.session.commit()
            await db.session.refresh(obj)

        if autoflush:
            await db.session.flush()

        return obj

    @classmethod
    async def get(cls, *args: Any, **kwargs: Any) -> ModelType:
        result = await db.session.execute(select(cls.model).filter(*args).filter_by(**kwargs))
        return result.scalar_one()

    @classmethod
    async def filter(cls, *args: Any, **kwargs: Any) -> list[ModelType]:
        statement = select(cls.model).filter(*args).filter_by(**kwargs)
        result = await db.session.execute(statement)
        return result.scalars().all()

example of crud class using this lib