✨ feat: async database driver

Author: Zerohertz

.github/workflows/ci.yaml

@@ -209,3 +209,72 @@ jobs:
         env:
           SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_CHECK }}
         if: always()
+
+  test-postgresql:
+    env:
+      DB_TYPE: "postgre"
+      DB_DRIVER: "asyncpg"
+      DB_HOST: "127.0.0.1"
+      DB_PORT: 5432
+      DB_USER: "test"
+      DB_PASSWORD: "test"
+      DB_NAME: "test"
+      DB_TABLE_CREATE: true
+    services:
+      postgresql:
+        image: postgres:17
+        env:
+          PGPORT: ${{ env.DB_PORT }}
+          POSTGRES_USER: ${{ env.DB_USER }}
+          POSTGERS_PASSWORD: ${{ env.DB_PASSWORD }}
+          POSTGRES_DB: ${{ env.DB_NAME }}
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd=pg_isready -U "$POSTGRES_USER" -d "dbname=$POSTGRES_DB" -h 127.0.0.1 -p 5432
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=3
+
+    name: Test PostgreSQL
+    needs: lint
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Cache virtualenv
+        uses: actions/cache@v4
+        with:
+          path: .venv
+          key: ${{ runner.os }}-python-uv-venv-${{ github.sha }}
+          restore-keys: |
+            ${{ runner.os }}-python-uv-venv-
+
+      - name: Run pytest
+        run: |
+          source .venv/bin/activate
+          make test
+
+      - name: Upload test results to Codecov
+        uses: codecov/test-results-action@v1
+        with:
+          flags: postgresql
+          token: ${{ secrets.CODECOV_TOKEN }}
+
+      - name: Upload results to Codecov
+        uses: codecov/codecov-action@v5
+        with:
+          flags: postgresql
+          token: ${{ secrets.CODECOV_TOKEN }}
+
+      - name: Slack webhook
+        uses: 8398a7/action-slack@v3
+        with:
+          status: ${{ job.status }}
+          author_name: fastapi-cookbook
+          fields: repo,message,commit,author,action,eventName,ref,workflow,job,took
+          if_mention: failure,cancelled
+        env:
+          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_CHECK }}
+        if: always()

Makefile

@@ -36,14 +36,14 @@ prod:
 
 .PHONY: k8s
 k8s:
-	kubectl delete -n fastapi -f k8s/dev.yaml 2> /dev/null || echo "Not deployed to the dev environment! 🌎"
-	kubectl apply -n fastapi -f k8s/secrets.yaml
-	kubectl apply -n fastapi -f k8s/dev.yaml
+	kubectl delete -n fastapi -f k8s/postgresql/dev.yaml 2> /dev/null || echo "Not deployed to the dev environment! 🌎"
+	kubectl apply -n fastapi -f k8s/postgresql/secrets.yaml
+	kubectl apply -n fastapi -f k8s/postgresql/dev.yaml
 
 .PHONY: exec
 exec:
 	kubectl exec -it -n fastapi deploy/fastapi-dev -- zsh
 
 .PHONY: expose
 expose:
-	 kubectl expose -n fastapi po/mysql-0 --port 3306 --type=NodePort
+	 kubectl expose -n fastapi po/postgresql-0 --port 5432 --type=NodePort

app/api/v1/endpoints/shields.py

@@ -3,7 +3,7 @@
 from fastapi import APIRouter
 
 from app.schemas.shields import Shields
-from app.services.shields import dday
+from app.utils.shields import dday
 
 router = APIRouter(prefix="/shields", tags=["shields.io"])
 

app/api/v1/endpoints/users.py

@@ -22,7 +22,7 @@ async def create_user(
     user: UserCreateRequest,
     service: UserService = Depends(Provide[Container.user_service]),
 ):
