feat(fc3b): migration 0012 + AppSetting model
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,36 @@
|
||||
"""fc3b: app_setting key/value table
|
||||
|
||||
Revision ID: 0012
|
||||
Revises: 0011
|
||||
Create Date: 2026-05-20
|
||||
|
||||
A simple key/value table for small app settings that don't fit
|
||||
ImportSettings. Initially seeds only `extension_api_key` (done in
|
||||
create_app on first boot — not in the migration, to keep it
|
||||
deterministic and independent of randomness).
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
revision: str = "0012"
|
||||
down_revision: Union[str, None] = "0011"
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_table(
|
||||
"app_setting",
|
||||
sa.Column("key", sa.String(length=64), primary_key=True),
|
||||
sa.Column("value", sa.Text(), nullable=False),
|
||||
sa.Column(
|
||||
"updated_at", sa.DateTime(timezone=True),
|
||||
nullable=False, server_default=sa.func.now(),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_table("app_setting")
|
||||
@@ -1,5 +1,6 @@
|
||||
"""All ORM models. Import this module to make every model visible to Alembic."""
|
||||
|
||||
from .app_setting import AppSetting
|
||||
from .artist import Artist
|
||||
from .base import Base
|
||||
from .credential import Credential
|
||||
@@ -22,6 +23,7 @@ from .tag_suggestion_rejection import TagSuggestionRejection
|
||||
|
||||
__all__ = [
|
||||
"Base",
|
||||
"AppSetting",
|
||||
"Artist",
|
||||
"Source",
|
||||
"Credential",
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
"""AppSetting — small key/value app config (extension API key, etc.).
|
||||
|
||||
For settings that need DB persistence + runtime mutation. Use
|
||||
`ImportSettings` for the structured import-tuning settings; this
|
||||
table is for one-off scalar values keyed by string.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import DateTime, String, Text, func
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from .base import Base
|
||||
|
||||
|
||||
class AppSetting(Base):
|
||||
__tablename__ = "app_setting"
|
||||
|
||||
key: Mapped[str] = mapped_column(String(64), primary_key=True)
|
||||
value: Mapped[str] = mapped_column(Text, nullable=False)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False,
|
||||
server_default=func.now(), onupdate=func.now(),
|
||||
)
|
||||
@@ -0,0 +1,32 @@
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
|
||||
from backend.app.models import AppSetting
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_app_setting_table_round_trip(db):
|
||||
db.add(AppSetting(key="extension_api_key", value="abc123"))
|
||||
await db.flush()
|
||||
row = (await db.execute(
|
||||
select(AppSetting).where(AppSetting.key == "extension_api_key")
|
||||
)).scalar_one()
|
||||
assert row.value == "abc123"
|
||||
assert row.updated_at is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_app_setting_upsert(db):
|
||||
db.add(AppSetting(key="k", value="v1"))
|
||||
await db.flush()
|
||||
row = (await db.execute(
|
||||
select(AppSetting).where(AppSetting.key == "k")
|
||||
)).scalar_one()
|
||||
row.value = "v2"
|
||||
await db.flush()
|
||||
again = (await db.execute(
|
||||
select(AppSetting.value).where(AppSetting.key == "k")
|
||||
)).scalar_one()
|
||||
assert again == "v2"
|
||||
Reference in New Issue
Block a user