script.pyΒΆ

 1# type: ignore
 2"""${message}
 3
 4Revision ID: ${up_revision}
 5Revises: ${down_revision | comma,n}
 6Create Date: ${create_date}
 7
 8"""
 9from __future__ import annotations
10
11import warnings
12from typing import TYPE_CHECKING
13
14import sqlalchemy as sa
15from alembic import op
16from advanced_alchemy.types import EncryptedString, EncryptedText, GUID, ORA_JSONB, DateTimeUTC
17from sqlalchemy import Text  # noqa: F401
18${imports if imports else ""}
19if TYPE_CHECKING:
20    from collections.abc import Sequence
21
22__all__ = ["downgrade", "upgrade", "schema_upgrades", "schema_downgrades", "data_upgrades", "data_downgrades"]
23
24sa.GUID = GUID
25sa.DateTimeUTC = DateTimeUTC
26sa.ORA_JSONB = ORA_JSONB
27sa.EncryptedString = EncryptedString
28sa.EncryptedText = EncryptedText
29
30# revision identifiers, used by Alembic.
31revision: str = ${repr(up_revision)}
32down_revision: str | None = ${repr(down_revision)}
33branch_labels: str  | Sequence[str] | None = ${repr(branch_labels)}
34depends_on: str  | Sequence[str] | None = ${repr(depends_on)}
35
36
37def upgrade() -> None:
38    with warnings.catch_warnings():
39        warnings.filterwarnings("ignore", category=UserWarning)
40        with op.get_context().autocommit_block():
41            schema_upgrades()
42            data_upgrades()
43
44def downgrade() -> None:
45    with warnings.catch_warnings():
46        warnings.filterwarnings("ignore", category=UserWarning)
47        with op.get_context().autocommit_block():
48            data_downgrades()
49            schema_downgrades()
50
51def schema_upgrades() -> None:
52    """schema upgrade migrations go here."""
53    ${upgrades if upgrades else "pass"}
54
55def schema_downgrades() -> None:
56    """schema downgrade migrations go here."""
57    ${downgrades if downgrades else "pass"}
58
59def data_upgrades() -> None:
60    """Add any optional data upgrade migrations here!"""
61
62def data_downgrades() -> None:
63    """Add any optional data downgrade migrations here!"""