117 lines
4.0 KiB
Python
117 lines
4.0 KiB
Python
"""Alembic env — SQLModel + app.config.DATABASE_URL + защита autogenerate."""
|
||
import sys
|
||
from logging.config import fileConfig
|
||
from pathlib import Path
|
||
|
||
from alembic import context
|
||
from sqlalchemy import engine_from_config, pool
|
||
from sqlmodel import SQLModel
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
if str(PROJECT_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
from app import config as app_config # noqa: E402
|
||
from app import models # noqa: E402,F401
|
||
|
||
config = context.config
|
||
config.set_main_option("sqlalchemy.url", app_config.DATABASE_URL)
|
||
|
||
if config.config_file_name is not None:
|
||
fileConfig(config.config_file_name)
|
||
|
||
target_metadata = SQLModel.metadata
|
||
|
||
|
||
# ── Таблицы, существовавшие до Alembic ───────────────────────────────
|
||
_EXISTING_TABLES = {
|
||
"event", "eventsubscriber", "globalstatus", "notification",
|
||
"portalfile", "project", "projectmember", "projectpost",
|
||
"projectpostattachment", "projectsubscription", "projecttask",
|
||
"user", "usersession", "workshop", "workshopassistant",
|
||
"workshopfeedback", "workshopmaterial", "workshoppost",
|
||
"workshoppostattachment", "workshopregistration",
|
||
"workshopsession", "workshopsubscription",
|
||
}
|
||
|
||
|
||
def include_object(obj, name, type_, reflected, compare_to):
|
||
"""
|
||
FK/index/unique для существующих таблиц — игнорируются
|
||
(модели vs БД расходятся в именах/ondelete, это косметика).
|
||
Отражённое в БД, но отсутствующее в модели — не дропаем.
|
||
Новые таблицы/колонки — разрешены.
|
||
"""
|
||
if type_ in ("foreign_key_constraint", "index", "unique_constraint"):
|
||
parent = getattr(obj, "table", None)
|
||
parent_name = getattr(parent, "name", None)
|
||
if parent_name in _EXISTING_TABLES:
|
||
return False
|
||
if reflected:
|
||
return False
|
||
if reflected and compare_to is None:
|
||
return False
|
||
return True
|
||
|
||
|
||
def render_item(type_, obj, autogen_context):
|
||
"""
|
||
SQLModel использует свои типы (AutoString, UTCDateTime). В миграции их
|
||
нельзя писать как `sqlmodel.sql.sqltypes.AutoString()` — не будет
|
||
импортировано. Рендерим как нативные SQLAlchemy типы.
|
||
|
||
Покрываемые типы:
|
||
• AutoString → sa.String() (str-поля)
|
||
• UTCDateTime → sa.DateTime(timezone=True) (datetime-поля)
|
||
• Integer, Boolean, Float — нативные, не трогаем
|
||
"""
|
||
if type_ != "type":
|
||
return False
|
||
try:
|
||
import sqlmodel
|
||
except ImportError:
|
||
return False
|
||
if isinstance(obj, sqlmodel.sql.sqltypes.UTCDateTime):
|
||
return "sa.DateTime(timezone=True)"
|
||
if isinstance(obj, sqlmodel.sql.sqltypes.AutoString):
|
||
return "sa.String()"
|
||
return False
|
||
|
||
|
||
def run_migrations_offline() -> None:
|
||
context.configure(
|
||
url=config.get_main_option("sqlalchemy.url"),
|
||
target_metadata=target_metadata,
|
||
literal_binds=True,
|
||
dialect_opts={"paramstyle": "named"},
|
||
compare_type=True,
|
||
include_object=include_object,
|
||
render_item=render_item,
|
||
)
|
||
with context.begin_transaction():
|
||
context.run_migrations()
|
||
|
||
|
||
def run_migrations_online() -> None:
|
||
connectable = engine_from_config(
|
||
config.get_section(config.config_ini_section, {}),
|
||
prefix="sqlalchemy.",
|
||
poolclass=pool.NullPool,
|
||
)
|
||
with connectable.connect() as connection:
|
||
context.configure(
|
||
connection=connection,
|
||
target_metadata=target_metadata,
|
||
compare_type=True,
|
||
include_object=include_object,
|
||
render_item=render_item,
|
||
)
|
||
with context.begin_transaction():
|
||
context.run_migrations()
|
||
|
||
|
||
if context.is_offline_mode():
|
||
run_migrations_offline()
|
||
else:
|
||
run_migrations_online()
|