Is there any ability to work with schema-based multi-tenancy? Multiple databases?

[afparsons]

I am implementing schema-based multi-tenancy loosely based upon this excellent blog post from MergeBoard (however, I am using MySQL instead of PostgreSQL, so "database" and "schema" are interchangeable terms in this case).

That is to say:

• Tenant 1: {database_schema_1}.{table}.{column}, available at mysql://user:password@host:port/database_schema_1
• Tenant 2: {database_schema_2}.{table}.{column}, available at mysql://user:password@host:port/database_schema_2
• . . .
• Tenant N: {database_schema_n}.{table}.{column}, available at mysql://user:password@host:port/database_schema_n

Problem: I have successfully implemented AuthenticationBackend and am able to allow/deny users access to the admin. However, because the Admin takes Engine in its constructor (which in my case doesn't point to a specific database schema), attempting to view models throws this error:

sqlalchemy.exc.OperationalError: (MySQLdb.OperationalError) (1046, 'No database selected')
[SQL: SELECT count(table.id) AS count_1 
FROM table]

Question: Is there a way to set up sqladmin with multiple database schemas, such that it can dynamically set the bind connectable using information from an authorized user?

---

Multi-tenancy Implementation Details

Not all of these functions are in the same file, and some details are elided. The excerpts below are only to provide an overview.

# Nota bene: no database schema is specified! The engine is updated with the correct database schema in the `tenancy` context manager.
ENGINE: Engine = make_engine("mysql://user:password@host:port")

class TenantConfiguration(pydantic.BaseModel):
    database_schema: str


def get_tenant(authorization: str) -> TenantConfiguration:
    """
    Returns the correct TenantConfiguration using a passed authentication token
    """
    ...


@contextmanager
def tenancy(database_schema: Optional[str]) -> Session:
    """
    """
    if database_schema:
        schema_translate_map = {None: database_schema}
    else:
        schema_translate_map = None

    connectable = ENGINE.execution_options(schema_translate_map=schema_translate_map)
    session = Session(autocommit=False, autoflush=False, bind=connectable)
    try:
        yield session
    finally:
        session.close()


def get_database_session(
    authorization: str = Header(alias="Authorization"),
) -> Generator[Session, None, None]:
    tenant_configuration: TenantConfiguration = get_tenant(authorization)
    with tenancy(tenant_configuration.database_schema) as session:
        yield session


@fastapi_app.get("/item")
def example(session: Session = Depends(get_database_session)):
    """
    Returns items from the authorized tenant's database schema
    """

response: Response = requests.get(url="https://example.com/example/endpoint", headers={"Authorization": "token"})

# response.json() will only return the items from the database corresponding to the passed token