MultiAlchemy is an experimental SQLAlchemy extension that makes it easy to write row-based multi-tenant applications without having to manually ensure data separation for every operation.
Here's an example schema, taken from the tests, that has a tenant table, one other global table (users), and one per-tenant table (posts).
from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
import multialchemy
engine = create_engine("sqlite:///multi_blog.db")
Base = declarative_base(cls=multialchemy.Base)
@Base.tenant_class
class Tenant(Base):
__tablename__ = 'tenants'
__multitenant__ = False
id = Column(Integer, primary_key=True)
name = Column(String(200))
class User(Base):
__tablename__ = 'users'
__multitenant__ = False
id = Column(Integer, primary_key=True)
name = Column(String(200))
class Post(Base):
__tablename__ = 'posts'
id = Column(Integer, primary_key=True)
title = Column(String(200))
author_id = Column(Integer, ForeignKey('users.id'))
author = relationship(User, backref='posts')If we look at the SQL emitted when creating these tables, we see that a
tenant_id foreign key was added automatically.
CREATE TABLE posts (
id INTEGER NOT NULL,
title VARCHAR(200),
author_id INTEGER,
tenant_id INTEGER,
PRIMARY KEY (id),
FOREIGN KEY(author_id) REFERENCES users (id),
FOREIGN KEY(tenant_id) REFERENCES tenants (id)
);Now let's create a multitenant-enabled session:
>>> TenantSession = sessionmaker(bind=engine, class_=multialchemy.TenantSession)
>>> session = TenantSession()Then we can find a tenant using an unsafe query that bypasses tenant checking, and bind it as the current session's tenant.
>>> tenant = session.query(Tenant, safe=False).first()
>>> session.tenant = tenantIf we do a query involving a multitenant model, a filter will automatically be added to limit the results to only those matching the current session's tenant.
>>> print(str(session.query(Post)))
SELECT posts.id AS posts_id, posts.title AS posts_title, posts.author_id AS posts_author_id, posts.tenant_id AS posts_tenant_id
FROM posts
WHERE posts.tenant_id = :tenant_id_1This works for joins too.
>>> print(str(session.query(User).join(Post)))
SELECT users.id AS users_id, users.name AS users_name
FROM users JOIN posts ON users.id = posts.author_id
WHERE posts.tenant_id = :tenant_id_1You can bypass the tenant checking by passing safe=False to session.query().
Instances also automatically get assigned the correct tenant_id when you add
them to the session:
>>> post = Post(title='Baz', author_id=1)
>>> session.add(post)
>>> post.tenant_id
1Copyright 2014 Michael White
Released under the MIT License. See LICENSE.txt.
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent