🏠 Multi-Tenant Setup

The central registry for multi-tenant SaaS applications - managing users, tenants, and access control.

🎯 Overview

The Host Database is the backbone of a multi-tenant architecture. It answers the critical questions:

Question	Model	Purpose
👤 Who is this person?	`GlobalUser`	Identity & authentication
🏢 Where is their data?	`TenantCatalog`	Database routing
🔑 What can they access?	`Membership`	Access control
💳 Are they paying?	`Subscription`	Billing status
📋 What happened?	`HostAuditLog`	Security audit trail
⚙️ Background work?	`SystemJob`	Async operations

🏗️ Architecture

                    ┌─────────────────────────────────────┐
                    │         🏠 HOST DATABASE            │
                    │  (Single source of truth)           │
                    ├─────────────────────────────────────┤
                    │  👤 GlobalUser    → Identity        │
                    │  🏢 TenantCatalog → DB Routing      │
                    │  🔑 Membership    → Access Control  │
                    │  💳 Subscription  → Billing         │
                    └───────────┬─────────────────────────┘
                                │
            ┌───────────────────┼───────────────────┐
            ▼                   ▼                   ▼
    ┌───────────────┐   ┌───────────────┐   ┌───────────────┐
    │ 🗄️ Tenant A   │   │ 🗄️ Tenant B   │   │ 🗄️ Tenant C   │
    │   Database    │   │   Database    │   │   Database    │
    │  (isolated)   │   │  (isolated)   │   │  (isolated)   │
    └───────────────┘   └───────────────┘   └───────────────┘

Key Principle: User authenticates once → Host routes to correct tenant → Tenant data is isolated

🛠️ Utilities

Helper functions used throughout the host database operations.

source

gen_id


def gen_id(
    
):

source

timestamp


def timestamp(
    
):

source

get_db_uri


def get_db_uri(
    db
)->str:

Extract SQLAlchemy connection URI from a Database object.

Safely renders the URL with the actual password (required for utilities like map_and_upsert that create new connections).

Why this is needed: - str(db.conn.engine.url) masks the password as *** - This causes “password authentication failed” errors - render_as_string(hide_password=False) reveals the actual password

Parameter	Description
`db`	fastsql/minidataapi Database object or HostDatabase instance

Returns: Full connection URI string with password

Example:

from fh_saas.db_host import get_db_uri, HostDatabase
from fh_saas.utils_polars_mapper import map_and_upsert

host_db = HostDatabase.from_env()
db_uri = get_db_uri(host_db.db)

# Now safe to use with map_and_upsert
map_and_upsert(df, 'my_table', 'id', db_uri)

Function	Description
`timestamp()`	🕐 Returns current UTC time in ISO format
`gen_id()`	🆔 Generates a unique 32-character hex ID
`get_db_uri(db)`	🔗 Extract full connection URI with password from Database object

📦 Core Models

These dataclasses define the host database schema. Each model maps to a table with the core_ or sys_ prefix.

source

StripeWebhookEvent


