Integration Cheatsheet 04 — Alembic + FastAPI Workflow

1. Edit src/myapp/models.py (add column / table)
2. alembic revision --autogenerate -m "msg"
3. Review migrations/versions/<rev>_msg.py
4. alembic upgrade head
5. Update Pydantic schemas (if needed)
6. Update API handlers
7. Update tests
8. uvicorn reload picks up changes

# 1. models.py
class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str]
    last_login_at: Mapped[datetime | None] = mapped_column(  # NEW
        DateTime(timezone=True), default=None,
    )

# 2. Generate
alembic revision --autogenerate -m "add user.last_login_at"

# 3. Review (migrations/versions/<rev>_add_user_last_login_at.py)
def upgrade():
    op.add_column("users", sa.Column("last_login_at", sa.DateTime(timezone=True), nullable=True))

def downgrade():
    op.drop_column("users", "last_login_at")

# 4. Apply
alembic upgrade head

# 5. Update Pydantic
class UserRead(BaseModel):
    id: int
    email: str
    last_login_at: datetime | None = None     # NEW
    model_config = {"from_attributes": True}

# 6. Handler
@app.post("/login")
async def login(form: OAuth2PasswordRequestForm = Depends(), db = Depends(get_db)):
    user = await authenticate(form.username, form.password, db)
    user.last_login_at = datetime.utcnow()    # NEW
    await db.commit()
    ...

# 7. Test
async def test_login_updates_last_login(client, db_session):
    user = UserFactory.build()
    db_session.add(user); await db_session.commit()
    
    r = await client.post("/login", data={"username": user.email, "password": "..."})
    await db_session.refresh(user)
    assert user.last_login_at is not None

# models.py — add new column
class User(Base):
    name: Mapped[str | None]      # OLD: still present
    full_name: Mapped[str | None] # NEW

alembic revision --autogenerate -m "add full_name (phase 1)"

def upgrade():
    op.add_column("users", sa.Column("full_name", sa.String, nullable=True))
    op.execute("UPDATE users SET full_name = name")

# models.py — remove old
class User(Base):
    full_name: Mapped[str]        # NOT NULL now

alembic revision --autogenerate -m "drop user.name (phase 2)"

def upgrade():
    op.alter_column("users", "full_name", nullable=False)
    op.drop_column("users", "name")

# models.py
class User(Base):
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
    title: Mapped[str]
    author: Mapped[User] = relationship(back_populates="posts")

alembic revision --autogenerate -m "add posts table"
alembic upgrade head

class PostCreate(BaseModel):
    title: str

class PostRead(BaseModel):
    id: int
    author_id: int
    title: str
    model_config = {"from_attributes": True}

@app.post("/posts", response_model=PostRead, status_code=201)
async def create_post(
    data: PostCreate,
    user: User = Depends(current_user),
    db: AsyncSession = Depends(get_db),
):
    post = Post(author_id=user.id, title=data.title)
    db.add(post)
    await db.commit()
    await db.refresh(post)
    return post

class UserRead(BaseModel):
    id: int
    email: str
    display_name: str = Field(alias="email")   # output only

    @computed_field
    @property
    def is_internal(self) -> bool:
        return self.email.endswith("@company.com")

# models.py
class User(Base):
    email: Mapped[str] = mapped_column(unique=True, index=True)   # add index=True

alembic revision --autogenerate -m "add ix_users_email"

def upgrade():
    op.execute("CREATE INDEX CONCURRENTLY ix_users_email ON users (email)")

# Terminal 1: API
uvicorn src.myapp.main:app --reload

# Terminal 2: editor; make changes
# Terminal 3:
alembic revision --autogenerate -m "..."
alembic upgrade head