Schema-level multi-tenancy with one common schema

[Gayathri-A-R]

Schema-level multi-tenancy with one common schema

I am referring https://alembic.sqlalchemy.org/en/latest/cookbook.html#rudimental-schema-level-multi-tenancy-for-postgresql-databases to implement schema level multi-tenancy. But what if I have an additional schema which is common for the whole database?

[zzzeek]

you would add another option handled by env.py that would also handle this special schema. if you had it explicitly in your models, then you'd bypass the logic that sets the schema, like:

from sqlalchemy import text

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    special_schema = context.get_x_argument(as_dictionary=True).get("special_schema")
    current_tenant = context.get_x_argument(as_dictionary=True).get("tenant")

    if current_tenant is None and special_schema is None:
       raise Exception("tenant or special_schema expected")
    elif current_tenant is not None and special_schema is not None:
       raise Exception("tenant and special_schema are mutually exclusive")

    with connectable.connect() as connection:

        # set search path on the connection, which ensures that
        # PostgreSQL will emit all CREATE / ALTER / DROP statements
        # in terms of this schema by default
        
        if current_tenant:
            connection.execute(text('set search_path to "%s"' % current_tenant))

            # make use of non-supported SQLAlchemy attribute to ensure
            # the dialect reflects tables in terms of the current tenant name
            connection.dialect.default_schema_name = current_tenant
        # otherwise it is assumed models have "schema='special_schema'" where they use this 
        # fixed schema

        context.configure(
            connection=connection,
            target_metadata=target_metadata,
        )

        with context.begin_transaction():
            context.run_migrations()

[Gayathri-A-R]

Did it with include_objects . Thank you.

[RicardoJDaleprane]

I know this discussion has been going on for a few years now, but could you tell me which flow I will use to be able to use Alembic's version control?
My env file

import asyncio
from logging.config import fileConfig


from alembic import context
from sqlalchemy import text
from sqlalchemy.ext.asyncio import async_engine_from_config
from sqlalchemy.ext.asyncio import AsyncEngine


from app.database.models import Base


URL_BD_ = 
config = context.config
config.set_main_option("sqlalchemy.url", URL_BD_)

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata



def run_migrations_offline() -> None:
   
    raise NotImplementedError("O modo offline não é implementado.")



async def run_async_migrations() -> None:
   
    connectable: AsyncEngine = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=None,
    )

 
    args = context.get_x_argument(as_dictionary=True)
    special_schema = args.get("special_schema")
    current_tenant = args.get("tenant")

    if current_tenant is None and special_schema is None:
        raise Exception("Você deve fornecer 'tenant' ou 'special_schema' como argumento.")
    elif current_tenant is not None and special_schema is not None:
        raise Exception("'tenant' e 'special_schema' não podem ser usados simultaneamente.")

    async with connectable.connect() as async_connection:
        await async_connection.run_sync(
            lambda connection: run_migrations_online_internal(connection, current_tenant, special_schema)
        )
    await connectable.dispose()


def run_migrations_online_internal(connection, current_tenant: str, special_schema: str) -> None:
    
    
    if current_tenant:
        connection.execute(text(f'SET search_path TO "{current_tenant}"'))
        connection.commit()
        

    
    def include_object(object, name, type_, reflected, compare_to):
       
        
        if type_ in {"index", "unique_constraint", "foreign_key_constraint"}:
            return True

        schema = getattr(object, "schema", None) 

        if special_schema and schema != special_schema:
            
            return False
        elif current_tenant and (schema is None or schema != "tenant"):
            
            return False

        return True

    
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        include_object=include_object,
    )

    with context.begin_transaction():
        context.run_migrations()



def run_migrations_online() -> None:
    asyncio.run(run_async_migrations())


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

The commands I am executing in this order:
alembic -x special_schema=shared revision -m "Initial Migration Shared" --autogenerate

alembic -x special_schema=shared upgrade head

alembic -x tenant=default_tenant revision -m "Initial Migration Tenant" --autogenerate

alembic -x special_schema=default_tenant upgrade head

When I try the second autogenerate I get the following error.

INFO [alembic.runtime.migration] Context impl PostgresqlImpl. INFO [alembic.runtime.migration] Will assume transactional DDL. ERROR [alembic.util.messaging] Target database is not up to date. FAILED: Target database is not up to date.

I probably didn't understand the correct flow I need to follow.

[zzzeek]

do the alembic -x special_schema=default_tenant upgrade head command before you run autogenerate as well then it will be up to date with whatever it's seeing

however if your search_paths here are overlapping then things might still not work out if the two separate migration streams see the same alembic_version table.

the main thing is to keep a separate alembic_version table for each series of migrations/autogenerates otherwise it's going to be confused about that. you can also based on the arguments coming in, point it to different alembic_version tables using version_table and version_table_schema

the main thing is: each directory of alembic version files gets its own alembic_version table

[RicardoJDaleprane]

I'm not sure how to proceed.
I configured 2 version tables, 1 for the common tables for all and 1 for the tenant tables
When I include the first tenant it works correctly, but when I try to add a second tenant it creates the tables and returns that the database is up to date.
Was your instruction to have a version table in each tenant schema?
or is it a single version table for all tenants? If this is the approach, how do you ensure that each new tenant can create the tables for that new tenant?
Tks