def StripeWebhookEvent(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Idempotency: Has this webhook already been processed?

Stores processed Stripe event IDs to prevent duplicate handling.

Attributes: id: Unique record ID (gen_id) event_id: Stripe event ID (e.g., evt_xxx) event_type: Stripe event type (e.g., customer.subscription.updated) status: Processing result: ‘processed’, ‘failed’, or ‘skipped_out_of_order’ payload_json: Optional JSON snapshot of the event result created_at: ISO timestamp of when the event was recorded

source

PricingPlan


def PricingPlan(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Pricing: Available subscription tiers with Stripe price IDs.

Stores pricing configuration in database for admin-configurable tiers. Each plan can have monthly and/or yearly billing intervals.

Attributes: id: Unique plan identifier (e.g., ‘basic’, ‘pro’, ‘enterprise’) name: Display name for UI (e.g., ‘Basic Plan’) description: Plan description for pricing page stripe_price_monthly: Stripe Price ID for monthly billing (price_xxx) stripe_price_yearly: Stripe Price ID for yearly billing (price_xxx) amount_monthly: Monthly price in cents (for display, e.g., 799 = $7.99) amount_yearly: Yearly price in cents (for display, e.g., 7900 = $79.00) currency: ISO currency code (default: ‘usd’) trial_days: Free trial period in days (default: 30) features: JSON array of feature keys enabled for this plan tier_level: Numeric level for feature gating (higher = more access) is_active: Whether plan is available for new subscriptions sort_order: Display order on pricing page created_at: Record creation timestamp

Example: >>> plan = PricingPlan( … id=‘pro’, … name=‘Pro Plan’, … stripe_price_monthly=‘price_1234’, … stripe_price_yearly=‘price_5678’, … amount_monthly=1999, … amount_yearly=19900, … tier_level=2, … features=‘[“api_access”, “exports”, “priority_support”]’, … )

source

SystemJob


def SystemJob(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Maintenance: Provisioning & Cleanups

source

HostAuditLog


def HostAuditLog(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Security: Who changed the system?

source

Subscription


def Subscription(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Billing: Are they allowed to use the app?

source

Membership


def Membership(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Router: Which tenants can they access?

source

TenantCatalog


def TenantCatalog(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Registry: Where is the database?

source

GlobalUser


def GlobalUser(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Identity: Who is this person?

📝 Model Details

Model	Table Name	Primary Key	Description
`GlobalUser`	`core_users`	`id`	OAuth identity, email, optional password hash, Stripe customer ID
`TenantCatalog`	`core_tenants`	`id`	Maps tenant ID to database URL, tracks plan tier and status
`Membership`	`core_memberships`	`id`	Links users to tenants with roles (`owner`, `admin`, `member`)
`Subscription`	`core_subscriptions`	`id`	Stripe subscription state for billing enforcement
`PricingPlan`	`core_pricing_plans`	`id`	Available subscription tiers with Stripe price IDs
`HostAuditLog`	`sys_audit_logs`	`id`	Immutable security log for compliance
`SystemJob`	`sys_jobs`	`id`	Background task queue for provisioning, cleanup
`StripeWebhookEvent`	`core_stripe_webhook_events`	`id`	Idempotency tracking for processed Stripe webhook events

🔌 HostDatabase Singleton

The HostDatabase class provides a singleton connection manager for the host database.

✅ Why Singleton? - Single connection pool shared across the application - Consistent transaction management - Easy dependency injection for testing

┌─────────────────────────────────────────────────────┐
│                  Application Start                   │
└─────────────────────┬───────────────────────────────┘
                      ▼
         ┌────────────────────────┐
         │  HostDatabase.from_env() │  ← Reads DB_* env vars
         └────────────┬───────────┘
                      ▼
         ┌────────────────────────┐
         │   Creates tables if    │
         │   they don't exist     │
         └────────────┬───────────┘
                      ▼
         ┌────────────────────────┐
         │  Returns singleton     │  ← Same instance everywhere
         │  instance              │
         └────────────────────────┘

source

HostDatabase


def HostDatabase(
    db_url:str=None
):

Singleton connection manager for the host database.

🔧 Methods

Method	Description
`from_env()`	🏭 Factory method - creates instance from environment variables
`commit()`	✅ Commit the current database transaction
`rollback()`	↩︎️ Rollback the current transaction on error
`reset_instance()`	🧪 Reset singleton (testing only)

🌍 Environment Variables for `from_env()`

Variable	Default	Description
`DB_TYPE`	`POSTGRESQL`	Database type (`POSTGRESQL` or `SQLITE`)
`DB_USER`	`postgres`	Database username
`DB_PASS`	(required)	Database password
`DB_HOST`	`localhost`	Database host
`DB_PORT`	`5432`	Database port
`DB_NAME`	`app_host`	Database name

source

HostDatabase.from_env


def from_env(
    
):

Create HostDatabase from DB_ environment variables.*

source

HostDatabase.commit


def commit(
    
):

Commit current transaction.

source

HostDatabase.rollback


def rollback(
    
):

Rollback current transaction.

source

HostDatabase.reset_instance


def reset_instance(
    
):

Reset singleton instance (testing only).

⚠️ Call close() first to release database connections!

🚀 Quick Start

from fh_saas.db_host import HostDatabase, GlobalUser, gen_id, timestamp

# Initialize singleton from environment
host_db = HostDatabase.from_env()

# Create a user
user = GlobalUser(
    id=gen_id(),
    email="user@example.com",
    oauth_id="google_123",
    created_at=timestamp()
)
host_db.global_users.insert(user)
host_db.commit()

# Query users
all_users = host_db.global_users()

💡 Tip: The singleton ensures you always get the same connection, so you can call HostDatabase.from_env() anywhere in your app.

🎯 Overview

🏗️ Architecture

🛠️ Utilities

gen_id

timestamp

get_db_uri

📦 Core Models

StripeWebhookEvent

PricingPlan

SystemJob

HostAuditLog

Subscription

Membership

TenantCatalog

GlobalUser

📝 Model Details

🔌 HostDatabase Singleton

HostDatabase

🔧 Methods

🌍 Environment Variables for from_env()

HostDatabase.from_env

HostDatabase.commit

HostDatabase.rollback

HostDatabase.reset_instance

🚀 Quick Start

🌍 Environment Variables for `from_env()`