Agentic AI with Java: Live Cohort
Claude CodeFull-Stack Integration with Claude Code

Implementing Backend Features - Part 1

Build the Task model, Pydantic v2 schemas and async database engine from the specs, configure Alembic for async SQLAlchemy, and run the first migration.

The skeleton is in place. Now we fill in the data layer: the database model, the request and response schemas, the async database engine and the first migration. All of it is driven by the two spec files.

Branch and session

!git checkout -b feature/task-model
/clear

In the video the /clear step was skipped here, and for two small features it was fine. As a habit, still start every feature with a clean context. It costs nothing and avoids surprises.

Step 1: Model, schemas and database engine

Prompt: task model
Read @specs/api-contract.md and @specs/seed-data.json.

Create the SQLAlchemy 2.0 async Task model in backend/app/models/task.py.
Field names must exactly match the snake_case keys in seed-data.json.
State and assigned_role must be Python Enums.

Create corresponding Pydantic v2 schemas in backend/app/schemas/task.py:
  - TaskCreate (request body for POST)
  - TaskUpdate (request body for PUT, all fields optional)
  - TaskResponse (response body for all endpoints)

Set up the async database engine using SQLite with aiosqlite.
Include a get_db dependency for FastAPI.

Every line of this prompt removes a guess. "Field names must exactly match the snake_case keys" stops Claude from inventing its own names. "State and assigned_role must be Python Enums" means invalid values are rejected automatically.

What Claude built

The model uses SQLAlchemy 2.0's typed Mapped[...] style, with enums that mirror the frontend's STATES and ROLE_CONFIG:

backend/app/models/task.py (trimmed)
class AssignedRole(str, enum.Enum):
    Admin = "Admin"
    Content = "Content"
    Editor = "Editor"
    Uploader = "Uploader"


class TaskState(str, enum.Enum):
    code_ready = "code_ready"
    recorded = "recorded"
    editing = "editing"
    uploaded = "uploaded"
    published = "published"


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    description: Mapped[str] = mapped_column(Text, nullable=False, default="")
    assigned_role: Mapped[AssignedRole] = mapped_column(
        SAEnum(AssignedRole, native_enum=False, length=32), nullable=False
    )
    state: Mapped[TaskState] = mapped_column(
        SAEnum(TaskState, native_enum=False, length=32),
        nullable=False,
        default=TaskState.code_ready,
    )
    created_at: Mapped[date] = mapped_column(Date, nullable=False)

The schemas solve the camelCase problem the API contract warned us about. The backend uses snake_case internally, but the frontend reads assignedRole and createdAt. Pydantic's alias generator converts between the two:

backend/app/schemas/task.py (trimmed)
from pydantic import BaseModel, ConfigDict, Field
from pydantic.alias_generators import to_camel


class _CamelBase(BaseModel):
    # snake_case in Python, camelCase on the wire
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)


class TaskCreate(_CamelBase):
    title: str = Field(..., min_length=1)
    description: str = Field(default="")
    assigned_role: AssignedRole
    state: TaskState = Field(default=TaskState.code_ready)


class TaskUpdate(_CamelBase):
    # every field optional, so the frontend can send only {"state": "editing"}
    title: str | None = Field(default=None, min_length=1)
    description: str | None = None
    assigned_role: AssignedRole | None = None
    state: TaskState | None = None


class TaskResponse(_CamelBase):
    model_config = ConfigDict(
        alias_generator=to_camel, populate_by_name=True, from_attributes=True
    )
    id: int
    title: str
    description: str
    assigned_role: AssignedRole
    state: TaskState
    created_at: date

The database module creates an async engine and a get_db dependency that FastAPI injects into each request:

backend/app/core/database.py (trimmed)
engine = create_async_engine(settings.database_url, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        yield session

Claude then ran a quick check that all three schemas accept valid data and reject malformed input, and committed.

Step 2: Alembic for async SQLAlchemy

Alembic's generated env.py is written for synchronous SQLAlchemy. Ours is async, so we ask Claude to adapt it and create the first migration:

Prompt: Alembic
Configure alembic/env.py to work with SQLAlchemy 2.0 async and our SQLite database.
Then generate the first migration for the Task model and apply it.
Use uv to run all alembic commands.

Claude rewrites env.py to reuse the app's async engine, and adds render_as_batch=True, which SQLite needs because it does not support most ALTER TABLE operations. Then it runs:

uv run alembic revision --autogenerate -m "create tasks table"
uv run alembic upgrade head
CommandWhat it does
alembic revision --autogenerateCompares your models with the database and writes a migration script
alembic upgrade headApplies all pending migrations, creating the tasks table
Alembic migration flow from model change to database

Why migrations matter

Six months from now you will need a new column. Do not add it by hand in the database. Change the model, generate a migration, apply it. That is why our CLAUDE.md says "Always create a migration when changing models". Every schema change is recorded, reviewable and repeatable on every environment.

SQLAlchemy 2.0 style at a glance

If you have used older SQLAlchemy tutorials, the code may look different. Our CLAUDE.md asks for the 2.0 style:

TaskOld 1.x style2.0 style (what we use)
Define a columntitle = Column(String(255))title: Mapped[str] = mapped_column(String(255))
Get all rowssession.query(Task).all()(await db.execute(select(Task))).scalars().all()
Get by idsession.query(Task).get(1)await db.get(Task, 1)
Base classBase = declarative_base()class Base(DeclarativeBase): pass

The typed Mapped[...] annotations also give your editor and type checker real information about each column.

Why async?

FastAPI is built for async code. While one request waits for the database, the server can handle other requests. With SQLite the difference is small, but using AsyncSession and aiosqlite from the start means switching to PostgreSQL later (with asyncpg) needs almost no code changes.

Alembic commands worth knowing

CommandWhat it does
uv run alembic currentShow which migration the database is at
uv run alembic historyList all migrations in order
uv run alembic revision --autogenerate -m "msg"Create a migration from model changes
uv run alembic upgrade headApply all pending migrations
uv run alembic downgrade -1Undo the most recent migration

Always read autogenerated migrations

Autogenerate is smart but not perfect. It can miss some changes (for example renamed columns look like a drop plus an add, which loses data). Open the file in alembic/versions/ and read upgrade() before you apply it.

Review before you merge

Run /diff and actually read it. In our case, a careful review finds a few things worth knowing:

FindingWhy it matters
config.py uses an inner class Config:That is Pydantic v1 style. With Pydantic v2 and pydantic-settings, prefer model_config = SettingsConfigDict(env_file=".env"). It still works, but our CLAUDE.md says Pydantic v2
echo=True on the engineLogs every SQL statement. Great for learning, too noisy for production
An extra app/routes/tasks.py with only commentsA leftover from the skeleton next to the real app/api/v1/tasks.py. Remove it so nobody edits the wrong file
backend/main.py prints "Hello from backend!"The uv init stub. The real app lives in app/main.py

None of these break anything today. But this is exactly the difference between done and verified. Ask Claude to fix them, or note them for later.

Commit, merge, delete

Commit the Alembic configuration and the first migration.
!git checkout main
!git merge feature/task-model
!git branch -d feature/task-model

Official references


Written By: Shramik Masti

How is this guide?

Last updated on