Prerequisites
- Basic understanding of programming concepts 📝
- Python installation (3.8+) 🐍
- VS Code or preferred IDE 💻
What you'll learn
- Understand the concept fundamentals 🎯
- Apply the concept in real projects 🏗️
- Debug common issues 🐛
- Write clean, Pythonic code ✨
🎯 Introduction
Welcome to this exciting tutorial on multi-tenancy database patterns! 🎉 In this guide, we’ll explore how to build applications that serve multiple clients (tenants) from a single database infrastructure.
You’ll discover how multi-tenancy can transform your Python applications into scalable, cost-effective solutions. Whether you’re building SaaS applications 🌐, enterprise software 🏢, or cloud platforms ☁️, understanding multi-tenancy patterns is essential for writing robust, maintainable code.
By the end of this tutorial, you’ll feel confident implementing multi-tenant architectures in your own projects! Let’s dive in! 🏊♂️
📚 Understanding Multi-tenancy
🤔 What is Multi-tenancy?
Multi-tenancy is like an apartment building 🏢. Think of it as a single building (application) that houses multiple families (tenants) who share common infrastructure (database, servers) while keeping their spaces (data) private and secure.
In Python terms, multi-tenancy allows a single application instance to serve multiple customers while ensuring data isolation and security. This means you can:
- ✨ Serve multiple clients from one codebase
- 🚀 Scale efficiently by sharing resources
- 🛡️ Keep each tenant’s data completely isolated
💡 Why Use Multi-tenancy?
Here’s why developers love multi-tenant architectures:
- Cost Efficiency 💰: Share infrastructure across tenants
- Easier Maintenance 🔧: One codebase to update and deploy
- Scalability 📈: Add new tenants without new deployments
- Resource Optimization ⚡: Better utilization of database connections
Real-world example: Imagine building a project management tool 📊. With multi-tenancy, you can serve hundreds of companies from the same application while keeping each company’s projects private!
🔧 Basic Syntax and Usage
📝 Pattern 1: Shared Database, Shared Schema
Let’s start with the simplest approach:
# 👋 Hello, Multi-tenancy!
import sqlite3
from contextlib import contextmanager
class SharedSchemaDB:
def __init__(self, db_path):
self.db_path = db_path
self._init_db()
def _init_db(self):
# 🎨 Creating our shared schema
with self._get_db() as conn:
conn.execute('''
CREATE TABLE IF NOT EXISTS tasks (
id INTEGER PRIMARY KEY,
tenant_id TEXT NOT NULL, # 🏢 Tenant identifier
title TEXT NOT NULL,
completed BOOLEAN DEFAULT 0
)
''')
conn.execute('CREATE INDEX IF NOT EXISTS idx_tenant ON tasks(tenant_id)')
@contextmanager
def _get_db(self):
# 🔄 Database connection manager
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
try:
yield conn
conn.commit()
finally:
conn.close()
def add_task(self, tenant_id, title):
# ➕ Add task for specific tenant
with self._get_db() as conn:
conn.execute(
'INSERT INTO tasks (tenant_id, title) VALUES (?, ?)',
(tenant_id, title)
)
print(f"✅ Added task for tenant {tenant_id}: {title}")💡 Explanation: Notice how we include tenant_id in every table! This ensures data isolation at the row level.
🎯 Pattern 2: Separate Schemas
Here’s a more isolated approach:
# 🏗️ Pattern 2: Each tenant gets their own schema
class SeparateSchemaDB:
def __init__(self, db_path):
self.db_path = db_path
self.tenants = {} # 🗂️ Track tenant schemas
def create_tenant_schema(self, tenant_id):
# 🎨 Create isolated schema for tenant
schema_name = f"tenant_{tenant_id}"
with sqlite3.connect(self.db_path) as conn:
# Create tenant-specific tables
conn.execute(f'''
CREATE TABLE IF NOT EXISTS {schema_name}_tasks (
id INTEGER PRIMARY KEY,
title TEXT NOT NULL,
completed BOOLEAN DEFAULT 0
)
''')
self.tenants[tenant_id] = schema_name
print(f"🏢 Created schema for tenant: {tenant_id}")
def get_tenant_connection(self, tenant_id):
# 🔌 Get connection for specific tenant
if tenant_id not in self.tenants:
self.create_tenant_schema(tenant_id)
return TenantConnection(self.db_path, self.tenants[tenant_id])
class TenantConnection:
def __init__(self, db_path, schema_name):
self.db_path = db_path
self.schema_name = schema_name
def add_task(self, title):
# ➕ Add task in tenant's schema
with sqlite3.connect(self.db_path) as conn:
conn.execute(
f'INSERT INTO {self.schema_name}_tasks (title) VALUES (?)',
(title,)
)
print(f"✨ Task added to {self.schema_name}!")💡 Practical Examples
🛒 Example 1: E-commerce Platform
Let’s build a multi-tenant e-commerce system:
# 🛍️ Multi-tenant e-commerce platform
from datetime import datetime
import json
class MultiTenantStore:
def __init__(self):
self.db_path = "ecommerce.db"
self._init_database()
def _init_database(self):
# 🎨 Create our multi-tenant schema
with sqlite3.connect(self.db_path) as conn:
conn.execute('''
CREATE TABLE IF NOT EXISTS stores (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
config TEXT, # 🎨 Store customization
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
conn.execute('''
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY,
store_id TEXT NOT NULL,
name TEXT NOT NULL,
price REAL NOT NULL,
stock INTEGER DEFAULT 0,
emoji TEXT, # Every product needs an emoji! 🎁
FOREIGN KEY (store_id) REFERENCES stores(id)
)
''')
conn.execute('''
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY,
store_id TEXT NOT NULL,
customer_email TEXT NOT NULL,
total REAL NOT NULL,
status TEXT DEFAULT 'pending',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (store_id) REFERENCES stores(id)
)
''')
def create_store(self, store_id, name, config=None):
# 🏪 Create a new tenant (store)
with sqlite3.connect(self.db_path) as conn:
conn.execute(
'INSERT INTO stores (id, name, config) VALUES (?, ?, ?)',
(store_id, name, json.dumps(config or {}))
)
print(f"🎉 Welcome {name} to our platform!")
return Store(self.db_path, store_id)
class Store:
def __init__(self, db_path, store_id):
self.db_path = db_path
self.store_id = store_id
def add_product(self, name, price, stock, emoji="📦"):
# 🛍️ Add product to this store
with sqlite3.connect(self.db_path) as conn:
conn.execute(
'''INSERT INTO products (store_id, name, price, stock, emoji)
VALUES (?, ?, ?, ?, ?)''',
(self.store_id, name, price, stock, emoji)
)
print(f"{emoji} Added {name} to your store!")
def create_order(self, customer_email, items):
# 🛒 Create order for this store
total = sum(item['price'] * item['quantity'] for item in items)
with sqlite3.connect(self.db_path) as conn:
conn.execute(
'''INSERT INTO orders (store_id, customer_email, total)
VALUES (?, ?, ?)''',
(self.store_id, customer_email, total)
)
print(f"💰 New order! Total: ${total:.2f}")
return total
def get_analytics(self):
# 📊 Get store-specific analytics
with sqlite3.connect(self.db_path) as conn:
conn.row_factory = sqlite3.Row
# Total revenue
revenue = conn.execute(
'''SELECT SUM(total) as revenue FROM orders
WHERE store_id = ? AND status != 'cancelled' ''',
(self.store_id,)
).fetchone()['revenue'] or 0
# Product count
products = conn.execute(
'SELECT COUNT(*) as count FROM products WHERE store_id = ?',
(self.store_id,)
).fetchone()['count']
# Order count
orders = conn.execute(
'SELECT COUNT(*) as count FROM orders WHERE store_id = ?',
(self.store_id,)
).fetchone()['count']
print(f"📊 Store Analytics:")
print(f" 💰 Revenue: ${revenue:.2f}")
print(f" 📦 Products: {products}")
print(f" 🛒 Orders: {orders}")
return {'revenue': revenue, 'products': products, 'orders': orders}
# 🎮 Let's use it!
platform = MultiTenantStore()
# Create two different stores
tech_store = platform.create_store("tech_haven", "Tech Haven", {
"theme": "dark",
"currency": "USD"
})
book_store = platform.create_store("book_world", "Book World", {
"theme": "light",
"currency": "USD"
})
# Add products to each store
tech_store.add_product("Laptop", 999.99, 10, "💻")
tech_store.add_product("Mouse", 29.99, 50, "🖱️")
book_store.add_product("Python Guide", 39.99, 25, "📘")
book_store.add_product("Coffee Table Book", 24.99, 15, "☕")🎯 Try it yourself: Add a feature to handle inventory updates when orders are placed!
🎮 Example 2: Multi-tenant Task Management
Let’s make a scalable task management system:
# 🏆 Multi-tenant task management with advanced features
from enum import Enum
from typing import Dict, List, Optional
import hashlib
class IsolationLevel(Enum):
ROW_LEVEL = "row" # 🔄 Shared tables with tenant_id
SCHEMA_LEVEL = "schema" # 🏗️ Separate schemas
DATABASE_LEVEL = "database" # 🏢 Separate databases
class TenantManager:
def __init__(self, isolation_level: IsolationLevel):
self.isolation_level = isolation_level
self.tenants: Dict[str, 'Tenant'] = {}
self._init_system()
def _init_system(self):
# 🎨 Initialize based on isolation level
if self.isolation_level == IsolationLevel.ROW_LEVEL:
print("🔄 Using row-level isolation")
self.db = RowLevelDB()
elif self.isolation_level == IsolationLevel.SCHEMA_LEVEL:
print("🏗️ Using schema-level isolation")
self.db = SchemaLevelDB()
else:
print("🏢 Using database-level isolation")
self.db = DatabaseLevelDB()
def create_tenant(self, tenant_id: str, name: str, plan: str = "free"):
# 🎯 Create new tenant
tenant = Tenant(tenant_id, name, plan, self.db)
self.tenants[tenant_id] = tenant
# Initialize tenant space
self.db.initialize_tenant(tenant_id)
print(f"🎉 Welcome {name}! Plan: {plan} 🚀")
return tenant
class Tenant:
def __init__(self, tenant_id: str, name: str, plan: str, db):
self.tenant_id = tenant_id
self.name = name
self.plan = plan
self.db = db
self.users: Dict[str, 'User'] = {}
def add_user(self, email: str, name: str, role: str = "member"):
# 👤 Add user to tenant
user_id = hashlib.md5(f"{self.tenant_id}:{email}".encode()).hexdigest()[:8]
user = User(user_id, email, name, role, self.tenant_id)
self.users[user_id] = user
self.db.add_user(self.tenant_id, user)
print(f"✅ Added {name} ({role}) to {self.name}")
return user
def get_usage_stats(self):
# 📊 Get tenant usage statistics
stats = self.db.get_tenant_stats(self.tenant_id)
# Apply plan limits
limits = {
"free": {"users": 5, "tasks": 100},
"pro": {"users": 50, "tasks": 10000},
"enterprise": {"users": -1, "tasks": -1} # Unlimited
}
plan_limits = limits.get(self.plan, limits["free"])
print(f"📊 Usage Stats for {self.name}:")
print(f" 👥 Users: {stats['users']}/{plan_limits['users'] if plan_limits['users'] > 0 else '∞'}")
print(f" 📝 Tasks: {stats['tasks']}/{plan_limits['tasks'] if plan_limits['tasks'] > 0 else '∞'}")
return stats, plan_limits
class User:
def __init__(self, user_id: str, email: str, name: str, role: str, tenant_id: str):
self.user_id = user_id
self.email = email
self.name = name
self.role = role
self.tenant_id = tenant_id
# 🔧 Database implementations
class RowLevelDB:
def __init__(self):
self.conn = sqlite3.connect(":memory:")
self._create_tables()
def _create_tables(self):
# 🏗️ Create shared tables with tenant_id
self.conn.execute('''
CREATE TABLE tenants (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
plan TEXT DEFAULT 'free',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
self.conn.execute('''
CREATE TABLE users (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
email TEXT NOT NULL,
name TEXT NOT NULL,
role TEXT DEFAULT 'member',
FOREIGN KEY (tenant_id) REFERENCES tenants(id),
UNIQUE(tenant_id, email)
)
''')
self.conn.execute('''
CREATE TABLE tasks (
id INTEGER PRIMARY KEY,
tenant_id TEXT NOT NULL,
user_id TEXT NOT NULL,
title TEXT NOT NULL,
completed BOOLEAN DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (tenant_id) REFERENCES tenants(id),
FOREIGN KEY (user_id) REFERENCES users(id)
)
''')
# 🚀 Create indexes for performance
self.conn.execute('CREATE INDEX idx_users_tenant ON users(tenant_id)')
self.conn.execute('CREATE INDEX idx_tasks_tenant ON tasks(tenant_id)')
def initialize_tenant(self, tenant_id: str):
# 🎯 Initialize tenant in shared tables
pass # No special initialization needed for row-level
def add_user(self, tenant_id: str, user: User):
# 👤 Add user with tenant isolation
self.conn.execute(
'''INSERT INTO users (id, tenant_id, email, name, role)
VALUES (?, ?, ?, ?, ?)''',
(user.user_id, tenant_id, user.email, user.name, user.role)
)
self.conn.commit()
def get_tenant_stats(self, tenant_id: str):
# 📊 Get stats with tenant filtering
cursor = self.conn.cursor()
users = cursor.execute(
'SELECT COUNT(*) FROM users WHERE tenant_id = ?',
(tenant_id,)
).fetchone()[0]
tasks = cursor.execute(
'SELECT COUNT(*) FROM tasks WHERE tenant_id = ?',
(tenant_id,)
).fetchone()[0]
return {'users': users, 'tasks': tasks}
# 🎮 Demo time!
if __name__ == "__main__":
# Create multi-tenant system
system = TenantManager(IsolationLevel.ROW_LEVEL)
# Create two companies
startup = system.create_tenant("startup_inc", "Startup Inc", "free")
enterprise = system.create_tenant("big_corp", "Big Corporation", "enterprise")
# Add users to each tenant
startup.add_user("[email protected]", "Alice", "admin")
startup.add_user("[email protected]", "Bob", "member")
enterprise.add_user("[email protected]", "CEO", "admin")
enterprise.add_user("[email protected]", "Manager", "admin")
enterprise.add_user("[email protected]", "Developer", "member")
# Check usage
print("\n📈 Checking usage limits:")
startup.get_usage_stats()
enterprise.get_usage_stats()🚀 Advanced Concepts
🧙♂️ Advanced Pattern: Hybrid Isolation
When you’re ready to level up, try this advanced pattern:
# 🎯 Advanced hybrid isolation strategy
class HybridTenantDB:
def __init__(self):
self.shared_db = sqlite3.connect("shared.db")
self.tenant_dbs = {} # 🗂️ Premium tenants get own DB
self._init_shared_schema()
def _init_shared_schema(self):
# 🏗️ Shared schema for free/basic tenants
self.shared_db.execute('''
CREATE TABLE IF NOT EXISTS tenant_data (
tenant_id TEXT NOT NULL,
key TEXT NOT NULL,
value TEXT,
encrypted BOOLEAN DEFAULT 0,
PRIMARY KEY (tenant_id, key)
)
''')
def get_connection(self, tenant_id: str, tier: str):
# 🪄 Route based on tenant tier
if tier in ["enterprise", "dedicated"]:
# Premium tenants get dedicated database
if tenant_id not in self.tenant_dbs:
self.tenant_dbs[tenant_id] = sqlite3.connect(f"{tenant_id}.db")
return self.tenant_dbs[tenant_id], "dedicated"
else:
# Free/Pro tenants share database
return self.shared_db, "shared"
def migrate_tenant(self, tenant_id: str, from_tier: str, to_tier: str):
# 🚀 Migrate tenant between isolation levels
print(f"🔄 Migrating {tenant_id} from {from_tier} to {to_tier}")
if from_tier in ["free", "pro"] and to_tier in ["enterprise", "dedicated"]:
# Move from shared to dedicated
self._migrate_to_dedicated(tenant_id)
elif from_tier in ["enterprise", "dedicated"] and to_tier in ["free", "pro"]:
# Move from dedicated to shared
self._migrate_to_shared(tenant_id)
print(f"✅ Migration complete!")🏗️ Advanced Pattern: Sharding by Tenant
For massive scale:
# 🚀 Tenant-based sharding for horizontal scaling
class ShardedTenantDB:
def __init__(self, shard_count: int = 4):
self.shard_count = shard_count
self.shards = {}
self._init_shards()
def _init_shards(self):
# 🎲 Initialize database shards
for i in range(self.shard_count):
self.shards[i] = sqlite3.connect(f"shard_{i}.db")
self._create_shard_schema(self.shards[i])
def _create_shard_schema(self, conn):
# 📋 Each shard has same schema
conn.execute('''
CREATE TABLE IF NOT EXISTS tenant_data (
tenant_id TEXT NOT NULL,
data_type TEXT NOT NULL,
data TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (tenant_id, data_type)
)
''')
def _get_shard(self, tenant_id: str):
# 🎯 Determine shard using consistent hashing
hash_value = int(hashlib.md5(tenant_id.encode()).hexdigest(), 16)
shard_id = hash_value % self.shard_count
return self.shards[shard_id], shard_id
def store_data(self, tenant_id: str, data_type: str, data: str):
# 💾 Store data in appropriate shard
conn, shard_id = self._get_shard(tenant_id)
conn.execute(
'''INSERT OR REPLACE INTO tenant_data (tenant_id, data_type, data)
VALUES (?, ?, ?)''',
(tenant_id, data_type, data)
)
conn.commit()
print(f"✨ Stored in shard {shard_id} for tenant {tenant_id}")⚠️ Common Pitfalls and Solutions
😱 Pitfall 1: Forgetting Tenant Context
# ❌ Wrong way - no tenant isolation!
def get_all_users():
return db.execute("SELECT * FROM users").fetchall() # 💥 Returns ALL tenants' users!
# ✅ Correct way - always filter by tenant!
def get_tenant_users(tenant_id):
return db.execute(
"SELECT * FROM users WHERE tenant_id = ?",
(tenant_id,)
).fetchall() # 🛡️ Only returns specific tenant's users🤯 Pitfall 2: Connection Pool Exhaustion
# ❌ Dangerous - creating connections per tenant!
class BadTenantDB:
def __init__(self):
self.connections = {}
def get_connection(self, tenant_id):
if tenant_id not in self.connections:
self.connections[tenant_id] = sqlite3.connect(f"{tenant_id}.db")
return self.connections[tenant_id] # 💥 Connections never closed!
# ✅ Safe - use connection pooling!
from queue import Queue
import threading
class SafeTenantDB:
def __init__(self, pool_size=10):
self.pool = Queue(maxsize=pool_size)
self._init_pool()
def _init_pool(self):
# 🏊 Initialize connection pool
for _ in range(self.pool.maxsize):
conn = sqlite3.connect("tenants.db", check_same_thread=False)
self.pool.put(conn)
@contextmanager
def get_connection(self):
# ✅ Safely borrow and return connections
conn = self.pool.get()
try:
yield conn
finally:
self.pool.put(conn)🛠️ Best Practices
- 🎯 Always Include Tenant Context: Never query without tenant_id filtering
- 📝 Use Database Indexes: Index on tenant_id for performance
- 🛡️ Implement Row-Level Security: Use database features when available
- 🎨 Plan for Migration: Design to move tenants between isolation levels
- ✨ Monitor Resource Usage: Track per-tenant metrics
🧪 Hands-On Exercise
🎯 Challenge: Build a Multi-tenant Blog Platform
Create a multi-tenant blogging system:
📋 Requirements:
- ✅ Support multiple blog sites from one application
- 🏷️ Each blog has its own posts, authors, and comments
- 👤 Authors can only see/edit their own blog’s content
- 📊 Track views and engagement per blog
- 🎨 Each blog needs custom theming!
🚀 Bonus Points:
- Add subscription tiers with different limits
- Implement data export for individual blogs
- Create a blog migration feature
💡 Solution
🔍 Click to see solution
# 🎯 Multi-tenant blog platform solution!
from datetime import datetime
import json
from typing import Optional, List, Dict
class BlogPlatform:
def __init__(self):
self.db = sqlite3.connect("blogs.db")
self._init_schema()
def _init_schema(self):
# 🏗️ Create multi-tenant blog schema
self.db.executescript('''
CREATE TABLE IF NOT EXISTS blogs (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
domain TEXT UNIQUE,
theme TEXT DEFAULT '{"color": "blue", "font": "sans-serif"}',
tier TEXT DEFAULT 'free',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS authors (
id INTEGER PRIMARY KEY,
blog_id TEXT NOT NULL,
email TEXT NOT NULL,
name TEXT NOT NULL,
role TEXT DEFAULT 'writer',
FOREIGN KEY (blog_id) REFERENCES blogs(id),
UNIQUE(blog_id, email)
);
CREATE TABLE IF NOT EXISTS posts (
id INTEGER PRIMARY KEY,
blog_id TEXT NOT NULL,
author_id INTEGER NOT NULL,
title TEXT NOT NULL,
content TEXT NOT NULL,
status TEXT DEFAULT 'draft',
views INTEGER DEFAULT 0,
published_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (blog_id) REFERENCES blogs(id),
FOREIGN KEY (author_id) REFERENCES authors(id)
);
CREATE TABLE IF NOT EXISTS comments (
id INTEGER PRIMARY KEY,
blog_id TEXT NOT NULL,
post_id INTEGER NOT NULL,
author_name TEXT NOT NULL,
content TEXT NOT NULL,
approved BOOLEAN DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (blog_id) REFERENCES blogs(id),
FOREIGN KEY (post_id) REFERENCES posts(id)
);
CREATE INDEX idx_posts_blog ON posts(blog_id);
CREATE INDEX idx_comments_blog ON comments(blog_id);
CREATE INDEX idx_authors_blog ON authors(blog_id);
''')
self.db.commit()
def create_blog(self, blog_id: str, name: str, domain: str, tier: str = "free"):
# 🎉 Create new blog
theme = {
"free": {"color": "blue", "font": "sans-serif"},
"pro": {"color": "custom", "font": "custom", "logo": True},
"business": {"color": "custom", "font": "custom", "logo": True, "custom_css": True}
}.get(tier, {})
self.db.execute(
'''INSERT INTO blogs (id, name, domain, theme, tier)
VALUES (?, ?, ?, ?, ?)''',
(blog_id, name, domain, json.dumps(theme), tier)
)
self.db.commit()
print(f"🎊 Blog '{name}' created at {domain}!")
return Blog(self.db, blog_id, tier)
class Blog:
def __init__(self, db, blog_id: str, tier: str):
self.db = db
self.blog_id = blog_id
self.tier = tier
self.limits = {
"free": {"posts": 10, "authors": 2, "storage_mb": 100},
"pro": {"posts": 100, "authors": 10, "storage_mb": 1000},
"business": {"posts": -1, "authors": -1, "storage_mb": 10000}
}.get(tier, {})
def add_author(self, email: str, name: str, role: str = "writer"):
# 👤 Add author with tier limits
current_authors = self.db.execute(
'SELECT COUNT(*) FROM authors WHERE blog_id = ?',
(self.blog_id,)
).fetchone()[0]
if self.limits["authors"] > 0 and current_authors >= self.limits["authors"]:
print(f"⚠️ Author limit reached! Upgrade to add more.")
return None
cursor = self.db.execute(
'''INSERT INTO authors (blog_id, email, name, role)
VALUES (?, ?, ?, ?)''',
(self.blog_id, email, name, role)
)
self.db.commit()
print(f"✅ Added {name} as {role}!")
return cursor.lastrowid
def create_post(self, author_id: int, title: str, content: str):
# 📝 Create post with limits
current_posts = self.db.execute(
'SELECT COUNT(*) FROM posts WHERE blog_id = ?',
(self.blog_id,)
).fetchone()[0]
if self.limits["posts"] > 0 and current_posts >= self.limits["posts"]:
print(f"⚠️ Post limit reached! Upgrade to create more.")
return None
cursor = self.db.execute(
'''INSERT INTO posts (blog_id, author_id, title, content)
VALUES (?, ?, ?, ?)''',
(self.blog_id, author_id, title, content)
)
self.db.commit()
print(f"📝 Created post: {title}")
return cursor.lastrowid
def publish_post(self, post_id: int):
# 🚀 Publish a post
self.db.execute(
'''UPDATE posts SET status = 'published', published_at = CURRENT_TIMESTAMP
WHERE id = ? AND blog_id = ?''',
(post_id, self.blog_id)
)
self.db.commit()
print(f"🎉 Post {post_id} published!")
def get_analytics(self):
# 📊 Get blog analytics
stats = self.db.execute('''
SELECT
COUNT(DISTINCT p.id) as total_posts,
COUNT(DISTINCT CASE WHEN p.status = 'published' THEN p.id END) as published_posts,
SUM(p.views) as total_views,
COUNT(DISTINCT c.id) as total_comments,
COUNT(DISTINCT a.id) as total_authors
FROM blogs b
LEFT JOIN posts p ON b.id = p.blog_id
LEFT JOIN comments c ON b.id = c.blog_id
LEFT JOIN authors a ON b.id = a.blog_id
WHERE b.id = ?
''', (self.blog_id,)).fetchone()
print(f"📊 Analytics for blog {self.blog_id}:")
print(f" 📝 Posts: {stats[1]}/{stats[0]} published")
print(f" 👀 Total views: {stats[2] or 0}")
print(f" 💬 Comments: {stats[3]}")
print(f" ✍️ Authors: {stats[4]}")
return dict(zip(['total_posts', 'published_posts', 'total_views',
'total_comments', 'total_authors'], stats))
def export_data(self):
# 📦 Export blog data
data = {
"blog_id": self.blog_id,
"posts": [],
"authors": [],
"comments": []
}
# Export posts
posts = self.db.execute(
'SELECT * FROM posts WHERE blog_id = ?',
(self.blog_id,)
).fetchall()
data["posts"] = [dict(zip([d[0] for d in self.db.description], post))
for post in posts]
print(f"📦 Exported {len(data['posts'])} posts")
return data
# 🎮 Test it out!
platform = BlogPlatform()
# Create two blogs
tech_blog = platform.create_blog("tech_tales", "Tech Tales", "techtales.com", "pro")
food_blog = platform.create_blog("foodie", "Foodie Finds", "foodiefinds.com", "free")
# Add authors
john_id = tech_blog.add_author("[email protected]", "John", "admin")
jane_id = tech_blog.add_author("[email protected]", "Jane", "writer")
# Create posts
post1 = tech_blog.create_post(john_id, "Python Multi-tenancy Guide", "Content here...")
post2 = tech_blog.create_post(jane_id, "Database Patterns", "More content...")
# Publish and analyze
tech_blog.publish_post(post1)
tech_blog.get_analytics()
# Check limits on free tier
foodie_id = food_blog.add_author("[email protected]", "Chef", "admin")
for i in range(3):
food_blog.add_author(f"writer{i}@foodie.com", f"Writer {i}", "writer")🎓 Key Takeaways
You’ve learned so much! Here’s what you can now do:
- ✅ Design multi-tenant architectures with confidence 💪
- ✅ Choose the right isolation level for your needs 🛡️
- ✅ Implement secure data separation in real projects 🎯
- ✅ Handle tenant migrations like a pro 🐛
- ✅ Build scalable SaaS applications with Python! 🚀
Remember: Multi-tenancy is about balance - security, performance, and cost. Choose wisely! 🤝
🤝 Next Steps
Congratulations! 🎉 You’ve mastered multi-tenancy database patterns!
Here’s what to do next:
- 💻 Practice with the blog platform exercise above
- 🏗️ Build a small SaaS project using these patterns
- 📚 Explore database-specific features (PostgreSQL schemas, MySQL databases)
- 🌟 Share your multi-tenant architecture experiences!
Remember: Every successful SaaS started with understanding multi-tenancy. Keep building, keep learning, and most importantly, have fun! 🚀
Happy coding! 🎉🚀✨