feat: add minimal database setup with SQLite and CRUD operations for heroes

Author: umbertogriffo

.env.example

@@ -3,9 +3,14 @@ SERVICE_NAME="FastAPI Service"
 HOST=0.0.0.0
 PORT=8080
 
+# Database Configuration
+DATABASE_URL=sqlite:///database.db
+# DATABASE_URL=postgresql://develop:develop_secret@localhost:5432/develop
+
 # Logging Configuration
 LOG_LEVEL=INFO
 LOG_SAVE_ON_FILE=false
+LOG_DATABASE_QUERIES=false
 
 # Authentication and Authorization
 API_KEY=your-secret-api-key-here

pyproject.toml

@@ -5,9 +5,10 @@ description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.10,<3.11"
 dependencies = [
-    "fastapi[standard]>=0.112.1",
-    "uvicorn==0.30.5",
-    "pydantic-settings==2.3.4"
+    "fastapi[standard]>=0.124.4",
+    "uvicorn==0.38.0",
+    "pydantic-settings==2.12.0",
+    "sqlmodel~=0.0.27",
 ]
 
 [dependency-groups]

src/api/deps.py

@@ -1,14 +1,29 @@
+from typing import Annotated, Generator
+
 from config import settings
+from database import engine
 from fastapi import Depends, HTTPException
 from fastapi.security import APIKeyHeader
+from sqlmodel import Session
 from starlette import status
 
 """
 Defines dependencies used by the endpoints.
 """
 
 
-def api_key_auth(api_key: str = Depends(APIKeyHeader(name="Authorization"))):
+def get_db_session() -> Generator[Session, None, None]:
+    """
+    Create a new database session and close the session after the operation has ended.
+    """
+    with Session(engine) as session:
+        yield session
+
+
+SessionDep = Annotated[Session, Depends(get_db_session)]
+
+
+def api_key_auth(api_key: str = Depends(APIKeyHeader(name="Authorization"))) -> None:
     # Validate the provided API key
     if api_key != settings.API_KEY.get_secret_value():
         raise HTTPException(

src/api/endpoints/heroes.py

@@ -0,0 +1,142 @@
+from typing import Annotated
+
+from api.deps import SessionDep, get_db_session
+from excepts import DatabaseEntryNotFound, get_error_content
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from models import Hero, HeroCreate, HeroPublic, HeroUpdate
+from sqlmodel import select
+from utils.log import get_logger
+
+logger = get_logger()
+router = APIRouter()
+
+
+@router.post(
+    path="/",
+    summary="Create a new hero",
+    status_code=status.HTTP_200_OK,
+    response_description="Returns the created hero",
+    response_model=HeroPublic,
+    dependencies=[Depends(get_db_session)],
+)
+def create_hero(hero: HeroCreate, session: SessionDep):
+    db_hero = Hero.model_validate(hero)
+    session.add(db_hero)
+    session.commit()
+    session.refresh(db_hero)
+    return db_hero
+
+
+@router.get(
+    path="/",
+    summary="Retrieve a list of heroes",
+    status_code=status.HTTP_200_OK,
+    response_description="Returns a list of heroes",
+    response_model=list[HeroPublic],
+    dependencies=[Depends(get_db_session)],
+)
+def read_heroes(
+    session: SessionDep,
+    offset: int = 0,
+    limit: Annotated[int, Query(le=100)] = 100,
+) -> list[Hero]:
+    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
+    return list(heroes)
+
+
+@router.get(
+    path="/{hero_id}",
+    summary="Retrieve a hero by ID",
+    status_code=status.HTTP_200_OK,
+    response_description="Returns the hero with the specified ID",
+    response_model=HeroPublic,
+    dependencies=[Depends(get_db_session)],
+)
+def read_hero(hero_id: int, session: SessionDep):
+    try:
+        hero = session.get(Hero, hero_id)
+        if not hero:
+            raise DatabaseEntryNotFound(f"Hero with ID {hero_id} not found")
+        return hero
+
+    except Exception as e:
+        error = get_error_content(e)
+        error_message = error.message
+
+        logger.error(
+            error_message,
+            exc_info=True,
+            stack_info=True,
+        )
+
+        raise HTTPException(
+            status_code=error.http_status_code,
+            detail=error_message,
+        )
+
+
+@router.patch(
+    path="/{hero_id}",
+    summary="Update a hero by ID",
+    status_code=status.HTTP_200_OK,
+    response_description="Returns the updated hero",
+    response_model=HeroPublic,
+    dependencies=[Depends(get_db_session)],
+)
+def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):
+    try:
+        hero_db = session.get(Hero, hero_id)
+        if not hero_db:
+            raise DatabaseEntryNotFound(f"Hero with ID {hero_id} not found")
+        hero_data = hero.model_dump(exclude_unset=True)
+        hero_db.sqlmodel_update(hero_data)
+        session.add(hero_db)
+        session.commit()
+        session.refresh(hero_db)
+        return hero_db
+
+    except Exception as e:
+        error = get_error_content(e)
+        error_message = error.message
+
+        logger.error(
+            error_message,
+            exc_info=True,
+            stack_info=True,
+        )
+
+        raise HTTPException(
+            status_code=error.http_status_code,
+            detail=error_message,
+        )
+
+
+@router.delete(
+    path="/{hero_id}",
+    summary="Delete a hero by ID",
+    status_code=status.HTTP_200_OK,
+    response_description="Indicates whether the hero was successfully deleted",
+    dependencies=[Depends(get_db_session)],
+)
+def delete_hero(hero_id: int, session: SessionDep):
+    try:
+        hero = session.get(Hero, hero_id)
+        if not hero:
+            raise DatabaseEntryNotFound("Hero not found")
+        session.delete(hero)
+        session.commit()
+        return {"ok": True}
+    except Exception as e:
+        error = get_error_content(e)
+        error_message = error.message
+
+        logger.error(
+            error_message,
+            exc_info=True,
+            stack_info=True,
+        )
+
+        raise HTTPException(
+            status_code=error.http_status_code,
+            detail=error_message,
+        )

