Created
June 25, 2022 22:56
-
-
Save tusharvikky/5e9fd36b58d9b1900ee7d0cb7c052ac3 to your computer and use it in GitHub Desktop.
Creating shared schema multi-tenancy using python
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
app = FastAPI( | |
title="Project X", | |
description="Project X API", | |
version="1.0.0" | |
], | |
) | |
app.include_router(v1_router, prefix="/api/v1", dependencies=[Depends(AppOrigin)]) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from fastapi import Request, Header, HTTPException | |
from core.db import session | |
from app.models import Tenant | |
class AppOrigin: | |
def __init__(self, request: Request, app_origin: int = Header(None)): | |
if not app_origin: | |
raise HTTPException(status_code=422, detail="App-Origin header invalid") | |
tenant = session.query(Tenant).filter(Team.id==app_origin).first() | |
request.tenant = tenant | |
session.info['tenant'] = tenant | |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from sqlalchemy import Column, Unicode, BigInteger, Boolean | |
from core.db import Base | |
from core.db.mixins import TenantMixin | |
class User(Base, TenantMixin): | |
__tablename__ = "users" | |
__multitenant__ = True | |
id = Column(BigInteger, primary_key=True, autoincrement=True) | |
password = Column(Unicode(255), nullable=False) | |
email = Column(Unicode(255), nullable=False, unique=True) | |
name = Column(Unicode(255)) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from sqlalchemy import Column, inspect, BigInteger, func, event | |
from sqlalchemy.ext.declarative import declared_attr | |
from sqlalchemy.orm import Query | |
from core.db import session | |
class TenantMixin: | |
@declared_attr | |
def tenant_id(cls): | |
if not cls.__multitenant__: | |
return None | |
return Column(BigInteger, nullable=False, default=_current_session_tenant) | |
def _current_session_tenant(): | |
try: | |
tenant = session.info.get('tenant', False) | |
if not tenant: | |
return 1 | |
return tenant.id | |
except: | |
return 1 | |
@event.listens_for(Query, 'before_compile', retval=True) | |
def before_compile(query): | |
tenant_safe = query._execution_options.get('tenant_safe', False) | |
if tenant_safe: | |
return query | |
for column in query.column_descriptions: | |
entity = column['entity'] | |
if entity is None: | |
continue | |
inspector = inspect(column['entity']) | |
mapper = getattr(inspector, 'mapper', None) | |
if mapper and issubclass(mapper.class_, TenantMixin): | |
query = query.enable_assertions(False).filter( | |
entity.tenant_id.__eq__(query.session.info.get('tenant').id), | |
) | |
return query |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
userById = session.query(User).filter(User.id == user_id).first() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment