| from logging.config import fileConfig |
| from sqlalchemy import engine_from_config |
| from sqlalchemy import pool |
| from alembic import context |
| import sys |
| from pathlib import Path |
|
|
| |
| sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) |
|
|
| from src.core.config import settings |
| from src.models import User, Task |
| from sqlmodel import SQLModel |
|
|
| |
| config = context.config |
|
|
| |
| config.set_main_option("sqlalchemy.url", settings.DATABASE_URL) |
|
|
| |
| if config.config_file_name is not None: |
| fileConfig(config.config_file_name) |
|
|
| |
| target_metadata = SQLModel.metadata |
|
|
|
|
| def run_migrations_offline() -> None: |
| """Run migrations in 'offline' mode.""" |
| url = config.get_main_option("sqlalchemy.url") |
| context.configure( |
| url=url, |
| target_metadata=target_metadata, |
| literal_binds=True, |
| dialect_opts={"paramstyle": "named"}, |
| ) |
|
|
| with context.begin_transaction(): |
| context.run_migrations() |
|
|
|
|
| def run_migrations_online() -> None: |
| """Run migrations in 'online' mode.""" |
| 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 |
| ) |
|
|
| with context.begin_transaction(): |
| context.run_migrations() |
|
|
|
|
| if context.is_offline_mode(): |
| run_migrations_offline() |
| else: |
| run_migrations_online() |
|
|