76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
"""add postgres extensions
|
|
|
|
Revision ID: 65aa95e8f8cf
|
|
Revises:
|
|
Create Date: 2026-03-14 11:59:55.364307
|
|
|
|
"""
|
|
|
|
import warnings
|
|
from typing import TYPE_CHECKING
|
|
|
|
import sqlalchemy as sa
|
|
from alembic import op
|
|
from advanced_alchemy.types import EncryptedString, EncryptedText, GUID, ORA_JSONB, DateTimeUTC, StoredObject, PasswordHash, FernetBackend
|
|
from advanced_alchemy.types.encrypted_string import PGCryptoBackend
|
|
from advanced_alchemy.types.password_hash.argon2 import Argon2Hasher
|
|
from advanced_alchemy.types.password_hash.passlib import PasslibHasher
|
|
from advanced_alchemy.types.password_hash.pwdlib import PwdlibHasher
|
|
from pwdlib.hashers.argon2 import Argon2Hasher as PwdlibArgon2Hasher
|
|
from sqlalchemy import Text # noqa: F401
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Sequence
|
|
|
|
__all__ = ["downgrade", "upgrade", "schema_upgrades", "schema_downgrades", "data_upgrades", "data_downgrades"]
|
|
|
|
sa.GUID = GUID
|
|
sa.DateTimeUTC = DateTimeUTC
|
|
sa.ORA_JSONB = ORA_JSONB
|
|
sa.EncryptedString = EncryptedString
|
|
sa.EncryptedText = EncryptedText
|
|
sa.StoredObject = StoredObject
|
|
sa.PasswordHash = PasswordHash
|
|
sa.Argon2Hasher = Argon2Hasher
|
|
sa.PasslibHasher = PasslibHasher
|
|
sa.PwdlibHasher = PwdlibHasher
|
|
sa.FernetBackend = FernetBackend
|
|
sa.PGCryptoBackend = PGCryptoBackend
|
|
|
|
# revision identifiers, used by Alembic.
|
|
revision = '65aa95e8f8cf'
|
|
down_revision = None
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def upgrade() -> None:
|
|
op.execute(sa.text('CREATE EXTENSION IF NOT EXISTS "pgcrypto"'))
|
|
op.execute(sa.text('CREATE EXTENSION IF NOT EXISTS "pg_trgm"'))
|
|
with warnings.catch_warnings():
|
|
warnings.filterwarnings("ignore", category=UserWarning)
|
|
with op.get_context().autocommit_block():
|
|
schema_upgrades()
|
|
data_upgrades()
|
|
|
|
def downgrade() -> None:
|
|
with warnings.catch_warnings():
|
|
warnings.filterwarnings("ignore", category=UserWarning)
|
|
with op.get_context().autocommit_block():
|
|
data_downgrades()
|
|
schema_downgrades()
|
|
|
|
def schema_upgrades() -> None:
|
|
"""schema upgrade migrations go here."""
|
|
pass
|
|
|
|
def schema_downgrades() -> None:
|
|
"""schema downgrade migrations go here."""
|
|
pass
|
|
|
|
def data_upgrades() -> None:
|
|
"""Add any optional data upgrade migrations here!"""
|
|
|
|
def data_downgrades() -> None:
|
|
"""Add any optional data downgrade migrations here!"""
|