db_session as argument of all methods its stupid

[h0rn3t]

better use global db session contextvar, code will be cleaner

[jonra1993]

Hello, @h0rn3t thanks for your sincere feedback 👍. Maybe you agree with me that it does not exist a perfect code and there is always an opportunity to improve it ✌️🙂. The current setup is sharing a database engine and creating an independent database session in each handler (to make sure that the session is started and closed after the request is handled using dependency injection) as is described in the official FastAPI docs here. I agree with you that if we follow the KISS and DRY principles this code can be more simple. I do not think that it is a good idea that handles to share the same session instead create a pool of sessions that can be used and released for each handle as described here.

This recommendation will be added to the TODO list to add it on a next update.

Reference links https://github.com/tiangolo/fastapi/issues/726 https://github.com/h0rn3t/fastapi-async-sqlalchemy https://fastapi.tiangolo.com/advanced/async-sql-databases/#connect-and-disconnect https://github.com/tiangolo/fastapi/issues/2894

[h0rn3t]

import math
from typing import Any, Generic, Optional

from fastapi_async_sqlalchemy import db
from sqlalchemy.future import select
from sqlalchemy.sql import Select, func

from app.schemas.paginator import Page
from app.utils.types import ModelType

class CRUDBase(Generic[ModelType]):
    def __init__(self, model: type[ModelType]):
        self.model = model

    async def get(self, id: Any) -> Optional[ModelType]:
        statement = select(self.model).where(self.model.id == id)
        result = await db.session.execute(statement)
        return result.scalar_one_or_none()

    async def paginate(self, *, statement: Select, page: int = 1, size: int = 25) -> Page[ModelType]:
        total_result = await db.session.execute(select(func.count()).select_from(statement))
        total = total_result.scalar()

        pages = math.ceil(total / size)
        if page > pages:
            data = []
        else:
            search_result = await db.session.execute(statement.limit(size).offset((page - 1) * size))
            data = search_result.scalars().all()

        return Page(
            total=total,
            current_page=page,
            next_page=page + 1 if page < pages else None,
            previous_page=page - 1 if page > 1 else None,
            size=size,
            data=data,
        )

example how look code without dependency injection :)

[h0rn3t]

btw sorry for my word "stupid", its cool that you want to improve your code

[jonra1993]

Hello, @h0rn3t thanks for the code. I think this improvement with fastapi_async_sqlalchemy will be really positive. I am going to improve it as soon as possible 🙂

[jonra1993]

Thanks for this recommendation @h0rn3t. Dependency injection has been removed in this commit https://github.com/jonra1993/fastapi-alembic-sqlmodel-async/commit/76965146baee85642b220e1d348e3e1ba3269f1f