How to handle Schema multi-tennancy with Python + Flask + sqlAlchemy + PostgreSQL

What is multi-tenancy ?

Where you might need multi-tenancy ?

Why am i writing this post ?

What you will learn in this post ?

A) Hypothetical business use case :

B) Architecture, strategy and steps

Architecture idea :

Strategy :

C) Create and populate the database

Connection with dbeaver:

Create the schemas for the database :

Populate the database :

D) Connect the database to python

Config connection and init a session :

import osfrom sqlalchemy import create_engine, 
from sqlalchemy.orm import sessionmaker
NAME_DB = os.getenv('NAME_DB', 'postgres')PORT_DB = os.getenv('PORT_DB', '5432')PWD_DB = os.getenv('PWD_DB', 'postgres')HOST_DB = os.getenv('HOST_DB', 'localhost')USER_DB = os.getenv('USER_DB', 'postgres')conString = "postgres://YourUserName:YourPassword@YourHost:5432/YourDatabase"URI = f'postgresql+psycopg2://{USER_DB}:{PWD_DB}@{HOST_DB}:{PORT_DB}/{NAME_DB}'engine = create_engine(URI)session_maker = sessionmaker(bind=engine)# Here we are initializing a new session
session = session_maker()

Model mapping :

from sqlalchemy import Column, DateTime, Stringfrom sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()metadata = Base.metadataclass User(Base):
__tablename__ = 'users'__table_args__ = {'schema': 'public'}
id = Column(String, primary_key=True)creation_timestamp = Column(DateTime)name = Column(String, nullable=False)password = Column(String, nullable=False)mail = Column(String, nullable=False)organization_id = Column(String, nullable=False)

Changing dynamically the schema

engine = create_engine(URI, echo=True)Base = declarative_base(metadata=metadata)gen_session = sessionmaker()SELECTED_SCHEMA = "client1"# Here we map all the schema that are undifined (in our case that
# are different from the schema 'public' of table user) to the
connection = engine.connect().execution_options(schema_translate_map={None: SELECTED_SCHEMA})session = gen_session(bind=connection)

E) Implement and expose a service per tenant

Method to expose

from lib.models.base import BaseModel, dbfrom lib.models.item import Itemfrom lib.models.order import Orderclass User(BaseModel):__tablename__ = "users"__table_args__ = {'schema': 'public'}id = db.Column(db.Unicode(), primary_key=True)name = db.Column(db.Unicode(), nullable=False)mail = db.Column(db.Unicode(), nullable=False)password = db.Column(db.Unicode(), nullable=False)organization_id = db.Column(db.Unicode(), nullable=False)@classmethoddef _get_user_priced_orders(cls):  return cls.session.query(,Item.description,Item.price,Order.quantity,(Order.quantity * Item.price).label('Total')).join(Order).join(Item)def get_my_priced_orders(self):
query = self._get_user_priced_orders()
return query.filter( ==

Changing schema dynamically at each server request :

F) Test via postman

Postman usage :

Expected result :

client1 :

client2 :


G) Github code

Full stack devops based in Paris, python advocate, let’s connect on Linkedin :

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store