src/api/router.py

@@ -1,9 +1,7 @@
-from api.endpoints import (
-    health,
-    users,
-)
+from api.endpoints import health, heroes, users
 from fastapi import APIRouter
 
 router = APIRouter()
 router.include_router(health.router, prefix="", tags=["health"])
 router.include_router(users.router, prefix="/users", tags=["users"])
+router.include_router(heroes.router, prefix="/heroes", tags=["heroes"])

src/config.py

@@ -29,9 +29,13 @@ class Settings(BaseSettings):
         f"http://localhost:{PORT}",
     ]
 
+    # Database Configuration
+    DATABASE_URL: str = "sqlite:///database.db"
+
     # Logging Configuration
     LOG_LEVEL: str = "INFO"
     LOG_SAVE_ON_FILE: bool = False
+    LOG_DATABASE_QUERIES: bool = False
 
     # Authentication and Authorization
     API_KEY: SecretStr

src/database.py

@@ -0,0 +1,40 @@
+from config import settings
+from sqlalchemy import Engine
+from sqlmodel import SQLModel, create_engine
+
+
+def create_db_engine(verbose: bool = False, **kwargs):
+    """
+    Create the SQLAlchemy engine for database interactions.
+
+    Args:
+        verbose (bool): If True, enables SQL query logging.
+        **kwargs: Additional keyword arguments for the engine.
+    Returns:
+        engine: The SQLAlchemy engine instance.
+    """
+    # Using check_same_thread=False allows FastAPI to use the same SQLite database in different threads.
+    connect_args = {
+        "check_same_thread": False,
+    }
+    engine = create_engine(
+        settings.DATABASE_URL,
+        connect_args=connect_args,
+        **{
+            "echo": verbose,
+            "pool_use_lifo": True,  # Avoid many idle connections
+            "pool_pre_ping": True,  # Gracefully handle connections closed by the server
+            **kwargs,
+        },
+    )
+    return engine
+
+
+def create_db_and_tables(engine: Engine):
+    """
+    Create database tables based on the defined SQLModel models.
+    """
+    SQLModel.metadata.create_all(engine)
+
+
+engine = create_db_engine()

src/excepts.py

@@ -42,6 +42,18 @@ class InvalidValue(BackendException):
     http_status_code: int = status.HTTP_422_UNPROCESSABLE_ENTITY
 
 
+class DatabaseException(BackendException): ...
+
+
+class DatabaseEntryNotFound(DatabaseException):
+    """
+    Raised when a database entry is not found.
+    """
+
+    default_message = "The requested database entry was not found"
+    http_status_code: int = status.HTTP_404_NOT_FOUND
+
+
 # Dictionary of errors that we want to propagate and expose to the API users
 ERRORS = {
     ValueRequired: ErrorContent(
@@ -50,6 +62,10 @@ class InvalidValue(BackendException):
     InvalidValue: ErrorContent(
         InvalidValue.default_message, InvalidValue.http_status_code
     ),
+    DatabaseEntryNotFound: ErrorContent(
+        DatabaseEntryNotFound.default_message,
+        DatabaseEntryNotFound.http_status_code,
+    ),
 }
 
 

src/main.py

@@ -5,6 +5,7 @@
 from api.exception_handlers import validation_exception_handler
 from api.router import router
 from config import init_whatever, settings
+from database import create_db_and_tables, engine
 from fastapi import FastAPI
 from fastapi.exceptions import RequestValidationError
 from fastapi.middleware.cors import CORSMiddleware
@@ -19,7 +20,14 @@ async def lifespan(app: FastAPI):
     # It may be stored in app state object:
     # app.state.some_resource = SomeResource()
     # https://github.com/fastapi/fastapi/discussions/13029
+
+    # TODO: For production you would probably use an Alembic migration script that runs before you start your app.
+    # This is just for demonstration purposes.
+    # https://alembic.sqlalchemy.org/en/latest/
+    create_db_and_tables(engine)
     yield
+    # Clean up
+    engine.dispose()
 
 
 app = FastAPI(title=settings.SERVICE_NAME, version="0.1.0", lifespan=lifespan)

src/models.py

@@ -0,0 +1,47 @@
+from sqlmodel import Field, SQLModel
+
+
+class HeroBase(SQLModel):
+    """Base model for Hero with common attributes."""
+
+    name: str = Field(title="Name", description="The public name of the hero.")
+    age: int | None = Field(
+        default=None, title="Age", description="The age of the hero."
+    )
+
+
+class Hero(HeroBase, table=True):
+    """Database model for Hero with the extra fields that are not always in the other models."""
+
+    id: int | None = Field(
+        default=None,
+        title="Hero ID",
+        description="The unique identifier for the hero.",
+        primary_key=True,
+    )
+    secret_name: str = Field(
+        title="Secret Name", description="The secret identity of the hero."
+    )
+
+
+class HeroPublic(HeroBase):
+    """Public model for Hero without sensitive information."""
+
+    id: int
+
+
+class HeroCreate(HeroBase):
+    """Model for creating a new Hero."""
+
+    secret_name: str
+
+
+class HeroUpdate(HeroBase):
+    """Model for updating an existing Hero.
+    We don't really need to inherit from HeroBase because we are re-declaring all the fields.
+    I'll leave it inheriting just for consistency, but this is not necessary.
+    """
+
+    name: str | None = None
+    age: int | None = None
+    secret_name: str | None = None