Running migrations for all databases using different metadata #1403
Replies: 1 comment 5 replies
-
|
if you are using the i dont have a clear picture of what you have so far since your example does not show any "for loop" or what One is, use the command API directly and make your own front-end that does the things you want, in this case just runs the desired command multiple times with different configs This is the safe / easy / maybe a little tedious way to go but it's very simple and Alembic isn't made to jump through any kind of hoops it was not designed for. The other is more like what you have already, which I think can work, but what you need to do specifically for the autogenerate case (and likely not for the upgrade/downgrade case) is set version_locations at config time to have all of the target directories present at once in a list; this just makes it possible to refer to any specific version_location later on. Then, when each autogenerate runs, you need to set the MigrationScript.version_script to the path you want for each autogenerate run. You can intercept MigrationScript using a process_revision_directives hook. then when the "revision" command does that single step to write all the files, it will deposit them into the individual directories named in each MigrationScript. |
Beta Was this translation helpful? Give feedback.
-
|
@dataclasses.dataclass(slots=True, kw_only=True, frozen=True)
class _MigrationConfig:
name: str
metadata: MetaData
settings: DatabaseSettingsAs I said Ideally I would like to just run |
Beta Was this translation helpful? Give feedback.
-
|
@zzzeek Do I understand correctly that in order for the latter option to work I would have to execute alembic multiple times? import asyncio
import dataclasses
import importlib
import pkgutil
from collections.abc import Iterable, Sequence
from logging.config import fileConfig
from alembic import context
from alembic.config import Config
from alembic.operations import MigrationScript
from alembic.runtime.migration import MigrationContext
from sqlalchemy import MetaData, engine_from_config, pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import AsyncEngine
from app.connectors.db.base import DocumentBase, UserBase
from app.settings import DatabaseSettings, db_documents, db_user_data
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
@dataclasses.dataclass(slots=True, kw_only=True, frozen=True)
class _MigrationConfig:
name: str
metadata: MetaData
settings: DatabaseSettings
def do_run_migrations(
connection: Connection,
migration_config: _MigrationConfig,
version_location: str,
) -> None:
def process_revision_directives(
context: MigrationContext, # noqa: ARG001
revision: str | Iterable[str | None] | Iterable[str], # noqa: ARG001
directives: list[MigrationScript],
) -> None:
for directive in directives:
directive.version_path = version_location
context.configure(
connection=connection,
target_metadata=migration_config.metadata,
compare_server_default=True,
process_revision_directives=process_revision_directives,
)
with context.begin_transaction():
context.run_migrations()
async def do_run_migrations_async(
connectable: AsyncEngine,
migration_config: _MigrationConfig,
version_location: str,
) -> None:
async with connectable.connect() as conn:
await conn.run_sync(
do_run_migrations,
migration_config=migration_config,
version_location=version_location,
)
def run_migrations(
migration_configs: Sequence[_MigrationConfig],
migration_config: _MigrationConfig,
config: Config,
) -> None:
if not config.get_main_option("sqlalchemy.url"):
config.set_main_option("sqlalchemy.url", migration_config.settings.url)
script_location = config.get_section_option(
config.config_ini_section,
"script_location",
)
context.script.version_locations = [
f"{script_location}/versions/{cfg.name}" for cfg in migration_configs
]
version_location = f"{script_location}/versions/{migration_config.name}"
connectable = context.config.attributes.get("connection")
if not connectable:
if (section := config.get_section(config.config_ini_section)) is None:
raise ValueError
connectable = AsyncEngine(
engine_from_config(
section,
poolclass=pool.NullPool,
future=True,
),
)
if isinstance(connectable, AsyncEngine):
asyncio.run(
do_run_migrations_async(
connectable=connectable,
migration_config=migration_config,
version_location=version_location,
),
)
else:
do_run_migrations(
connectable=connectable,
migration_config=migration_config,
version_location=version_location,
)
def main() -> None:
migration_configs = [
_MigrationConfig(
metadata=UserBase.metadata,
settings=db_documents,
name="users",
),
_MigrationConfig(
metadata=DocumentBase.metadata,
settings=db_user_data,
name="documents",
),
]
for migration_config in migration_configs:
run_migrations(
migration_configs=migration_configs,
migration_config=migration_config,
config=config,
)
main() |
Beta Was this translation helpful? Give feedback.
-
|
I think you need to set those version_locations in the config directly using set_section_option, or otherwise you need to look into ScriptDirectory and reset the memoized _version_locations attribute. this is not a developed usecase of alembic so I'd use print / pdb to see how it proceeds. if some obvious hooks to make this easier stand out we can add them |
Beta Was this translation helpful? Give feedback.
-
|
Assigning to def process_revision_directives(
context: MigrationContext, # noqa: ARG001
revision: str | Iterable[str | None] | Iterable[str], # noqa: ARG001
directives: list[MigrationScript],
) -> None:
print(directives, version_location)
for directive in directives:
print(directive.upgrade_ops_list)
directive.version_path = version_locationseems like with the way I'm trying to do it the same |
Beta Was this translation helpful? Give feedback.
-
|
there should be multiple MigrationScript instances. feel free to make new ones and append them to the list inside the hook |
Beta Was this translation helpful? Give feedback.
-
Hello, I need to generate migrations for multiple databases and upgrade them.
Ideally I would want to be able to do autogeneration and upgrades with a single command (
alembic revision --autogenerate, etc) without having to run that separately for each database.To achieve that I've tried to run
run_migrationsfunction in a for loop with different arguments (db connection, metadata, etc):here I'm trying to override
version_locationsto a different subdirectory depending on the config, but it seems like alembic only generates revisions afterenv.pyis done executing.To work around that I decided to use the
-xargument and pass db I need to migrate into it to select a_MigrationConfigI need. This isn't quite the desired behavior, and I don't think built-inmultidbtemplate solves it either (since I want each db to have it's own Metadata), so I'm wondering if there's a solution for this problem. It seems like with this configuration migrations themselves should work fine, but revision autogeneration does not.Beta Was this translation helpful? Give feedback.
All reactions