Create a New Table + CRUD

Add a new database table with full Create/Read operations. Learn the end-to-end flow.

The Schema

What You'll Build

• SQLAlchemy Model: ScanNote in db/models/
• Pydantic Schemas: Request/Response models
• Service Layer: Business logic for create/list
• API Routes: POST and GET endpoints
• Alembic Migration: Database schema change

Implementation Guide

Step 1: Create the SQLAlchemy Model

Create platform/db/models/scan_note.py:

from datetime import datetime
from uuid import UUID, uuid4

from sqlalchemy import ForeignKey, Text, text
from sqlalchemy.orm import Mapped, mapped_column, relationship

from db.base import Base


class ScanNote(Base):
    __tablename__ = "scan_note"

    id: Mapped[UUID] = mapped_column(
        primary_key=True,
        default=uuid4,
        server_default=text("gen_random_uuid()"),
    )
    scan_id: Mapped[UUID] = mapped_column(
        ForeignKey("scan.id", ondelete="CASCADE"),
        nullable=False,
        index=True,
    )
    content: Mapped[str] = mapped_column(Text, nullable=False)
    created_by: Mapped[str] = mapped_column(nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        nullable=False,
        server_default=text("now()"),
    )

    # Relationship (optional but useful)
    # scan = relationship("Scan", back_populates="notes")

Step 2: Register the Model

In platform/db/models/__init__.py, add:

from db.models.scan_note import ScanNote

Step 3: Create Pydantic Schemas

Create platform/api/schemas/v1/scan_note.py:

from datetime import datetime
from uuid import UUID

from pydantic import BaseModel, Field


class ScanNoteCreate(BaseModel):
    content: str = Field(..., min_length=1, max_length=10000)


class ScanNoteResponse(BaseModel):
    id: UUID
    scan_id: UUID
    content: str
    created_by: str
    created_at: datetime

    model_config = {"from_attributes": True}

Step 4: Create the Service

Create platform/api/services/scan_note_service.py:

from uuid import UUID

from sqlalchemy import select
from sqlalchemy.orm import Session

from db.models.scan_note import ScanNote
from api.dependencies.db import TenantDB


class ScanNoteService:
    def __init__(self, tenant_db: TenantDB):
        self.db = tenant_db

    async def create_note(
        self,
        scan_id: UUID,
        content: str,
        created_by: str,
    ) -> ScanNote:
        note = ScanNote(
            scan_id=scan_id,
            content=content,
            created_by=created_by,
        )
        self.db.session.add(note)
        await self.db.session.commit()
        await self.db.session.refresh(note)
        return note

    async def list_notes(
        self,
        scan_id: UUID,
        limit: int = 100,
    ) -> list[ScanNote]:
        stmt = (
            select(ScanNote)
            .where(ScanNote.scan_id == scan_id)
            .order_by(ScanNote.created_at.desc())
            .limit(limit)
        )
        result = await self.db.session.execute(stmt)
        return list(result.scalars().all())

Step 5: Add the Routes

In platform/api/routes/v1/scans.py (or create a new file):

from api.schemas.v1.scan_note import ScanNoteCreate, ScanNoteResponse
from api.services.scan_note_service import ScanNoteService


@router.post(
    "/v1/scans/{scan_id}/notes",
    response_model=ScanNoteResponse,
    status_code=201,
)
async def create_scan_note(
    scan_id: UUID,
    data: ScanNoteCreate,
    user: ActiveUser = Depends(get_current_user),
    tenant_db: TenantDB = Depends(get_tenant_db),
) -> ScanNoteResponse:
    """Create a note on a scan."""
    # Verify scan exists
    scan_service = ScanService(tenant_db, encryption_service)
    try:
        await scan_service.get_scan(scan_id)
    except NotFoundError:
        raise HTTPException(status_code=404, detail="Scan not found")

    note_service = ScanNoteService(tenant_db)
    note = await note_service.create_note(
        scan_id=scan_id,
        content=data.content,
        created_by=user.email,
    )
    return ScanNoteResponse.model_validate(note)


@router.get(
    "/v1/scans/{scan_id}/notes",
    response_model=list[ScanNoteResponse],
)
async def list_scan_notes(
    scan_id: UUID,
    user: ActiveUser = Depends(get_current_user),
    tenant_db: TenantDB = Depends(get_tenant_db),
) -> list[ScanNoteResponse]:
    """List notes for a scan, newest first."""
    note_service = ScanNoteService(tenant_db)
    notes = await note_service.list_notes(scan_id)
    return [ScanNoteResponse.model_validate(n) for n in notes]

Step 6: Create Alembic Migration

# Generate migration
cd platform
alembic revision --autogenerate -m "add scan_note table"

# Review the generated migration file
# Then apply it
alembic upgrade head

Stretch Goals

• Add DELETE endpoint (only note author can delete)
• Add pagination to the list endpoint
• Add full-text search on note content
• Write unit tests for the service layer