Thread-safety of alembic.op

[MrP4p3r]

We have an application with schema-based multi-tenancy. Migrations are run by our database tool which internally uses alembic . We need to migrate a lot of databases, so we would like to do in parallel.

But there is an issue with alembic.op (and AFAIU alembic.context). It is not thread-safe. E. g. if one upgrade() starts just after another, Operations.context() overwrites contents of alembic.op with an _install_proxy() call. As a result, both migrations, running in parallel, share the same connection and MigrationContext/Operations instances, which obviously leads to severe consequences.

As far as I can see, there are multiple workarounds. E. g. we could pass Operations instance as an argument to upgrade() and downgrade() functions. However, the documented way of calling operations via alembic.op is preferred.

I would be glad to hear any thoughts on this issue.

[zzzeek]

We have an application with schema-based multi-tenancy. Migrations are run by our database tool which internally uses alembic . We need to migrate a lot of databases, so we would like to do in parallel.

But there is an issue with alembic.op (and AFAIU alembic.context). It is not thread-safe. E. g. if one upgrade() starts just after another, Operations.context() overwrites contents of alembic.op with an _install_proxy() call. As a result, both migrations, running in parallel, share the same connection and MigrationContext/Operations instances, which obviously leads to severe consequences.

that's correct, alembic.op is a global variable that is established within the scope of where env.py calls upon environment_context.run_migrations().

As far as I can see, there are multiple workarounds. E. g. we could pass Operations instance as an argument to upgrade() and downgrade() functions.

yes, you could do that, and it would not be too difficult. Look at the source (Well I think you did) to EnvironmentContext..run_migrations() and Operations.context() and we could write this to not use the global op namespace as:

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,
        )

        with context.begin_transaction():
            migration_context = context.get_context()
            op = Operations(migration_context)
            migration_context.run_migrations(op=op)

then you would have def upgrade(op): and def downgrade(op) methods in your migration scripts.

this is simple and useful enough that I could support adding a new template to Alembic for this; as a calling style, not as having any multithreading in the template itself.

As far as running multiple threads, you would still need to be careful that this occurs only for upgrade/downgrade operations and not operations that write into the ScriptDirectory, etc.

The other way to do it, and this might be better anyway for actual "parallelism", is to use multiprocessing instead of threading. Then you only need spin up workers inside of your env.py (I assume these databases all have their own alembic_version tables). You even could keep things very simple and run the top level commands (using the form at https://alembic.sqlalchemy.org/en/latest/api/commands.html) from mutiprocessing workers. That's basically equivalent to just running multiple alembic upgrade commands at the same time.

for any of these approaches, the main thing to be concerned about is that you are not changing the scripts folder with multiple commands at once (should be fine for upgrade/downgrade) and for any particular database / alembic_version table, only one connection at a time upgrades that particular database.

[MrP4p3r]

Thank you for your reply and I apologize for the delay.

then you would have def upgrade(op): and def downgrade(op) methods in your migration scripts.

I think we'll use this solution then. Supporting this style of calls in the library and documentation would be nice.

But I also think it would be great to somehow prevent op/context from being overwritten, because otherwise it can really break a lot of things, as it almost happened to us. I can't help but feel that this is a kind of a library API bug. I hope you don't mind considering the possible fixes to Alembic below.

• Raise an exception if the op/context proxy is already installed;
• Add a Lock to the proxy installation. Though it may break compatibility with asyncio in the future. So I guess it's not an option (?)
• Use contextvars to store a proxied object and a module-level __getattr__ to access its attributes.

[CaselIT]

I think the fist option, raising an error, may make sense and it should have fairly limited impacts
I think it should be enough to add a check here to verify that _proxy is missing or None

alembic/alembic/util/langhelpers.py

Line 59 in df26520

globals_["_proxy"] = self

@zzzeek what do you think?

[zzzeek]

re-entrant calls to an _install_proxy / _remove_proxy block should absolutely raise an exception. however, it would make existing approaches to concurrent migrations, like the one I suggested above, non-working,because there is no way to prevent the global proxy from being installed.

so a complete solution would be more involved:

1. global proxy raises if reentrant / concurrent
2. option to disable the global proxy instead

this would be 1.10 if you are motivated to work on it

I think we should stop there. because otherwise it gets into:

3. (I would rather not do this) cookbooks etc. for the above
4. from 3. means we will be getting bug reports / confused users / people wanting to do concurrent migrations for bad reasons

[MrP4p3r]

if you are motivated to work on it

I guess, it is addressed to me, right? 😅 If so, I will try to create an issue and play around with the source code as soon I have time.

2. option to disable the global proxy instead

Does it imply the implementation of the calling style choice (upgrade() vs upgrade(op))?