[zzzeek]

it really depends on what you are doing and how you intend for it to work.

if you want each tenant to run through the version files independently of the rest, then yes you'd need an alembic_version table per tenant.

this is just a table that stores what the latest version file is for a certain set of files on your filesystem somewhere. so if you want to run those files for like three different schemas independently, then they each have their own version.

[RicardoJDaleprane]

In the scope of my project, all tenants will have the same version of tables, so the idea was to have 2 version tables.
One table for the shared tables of the shared schema and one table for the tenants.

import logging
from logging.config import fileConfig
from alembic import context
from sqlalchemy import text, pool
from sqlalchemy.ext.asyncio import async_engine_from_config
from sqlalchemy.ext.asyncio import AsyncEngine
from app.database.models import Base

URL_BD_ = "url"
config = context.config
config.set_main_option("sqlalchemy.url", URL_BD_)

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata

logging.basicConfig()
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)


def run_migrations_offline() -> None:
    raise NotImplementedError("Modo offline não implementado.")


async def run_async_migrations() -> None:
    connectable: AsyncEngine = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    args = context.get_x_argument(as_dictionary=True)
    special_schema = args.get("special_schema")
    current_tenant = args.get("tenant")

    if current_tenant is None and special_schema is None:
        raise Exception("Você deve fornecer 'tenant' ou 'special_schema' como argumento.")
    elif current_tenant is not None and special_schema is not None:
        raise Exception("'tenant' e 'special_schema' não podem ser usados simultaneamente.")

    async with connectable.connect() as async_connection:
        await async_connection.run_sync(
            lambda connection: run_migrations_online_internal(connection, current_tenant, special_schema)
        )
    await connectable.dispose()


def run_migrations_online_internal(connection, current_tenant: str, special_schema: str) -> None:
   
    
    if special_schema == "shared":
        script_location = "alembic/versions/shared"
        version_table = "alembic_version_shared"
        schema_name = "shared"
    else:
        script_location = "alembic/versions/tenants"
        version_table = "alembic_version"
        schema_name = current_tenant

    version_table_schema = "public"  

   
    context.script.version_locations = [script_location]

   
    create_schema_if_not_exists(connection, schema_name)

    
    if current_tenant:
        connection.execute(text(f'SET search_path TO "{current_tenant}"'))
        connection.commit()

    def include_object(object, name, type_, reflected, compare_to):
        
        schema = getattr(object, "schema", None)
        args = context.get_x_argument(as_dictionary=True)
        special_schema = args.get("special_schema")

        
        if name == "alembic_version_shared" and schema == "public" and special_schema == "shared":
            return True
        elif name == "alembic_version" and schema == "public" and special_schema is None:
            return True

        
        if special_schema == "shared":
            if schema == "shared":
                return True
            else:
                return False

        
        if special_schema is None:
            if schema is None:
                return True
            else:
                return False

        return False

   
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        include_object=include_object,
        version_table=version_table,
        version_table_schema=version_table_schema,
    )

    with context.begin_transaction():
        context.run_migrations()


def create_schema_if_not_exists(connection, schema_name: str):
   
    query = text(f'CREATE SCHEMA IF NOT EXISTS "{schema_name}"')
    connection.execute(query)
    connection.commit()
    print(f"DEBUG: Esquema {schema_name} garantido no banco de dados.")


def run_migrations_online() -> None:
    asyncio.run(run_async_migrations())


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

I run these commands:

alembic -x special_schema=shared revision -m "Initial Migration Shared" --autogenerate
alembic -x special_schema=shared upgrade head
alembic -x tenant=default_tenant revision -m "Initial Migration Tenant" --autogenerate
alembic -x tenant=default_tenant upgrade head

With these commands I have the expected result:

• I create and save the version files in the correct location with the correct tables.
• I create the shared schema tables correctly
• I create the defautl_tenant schema tables correctly

The problem is when I try to create a new tenant with the command:
alembic -x tenant=tenant1 upgrade head

INFO [sqlalchemy.engine.Engine] SELECT public.alembic_version.version_num FROM public.alembic_version INFO [sqlalchemy.engine.Engine] [generated in 0.00039s] ()
I've done a lot of research but I haven't found a way to create a second scheme using the latest version of the migration file.

[zzzeek]

In the scope of my project, all tenants will have the same version of tables, so the idea was to have 2 version tables.
One table for the shared tables of the shared schema and one table for the tenants.

that's not possible here. you must have an alembic_version table per tenant since the tenant databases are upgraded individually. the "upgrade" command uses the data in alembic_version to know what commands it intends to issue, and it then stores the work that was done. if you run upgrade for tenant A, alembic_version is now updated and stores the information that corresponds to tenant A. if you then run upgrade for tenant B, if the same alembic_version table is consulted, it will show that work was already done and nothing needs to proceed.

therefore since you are running "upgrade" multiple times, once per tenant, you need multiple "alembic_version" tables, once per tenant