FastAPIとSQLModelでつくる簡単なToDoアプリ

fastapi==0.95.1
sqlmodel==0.0.8
alembic==1.10.4

.
├── alembic.ini
├── cruds/
│   └── task.py
├── models/
│   └── task.py
├── routers/
│   └── task.py
├── migrations/
│   ├── env.py
│   └── script.py.mako
├── db.py
├── main.py
└── settings.py

from typing import Optional

from sqlmodel import SQLModel, Field


class TaskBase(SQLModel):
    title: str = Field(description='task name', nullable=False)
    done: bool = Field(default=False, description='done flag', nullable=False)


class Task(TaskBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class TaskCreate(TaskBase):
    pass


class TaskRead(TaskBase):
    id: int


class TaskUpdate(SQLModel):
    title: str
    done: bool

SQLITE_NAME = 'db.sqlite3'
DB_URI = f'sqlite:///{SQLITE_NAME}'

from collections.abc import Generator

from sqlmodel import create_engine, Session

import settings


engine = create_engine(settings.DB_URI, echo=True)

def get_session() -> Generator[Session, None, None]:
    with Session(engine) as session:
        yield session

from sqlmodel import select, Session

from models.task import Task, TaskCreate, TaskUpdate


def read_task(session: Session, task_id: int) -> Task | None:
    task: Task | None = session.get(Task, task_id)
    return task


def read_tasks(session: Session, done: bool | None = None) -> list[Task]:
    query = select(Task)
    if done is not None:
        query = query.where(Task.done == done)
    tasks: list[Task] = session.exec(query).all()
    return tasks


def create_task(session: Session, task: TaskCreate) -> Task:
    db_task: Task = Task.from_orm(task)
    session.add(db_task)
    session.commit()
    session.refresh(db_task)
    return db_task


def update_task(session: Session, db_task: Task, task: TaskUpdate) -> Task:
    task_data = task.dict(exclude_unset=True)
    for key, value in task_data.items():
        setattr(db_task, key, value)
    session.add(db_task)
    session.commit()
    session.refresh(db_task)
    return db_task


def delete_task(session: Session, db_task: Task) -> None:
    session.delete(db_task)
    session.commit()
    return

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlmodel import Session

import cruds.task as task_crud
from models.task import Task, TaskRead, TaskCreate, TaskUpdate
from db import get_session


router = APIRouter(
    prefix='/tasks',
    tags=['Task'],
    responses={404: {'message': 'Not found'}})


@router.get('/', response_model=list[TaskRead])
def read_tasks(
        *, 
        session: Session = Depends(get_session), 
        done: bool | None = Query(None)
    ) -> list[TaskRead]:
    """
    Read tasks from the database, filtered by the 'done' flag.

    Parameters
    ----------
    done : bool | None, optional, **[query parameter]**\n
        The value of the 'done' flag to filter tasks by, by default None.

    Returns
    -------
    list[TaskRead]\n
        The list of tasks.
    """
    return task_crud.read_tasks(session=session, done=done)


@router.get('/{task_id}', response_model=TaskRead)
def read_task(
        *, 
        session: Session = Depends(get_session), 
        task_id: int
    ) -> TaskRead:
    """
    Read a task from the database by its ID.

    Parameters
    ----------
    task_id : int, **[path parameter]**\n
        The ID of the task to read.

    Returns
    -------
    TaskRead\n
        The requested task if found, otherwise raises a 404 HTTPException.
    """

    task: Task | None = task_crud.read_task(session=session, task_id=task_id)
    if task is None:
        raise HTTPException(status_code=404, detail='Task not found')
    return task


@router.post('/', response_model=TaskCreate)
def create_task(
        *, 
        session: Session = Depends(get_session), 
        task: TaskCreate
    ) -> TaskCreate:
    """
    Create a new task in the database.

    Parameters
    ----------
    task : TaskCreate, **[body parameter]**\n
        The task creation object.

    Returns
    -------
    TaskCreate\n
        The created task.
    """
    created_task = task_crud.create_task(session=session, task=task)
    return created_task


@router.patch('/{task_id}', response_model=TaskRead)
def update_task(
        *, 
        session: Session = Depends(get_session), 
        task_id: int, 
        task: TaskUpdate
    ) -> TaskRead:
    """
    Update a task in the database.

    Parameters
    ----------
    task_id : int, **[path parameter]**\n
        The ID of the task to update.
    task : TaskUpdate, **[body parameter]**\n
        The update object containing the new task data.

    Returns
    -------
    TaskRead\n
        The updated task.
    """
    org_task: Task | None = task_crud.read_task(session=session, task_id=task_id)
    if org_task is None:
        raise HTTPException(status_code=404, detail='Task not found')
    return task_crud.update_task(session=session, db_task=org_task, task=task)


@router.delete('/{task_id}')
def delete_task(
        *, 
        session: Session = Depends(get_session), 
        task_id: int
    ) -> dict[str, bool]:
    """
    Delete a task from the database.

    Parameters
    ----------
    task_id : int, **[path parameter]**\n
        The ID of the task to delete.

    Returns
    -------
    dict[str, bool]\n
        A dictionary with a single key 'ok' and a value of True, indicating successful deletion.
    """
    task_to_delete: Task | None = task_crud.read_task(session=session, task_id=task_id)
    if task_to_delete is None:
        raise HTTPException(status_code=404, detail='Task not found')
    task_crud.delete_task(session=session, db_task=task_to_delete)
    return {'ok': True}

from fastapi import FastAPI

from routers import task


app = FastAPI()
app.include_router(task.router)

$ alembic init migrations

from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
from sqlmodel import SQLModel

from settings import DB_URI
from models.task import Task

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = SQLModel.metadata

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """
    conf = config.get_section(config.config_ini_section, {})
    conf['sqlalchemy.url'] = DB_URI
    connectable = engine_from_config(
        conf,
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection, target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

"""${message}

Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}

"""
from alembic import op
import sqlalchemy as sa
import sqlmodel
${imports if imports else ""}

# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}


def upgrade() -> None:
    ${upgrades if upgrades else "pass"}


def downgrade() -> None:
    ${downgrades if downgrades else "pass"}

$ alembic revision --autogenerate -m "init"
$ alembic upgrade head

$ uvicorn main:app --reload