wip

fastapp/fastapp/db/base.py

@@ -0,0 +1,9 @@
+from sqlalchemy.orm import DeclarativeBase
+
+from fastapp.db.meta import meta
+
+
+class Base(DeclarativeBase):
+    """Base for all models."""
+
+    metadata = meta

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/dependencies.py

@@ -0,0 +1,20 @@
+from typing import AsyncGenerator
+
+from sqlalchemy.ext.asyncio import AsyncSession
+from starlette.requests import Request
+
+
+async def get_db_session(request: Request) -> AsyncGenerator[AsyncSession, None]:
+    """
+    Create and get database session.
+
+    :param request: current request.
+    :yield: database session.
+    """
+    session: AsyncSession = request.app.state.db_session_factory()
+
+    try:
+        yield session
+    finally:
+        await session.commit()
+        await session.close()

fastapp/fastapp/db/meta.py

@@ -0,0 +1,3 @@
+import sqlalchemy as sa
+
+meta = sa.MetaData()

fastapp/fastapp/db/models/__init__.py

@@ -0,0 +1,15 @@
+"""fastapp models."""
+
+import pkgutil
+from pathlib import Path
+
+
+def load_all_models() -> None:
+    """Load all models from this folder."""
+    package_dir = Path(__file__).resolve().parent
+    modules = pkgutil.walk_packages(
+        path=[str(package_dir)],
+        prefix="fastapp.db.models.",
+    )
+    for module in modules:
+        __import__(module.name)

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

@@ -0,0 +1,13 @@
+from pathlib import Path
+
+from fastapp.settings import settings
+
+
+async def create_database() -> None:
+    """Create a database."""
+
+
+async def drop_database() -> None:
+    """Drop current database."""
+    if settings.db_file.exists():
+        Path(settings.db_file).unlink()

fastapp/fastapp/settings.py

@@ -49,7 +49,7 @@ def db_url(self) -> URL:
 
         :return: database URL.
         """
-        return URL.build(scheme="sqlite", path=f"///{self.db_file}")
+        return URL.build(scheme="sqlite+aiosqlite", path=f"///{self.db_file}")
 
     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"])

fastapp/fastapp/web/lifespan.py

@@ -2,6 +2,39 @@
 from typing import AsyncGenerator
 
 from fastapi import FastAPI
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from fastapp.db.meta import meta
+from fastapp.db.models import load_all_models
+from fastapp.settings import settings
+
+
+def _setup_db(app: FastAPI) -> None:  # pragma: no cover
+    """
+    Creates connection to the database.
+
+    This function creates SQLAlchemy engine instance,
+    session_factory for creating sessions
+    and stores them in the application's state property.
+
+    :param app: fastAPI application.
+    """
+    engine = create_async_engine(str(settings.db_url), echo=settings.db_echo)
+    session_factory = async_sessionmaker(
+        engine,
+        expire_on_commit=False,
+    )
+    app.state.db_engine = engine
+    app.state.db_session_factory = session_factory
+
+
+async def _create_tables() -> None:  # pragma: no cover
+    """Populates tables in the database."""
+    load_all_models()
+    engine = create_async_engine(str(settings.db_url))
+    async with engine.begin() as connection:
+        await connection.run_sync(meta.create_all)
+    await engine.dispose()
 
 
 @asynccontextmanager
@@ -19,6 +52,9 @@ async def lifespan_setup(
     """
 
     app.middleware_stack = None
+    _setup_db(app)
+    await _create_tables()
     app.middleware_stack = app.build_middleware_stack()
 
     yield
+    await app.state.db_engine.dispose()