[Python] SQLAlchemy multitenancy (shared table)

I am setting up a new multitenant application and wonder how to handle database access. I am using a Postgres database. A Flask application is reading from the database. A Prefect job uses the same code base and updates the database regularly.

Decisions made so far:

• As onboarding/offboarding of tenants in a "one schema per tenant" database is pretty hard, we decided to go with shared tables (shared tables have a column tenant_id).
• As we may migrate to another database system, we do not want to use the RLS feature of postgres.

Now I am wondering if/why there is no buildin support for this shared-table-multitenancy setup in SQLAlchemy?! I found the old MultiAlchemy library, but it was archived in 2022 and the last commit was 9 years ago: https://github.com/mwhite/MultiAlchemy

I started writing my own multitenancy contextmanager (using ORM global criteria), but it's pretty hard to do so...

class HasClientColumn(object):
"""Mixin that identifies a class as having a tenant/client column"""
client_id = Column(BigInteger, nullable=False)


class Testerle(HasClientColumn, Base):
__tablename__ = "testerle"
id = Column(BigInteger, primary_key=True, autoincrement=True)
secret = Column(Text)


class Testerle2(HasClientColumn, Base):
__tablename__ = "testerle2"
id = Column(BigInteger, primary_key=True, autoincrement=True)
secret = Column(Text)

db_sessionmaker = sessionmaker(
autocommit=False, autoflush=False, bind=db_engine, future=True
)

def get_client_scoped_db_session(client_id: int):
session = db_sessionmaker()

# Overwrite execute method so nobody uses it without ORM
session.execute = lambda x, *args, **kwargs: _raise(
Exception(
"client_scoped_db_session not meant to be used with execute command (unsafe). Use ORM!"
)
)

# Docs: https://docs.sqlalchemy.org/en/14/orm/session_events.html#do-orm-execute-global-criteria
@event.listens_for(session, "do_orm_execute")
def _do_orm_execute(orm_execute_state):
if (
orm_execute_state.is_select
and not orm_execute_state.is_column_load
and not orm_execute_state.is_relationship_load
):
orm_execute_state.statement = orm_execute_state.statement.options(
with_loader_criteria(
HasClientColumn,
lambda cls: cls.client_id == client_id,
include_aliases=True,
)
)

return session

@contextmanager
def with_client_scoped_db_session(client_id: int):
session = False
try:
logging.debug(f"Establishing DB session for client {client_id}")
session = get_client_scoped_db_session(client_id)
yield session
session.commit()
except Exception as e:
logging.exception(f"Exception in with_client_scoped_db({client_id}): {e}")
if session:
session.rollback()
raise e
finally:
logging.debug(f"Closing DB session for client {client_id}")
if session:
session.close()


This works pretty good when just calling things like

with with_client_scoped_db_session(1) as session:
session.query(Testerle).all()


but whenever I do a join like

with with_client_scoped_db_session(1) as session:
session.query(Testerle, Testerle2.secret).join(Testerle2, Testerle.id == Testerle2.id).all()


it runs into my custom exception saying that the execute method is forbidden (client_scoped_db_session not meant to be used with execute command (unsafe). Use ORM!). Of course I could not overwrite session.execute, but then every developer could use the client scoped session to run SQL on all clients data (e.g. session.execute(text("SELECT * from testerle")).fetchall()) -> unsafe.

Any ideas how to implement stable and secure multitenancy with shared tables using SQLAlchemy without Postgress RLS?

Continue reading...