env.py 832 B

12345678910111213141516171819
  1. """迁移使用同一集中配置;离线 SQL 与在线 SQLite 共享元数据。"""
  2. from alembic import context
  3. from sqlalchemy import create_engine
  4. from app.domain.models import Base
  5. from app.shared.config import settings
  6. url = settings.db_url.replace('sqlite+aiosqlite://', 'sqlite://')
  7. settings.data_path.mkdir(parents=True, exist_ok=True)
  8. if context.is_offline_mode():
  9. context.configure(url=url, target_metadata=Base.metadata, literal_binds=True, render_as_batch=True)
  10. with context.begin_transaction():
  11. context.run_migrations()
  12. else:
  13. engine = create_engine(url)
  14. with engine.connect() as connection:
  15. context.configure(connection=connection, target_metadata=Base.metadata, render_as_batch=True)
  16. with context.begin_transaction():
  17. context.run_migrations()
  18. engine.dispose()