Db up downgrade snapshot

[JustKuzya]

The code changes fixes #474 and has up-/down-grade logic for alembic up-/down-grades

[jkglasbrenner]

The migration script is mixing the ORM and Core styles of SQLAlchemy in a way that I don't think you need to do here, and its making it harder to understand what is happening. Also, in downgrade(), you shouldn't need that counting operation because the resource_type field is a primary key already. The way the migration scripts are applied (e.g. you cannot apply the same upgrade or downgrade twice in a row) should also prevent a duplicate row from being inserted even if the database engine ignores the uniqueness constraint.

[jkglasbrenner]

Here's how you can write this migration script using just the ORM. Note that you need to create both Upgrade and Downgrade versions of your ORM classes in order for things to work correctly (based on the comments in your code, I suspect that you ran into this and that's why you started using operations from SQLAlchemy Core). Both the upgrade() and the downgrade() migrations query and check for the record where resource_type = 'resource_snapshot', and use that as a guard against deleting or inserting the record when its unexpectedly missing/present.

"""Up/Down-grade resource snapshot types as delete/insert 'resource_snapshot' [GH-Issue #474]

Revision ID: 0ca6eca33569
Revises: 6a75ede23821
Create Date: 2024-12-12 00:47:23.575103

"""

from typing import Annotated

import sqlalchemy as sa
from alembic import op
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    MappedAsDataclass,
    mapped_column,
    sessionmaker,
)

# revision identifiers, used by Alembic.
revision = "0ca6eca33569"
down_revision = "6a75ede23821"
branch_labels = None
depends_on = None

text_ = Annotated[str, mapped_column(sa.Text())]


class UpgradeBase(DeclarativeBase, MappedAsDataclass):
    pass


class DowngradeBase(DeclarativeBase, MappedAsDataclass):
    pass


class ResourceTypeUpgrade(UpgradeBase):
    __tablename__ = "resource_types"
    resource_type: Mapped[text_] = mapped_column(primary_key=True)


class ResourceTypeDowngrade(DowngradeBase):
    __tablename__ = "resource_types"
    resource_type: Mapped[text_] = mapped_column(primary_key=True)


def upgrade():
    """Alembic Upgrade Hook-Point that deletes the entry from
    resource_types table with value 'resource_snapshot'
    """
    bind = op.get_bind()
    Session = sessionmaker(bind=bind)

    with Session() as session:
        stmt = sa.select(ResourceTypeUpgrade).where(
            ResourceTypeUpgrade.resource_type == "resource_snapshot"
        )
        resource_snapshot_type = session.scalar(stmt)

        if resource_snapshot_type is not None:
            session.delete(resource_snapshot_type)

        session.commit()


def downgrade():
    """Alembic Downgrade Hook-Point that reinstates the entry to
    resource_types table with value 'resource_snapshot'
    """
    bind = op.get_bind()
    Session = sessionmaker(bind=bind)

    with Session() as session:
        stmt = sa.select(ResourceTypeUpgrade).where(
            ResourceTypeUpgrade.resource_type == "resource_snapshot"
        )
        resource_snapshot_type = session.scalar(stmt)

        if resource_snapshot_type is None:
            session.add(ResourceTypeUpgrade(resource_type="resource_snapshot"))

        session.commit()

[jkglasbrenner]

See #703 (comment).