Ch7

fastapp/Dockerfile

@@ -1,4 +1,7 @@
 FROM python:3.11.4-slim-bullseye as prod
+RUN apt-get update && apt-get install -y \
+  gcc \
+  && rm -rf /var/lib/apt/lists/*
 
 
 RUN pip install poetry==1.8.2
@@ -13,6 +16,10 @@ WORKDIR /app/src
 
 # Installing requirements
 RUN --mount=type=cache,target=/tmp/poetry_cache poetry install --only main
+# Removing gcc
+RUN apt-get purge -y \
+  gcc \
+  && rm -rf /var/lib/apt/lists/*
 
 # Copying actuall application
 COPY . /app/src/

fastapp/README.md

@@ -116,6 +116,12 @@ docker-compose down
 ```
 
 For running tests on your local machine.
+1. you need to start a database.
+
+I prefer doing it with docker:
+```
+docker run -p "5432:5432" -e "POSTGRES_PASSWORD=fastapp" -e "POSTGRES_USER=fastapp" -e "POSTGRES_DB=fastapp" postgres:16.3-bullseye
+```
 
 
 2. Run the pytest.

fastapp/docker-compose.yml

@@ -7,11 +7,32 @@ services:
     restart: always
     env_file:
       - .env
+    depends_on:
+      db:
+        condition: service_healthy
     environment:
       FASTAPP_HOST: 0.0.0.0
-      FASTAPP_DB_FILE: /db_data/db.sqlite3
+      FASTAPP_DB_HOST: fastapp-db
+      FASTAPP_DB_PORT: 5432
+      FASTAPP_DB_USER: fastapp
+      FASTAPP_DB_PASS: fastapp
+      FASTAPP_DB_BASE: fastapp
+
+  db:
+    image: postgres:16.3-bullseye
+    hostname: fastapp-db
+    environment:
+      POSTGRES_PASSWORD: "fastapp"
+      POSTGRES_USER: "fastapp"
+      POSTGRES_DB: "fastapp"
     volumes:
-      - fastapp-db-data:/db_data/
+      - fastapp-db-data:/var/lib/postgresql/data
+    restart: always
+    healthcheck:
+      test: pg_isready -U fastapp
+      interval: 2s
+      timeout: 3s
+      retries: 40
 
 
 

fastapp/fastapp/db/dao/__init__.py

@@ -0,0 +1 @@
+"""DAO classes."""

fastapp/fastapp/db/dao/dummy_dao.py

@@ -0,0 +1,50 @@
+from typing import List, Optional
+
+from fastapi import Depends
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from fastapp.db.dependencies import get_db_session
+from fastapp.db.models.dummy_model import DummyModel
+
+
+class DummyDAO:
+    """Class for accessing dummy table."""
+
+    def __init__(self, session: AsyncSession = Depends(get_db_session)) -> None:
+        self.session = session
+
+    async def create_dummy_model(self, name: str) -> None:
+        """
+        Add single dummy to session.
+
+        :param name: name of a dummy.
+        """
+        self.session.add(DummyModel(name=name))
+
+    async def get_all_dummies(self, limit: int, offset: int) -> List[DummyModel]:
+        """
+        Get all dummy models with limit/offset pagination.
+
+        :param limit: limit of dummies.
+        :param offset: offset of dummies.
+        :return: stream of dummies.
+        """
+        raw_dummies = await self.session.execute(
+            select(DummyModel).limit(limit).offset(offset),
+        )
+
+        return list(raw_dummies.scalars().fetchall())
+
+    async def filter(self, name: Optional[str] = None) -> List[DummyModel]:
+        """
+        Get specific dummy model.
+
+        :param name: name of dummy instance.
+        :return: dummy models.
+        """
+        query = select(DummyModel)
+        if name:
+            query = query.where(DummyModel.name == name)
+        rows = await self.session.execute(query)
+        return list(rows.scalars().fetchall())

fastapp/fastapp/db/models/dummy_model.py

@@ -0,0 +1,13 @@
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.sql.sqltypes import String
+
+from fastapp.db.base import Base
+
+
+class DummyModel(Base):
+    """Model for demo purpose."""
+
+    __tablename__ = "dummy_model"
+
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    name: Mapped[str] = mapped_column(String(length=200))

fastapp/fastapp/db/utils.py

@@ -1,13 +1,44 @@
-from pathlib import Path
+from sqlalchemy import text
+from sqlalchemy.engine import make_url
+from sqlalchemy.ext.asyncio import create_async_engine
 
 from fastapp.settings import settings
 
 
 async def create_database() -> None:
     """Create a database."""
+    db_url = make_url(str(settings.db_url.with_path("/postgres")))
+    engine = create_async_engine(db_url, isolation_level="AUTOCOMMIT")
+
+    async with engine.connect() as conn:
+        database_existance = await conn.execute(
+            text(
+                f"SELECT 1 FROM pg_database WHERE datname='{settings.db_base}'",  # noqa: S608
+            ),
+        )
+        database_exists = database_existance.scalar() == 1
+
+    if database_exists:
+        await drop_database()
+
+    async with engine.connect() as conn:
+        await conn.execute(
+            text(
+                f'CREATE DATABASE "{settings.db_base}" ENCODING "utf8" TEMPLATE template1',  # noqa: E501
+            ),
+        )
 
 
 async def drop_database() -> None:
     """Drop current database."""
-    if settings.db_file.exists():
-        Path(settings.db_file).unlink()
+    db_url = make_url(str(settings.db_url.with_path("/postgres")))
+    engine = create_async_engine(db_url, isolation_level="AUTOCOMMIT")
+    async with engine.connect() as conn:
+        disc_users = (
+            "SELECT pg_terminate_backend(pg_stat_activity.pid) "  # noqa: S608
+            "FROM pg_stat_activity "
+            f"WHERE pg_stat_activity.datname = '{settings.db_base}' "
+            "AND pid <> pg_backend_pid();"
+        )
+        await conn.execute(text(disc_users))
+        await conn.execute(text(f'DROP DATABASE "{settings.db_base}"'))

fastapp/fastapp/settings.py

@@ -39,7 +39,11 @@ class Settings(BaseSettings):
 
     log_level: LogLevel = LogLevel.INFO
     # Variables for the database
-    db_file: Path = TEMP_DIR / "db.sqlite3"
+    db_host: str = "localhost"
+    db_port: int = 5432
+    db_user: str = "fastapp"
+    db_pass: str = "fastapp"
+    db_base: str = "admin"
     db_echo: bool = False
 
     @property
@@ -49,7 +53,14 @@ def db_url(self) -> URL:
 
         :return: database URL.
         """
-        return URL.build(scheme="sqlite+aiosqlite", path=f"///{self.db_file}")
+        return URL.build(
+            scheme="postgresql+asyncpg",
+            host=self.db_host,
+            port=self.db_port,
+            user=self.db_user,
+            password=self.db_pass,
+            path=f"/{self.db_base}",
+        )
 
     model_config = SettingsConfigDict(
         env_file=".env",

fastapp/fastapp/web/api/dummy/__init__.py

@@ -0,0 +1,5 @@
+"""Dummy model API."""
+
+from fastapp.web.api.dummy.views import router
+
+__all__ = ["router"]

fastapp/fastapp/web/api/dummy/schema.py

@@ -0,0 +1,20 @@
+from pydantic import BaseModel, ConfigDict
+
+
+class DummyModelDTO(BaseModel):
+    """
+    DTO for dummy models.
+
+    It returned when accessing dummy models from the API.
+    """
+
+    id: int
+    name: str
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class DummyModelInputDTO(BaseModel):
+    """DTO for creating new dummy model."""
+
+    name: str

fastapp/fastapp/web/api/dummy/views.py

@@ -0,0 +1,41 @@
+from typing import List
+
+from fastapi import APIRouter
+from fastapi.param_functions import Depends
+
+from fastapp.db.dao.dummy_dao import DummyDAO
+from fastapp.db.models.dummy_model import DummyModel
+from fastapp.web.api.dummy.schema import DummyModelDTO, DummyModelInputDTO
+
+router = APIRouter()
+
+
+@router.get("/", response_model=List[DummyModelDTO])
+async def get_dummy_models(
+    limit: int = 10,
+    offset: int = 0,
+    dummy_dao: DummyDAO = Depends(),
+) -> List[DummyModel]:
+    """
+    Retrieve all dummy objects from the database.
+
+    :param limit: limit of dummy objects, defaults to 10.
+    :param offset: offset of dummy objects, defaults to 0.
+    :param dummy_dao: DAO for dummy models.
+    :return: list of dummy objects from database.
+    """
+    return await dummy_dao.get_all_dummies(limit=limit, offset=offset)
+
+
+@router.put("/")
+async def create_dummy_model(
+    new_dummy_object: DummyModelInputDTO,
+    dummy_dao: DummyDAO = Depends(),
+) -> None:
+    """
+    Creates dummy model in the database.
+
+    :param new_dummy_object: new dummy model item.
+    :param dummy_dao: DAO for dummy models.
+    """
+    await dummy_dao.create_dummy_model(name=new_dummy_object.name)

fastapp/fastapp/web/api/router.py

@@ -1,7 +1,8 @@
 from fastapi.routing import APIRouter
 
-from fastapp.web.api import echo, monitoring
+from fastapp.web.api import dummy, echo, monitoring
 
 api_router = APIRouter()
 api_router.include_router(monitoring.router)
 api_router.include_router(echo.router, prefix="/echo", tags=["echo"])
+api_router.include_router(dummy.router, prefix="/dummy", tags=["dummy"])