-    return service.create(user)
+    return await service.create(user)
 
 
 @router.get(
@@ -37,7 +37,7 @@ async def get_user(
     id: int,
     service: UserService = Depends(Provide[Container.user_service]),
 ):
-    return service.get_by_id(id)
+    return await service.get_by_id(id)
 
 
 @router.put(
@@ -53,7 +53,7 @@ async def put_user(
     user: UserCreateRequest,
     service: UserService = Depends(Provide[Container.user_service]),
 ):
-    return service.put_by_id(id=id, schema=user)
+    return await service.put_by_id(id=id, schema=user)
 
 
 @router.patch(
@@ -69,7 +69,7 @@ async def patch_user(
     user: UserCreateRequest,
     service: UserService = Depends(Provide[Container.user_service]),
 ):
-    return service.patch_by_id(id=id, schema=user)
+    return await service.patch_by_id(id=id, schema=user)
 
 
 @router.delete(
@@ -84,4 +84,4 @@ async def delete_user(
     id: int,
     service: UserService = Depends(Provide[Container.user_service]),
 ):
-    return service.delete_by_id(id)
+    return await service.delete_by_id(id)

app/core/configs.py

@@ -33,6 +33,8 @@ def DB_SCHEME(self) -> str:
     @property
     def DATABASE_URI(self) -> str:
         if self.DB_TYPE == "sqlite" and self.DB_PORT == 0:
+            if self.DB_DRIVER:
+                return f"{self.DB_TYPE}+{self.DB_DRIVER}:///{self.DB_NAME}"
             return f"{self.DB_TYPE}:///{self.DB_NAME}"
         return str(
             MultiHostUrl.build(

app/core/database.py

@@ -1,7 +1,7 @@
-from contextlib import contextmanager
+from contextlib import asynccontextmanager
 from typing import Any, Generator
 
-from sqlalchemy import create_engine
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 from sqlalchemy.orm import Session, scoped_session, sessionmaker
 
 from app.core.configs import configs
@@ -10,25 +10,22 @@
 
 class Database:
     def __init__(self) -> None:
-        self.engine = create_engine(configs.DATABASE_URI, echo=configs.DB_ECHO)
-        self.scoped_session = scoped_session(
-            sessionmaker(
-                autocommit=False,
-                autoflush=False,
-                bind=self.engine,
-            ),
+        self.engine = create_async_engine(configs.DATABASE_URI, echo=configs.DB_ECHO)
+        self.sessionmaker = sessionmaker(
+            bind=self.engine, class_=AsyncSession, expire_on_commit=False
         )
 
-    def create_all(self) -> None:
-        BaseModel.metadata.create_all(self.engine)
+    async def create_all(self) -> None:
+        async with self.engine.begin() as conn:
+            await conn.run_sync(BaseModel.metadata.create_all)
 
-    @contextmanager
-    def session(self) -> Generator[Any, Any, None]:
-        session: Session = self.scoped_session()
-        try:
-            yield session
-        except Exception:
-            session.rollback()
-            raise
-        finally:
-            session.close()
+    @asynccontextmanager
+    async def session(self) -> Generator[Any, Any, None]:
+        async with self.sessionmaker() as session:
+            try:
+                yield session
+            except Exception:
+                await session.rollback()
+                raise
+            finally:
+                await session.close()

app/core/lifespan.py

@@ -31,6 +31,6 @@ async def lifespan(app: FastAPI):  # pylint: disable=unused-argument
     if configs.DB_TABLE_CREATE:
         logger.warning("Create database")
         database = container.database()
-        database.create_all()
+        await database.create_all()
 
     yield

app/repositories/base.py

@@ -1,6 +1,9 @@
 from contextlib import AbstractContextManager
 from typing import Any, Callable, Generic, Type, TypeVar
 
+from loguru import logger
+from sqlalchemy import update
+from sqlalchemy.future import select
 from sqlalchemy.orm import Session, joinedload
 
 from app.exceptions.database import EntityNotFound
@@ -18,49 +21,59 @@ def __init__(
         self.session = session
         self.model = model
 
-    def create(self, model: T) -> T:
-        with self.session() as session:
+    async def create(self, model: T) -> T:
+        async with self.session() as session:
             session.add(model)
-            session.commit()
-            session.refresh(model)
+            await session.commit()
+            await session.refresh(model)
         return model
 
-    def read_by_id(self, id: int, eager: bool = False) -> T:
-        with self.session() as session:
-            query = session.query(self.model)
+    async def read_by_id(self, id: int, eager: bool = False) -> T:
+        async with self.session() as session:
+            query = select(self.model)
             if eager:
                 for _eager in getattr(self.model, "eagers"):
                     query = query.options(joinedload(getattr(self.model, _eager)))
-            result = query.filter(self.model.id == id).first()
+            query = query.filter(self.model.id == id)
+            result = await session.execute(query)
+            result = result.scalar_one_or_none()
             if not result:
                 raise EntityNotFound
         return result
 
-    def update_by_id(self, id: int, model: dict) -> T:
-        with self.session() as session:
-            session.query(self.model).filter(self.model.id == id).update(model)
-            session.commit()
-            result = session.query(self.model).filter(self.model.id == id).first()
+    async def update_by_id(self, id: int, model: dict) -> T:
+        async with self.session() as session:
+            query = select(self.model).filter(self.model.id == id)
+            result = await session.execute(query)
+            result = result.scalar_one_or_none()
             if not result:
                 raise EntityNotFound
+            logger.warning(result.updated_at)
+            for key, value in model.items():
+                setattr(result, key, value)
+            await session.commit()
+            await session.refresh(result)
+            logger.warning(result.updated_at)
         return result
 
-    def update_attr_by_id(self, id: int, column: str, value: Any) -> T:
-        with self.session() as session:
-            session.query(self.model).filter(self.model.id == id).update(
-                {column: value}
-            )
-            session.commit()
-            result = session.query(self.model).filter(self.model.id == id).first()
+    async def update_attr_by_id(self, id: int, column: str, value: Any) -> T:
+        async with self.session() as session:
+            query = select(self.model).filter(self.model.id == id)
+            result = await session.execute(query)
+            result = result.scalar_one_or_none()
             if not result:
                 raise EntityNotFound
+            setattr(result, column, value)
+            await session.commit()
         return result
 
-    def delete_by_id(self, id: int) -> T:
-        with self.session() as session:
-            query = session.query(self.model).filter(self.model.id == id).first()
-            if not query:
+    async def delete_by_id(self, id: int) -> T:
+        async with self.session() as session:
+            query = select(self.model).filter(self.model.id == id)
+            result = await session.execute(query)
+            result = result.scalar_one_or_none()
+            if not result:
                 raise EntityNotFound
-            session.delete(query)
-            session.commit()
-        return query
+            await session.delete(result)
+            await session.commit()
+        return result

app/services/base.py

@@ -22,29 +22,33 @@ def mapper(self, data: BaseSchemaRequest | T) -> T | BaseSchemaResponse:
             return self.repository.model(**data.model_dump())
         return BaseSchemaResponse.model_validate(data)
 
-    def create(self, schema: BaseSchemaRequest) -> BaseSchemaResponse:
+    async def create(self, schema: BaseSchemaRequest) -> BaseSchemaResponse:
         model = self.mapper(schema)
-        model = self.repository.create(model=model)
+        model = await self.repository.create(model=model)
         return self.mapper(model)
 
-    def get_by_id(self, id: int) -> BaseSchemaResponse:
-        model = self.repository.read_by_id(id=id)
+    async def get_by_id(self, id: int) -> BaseSchemaResponse:
+        model = await self.repository.read_by_id(id=id)
         return self.mapper(model)
 
-    def put_by_id(self, id: int, schema: BaseSchemaRequest) -> BaseSchemaResponse:
-        model = self.repository.update_by_id(id=id, model=schema.model_dump())
+    async def put_by_id(self, id: int, schema: BaseSchemaRequest) -> BaseSchemaResponse:
+        model = await self.repository.update_by_id(id=id, model=schema.model_dump())
         return self.mapper(model)
 
-    def patch_by_id(self, id: int, schema: BaseSchemaRequest) -> BaseSchemaResponse:
-        model = self.repository.update_by_id(
+    async def patch_by_id(
+        self, id: int, schema: BaseSchemaRequest
+    ) -> BaseSchemaResponse:
+        model = await self.repository.update_by_id(
             id=id, model=schema.model_dump(exclude_none=True)
         )
         return self.mapper(model)
 
-    def patch_attr_by_id(self, id: int, attr: str, value: Any) -> BaseSchemaResponse:
-        model = self.repository.update_attr_by_id(id=id, column=attr, value=value)
+    async def patch_attr_by_id(
+        self, id: int, attr: str, value: Any
+    ) -> BaseSchemaResponse:
+        model = await self.repository.update_attr_by_id(id=id, column=attr, value=value)
         return self.mapper(model)
 
-    def delete_by_id(self, id: int) -> BaseSchemaResponse:
-        model = self.repository.delete_by_id(id=id)
+    async def delete_by_id(self, id: int) -> BaseSchemaResponse:
+        model = await self.repository.delete_by_id(id=id)
         return self.mapper(model)