Adopt Alembic: autogenerated baseline, stamp-or-upgrade startup

The baseline (c3da001a0eb3) was autogenerated against an empty Postgres
and schema-diffed against the live dev schema until column-identical
(two model fixes fell out: deferred_to_next_cycle now declares its
server_default, usersettings.data is JSONB on Postgres to match the
live type). Startup replaces init_db/run_migrations with
run_alembic_upgrade(): pre-Alembic databases that already match the
baseline are adopted via stamp; fresh databases build from the
migration. The section is serialized with a pg advisory lock because
prod uvicorn runs 2 workers, each executing the lifespan. Verified:
adoption + idempotent re-run on the dev DB, fresh 13-table build on an
empty DB, full container boot, 55/55 tests. (ARCH-02, BE-19)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Carlos Escalante
2026-06-09 20:31:55 -06:00
parent c725c1487d
commit 9a25c4baab
10 changed files with 519 additions and 73 deletions

View File

@@ -1,12 +1,17 @@
import enum
from datetime import date, datetime
from app.timeutil import utcnow
from typing import Optional
from sqlalchemy import JSON, Column, UniqueConstraint
from sqlalchemy.dialects.postgresql import JSONB
from sqlmodel import Field, Relationship, SQLModel
from app.timeutil import utcnow
# Live DBs store usersettings.data as jsonb; keep plain JSON elsewhere and on
# SQLite (tests).
JSON_OR_JSONB = JSON().with_variant(JSONB(), "postgresql")
class RecurringItemType(str, enum.Enum):
INCOME = "INCOME"
@@ -144,7 +149,9 @@ class TransactionBase(SQLModel):
bank: Bank = Bank.BAC
notes: Optional[str] = None
category_id: Optional[int] = Field(default=None, foreign_key="category.id")
deferred_to_next_cycle: bool = Field(default=False)
deferred_to_next_cycle: bool = Field(
default=False, sa_column_kwargs={"server_default": "false"}
)
class Transaction(TransactionBase, table=True):
@@ -227,7 +234,7 @@ class UserSettings(SQLModel, table=True):
key: str = Field(index=True, unique=True, default="default")
data: dict = Field(
default_factory=dict,
sa_column=Column(JSON, nullable=False, server_default="{}"),
sa_column=Column(JSON_OR_JSONB, nullable=False, server_default="{}"),
)
updated_at: datetime = Field(default_factory=utcnow)