🗄️ Table Management

Atomic table and index management utilities for tenant databases.

🎯 Overview

Function	Purpose
register_table	Create table from dataclass model
register_tables	Create multiple tables atomically
drop_table	Drop table if exists
create_index	Create index with dialect-specific SQL
drop_index	Drop index if exists
list_tables	List all tables in database
list_indexes	List indexes for a table

📋 Table Registration

Function	Purpose
register_table	Create single table from dataclass
register_tables	Create multiple tables atomically
drop_table	Drop table if exists

Example:

class Transaction:
    id: str
    amount: float
    date: str
    category: str = None

tenant_db = get_or_create_tenant_db("tenant_123")
transactions = register_table(tenant_db, Transaction, "transactions")

# Use table object for CRUD
transactions.insert(Transaction(id="t1", amount=100.0, date="2024-01-01"))

---

source

register_table

def register_table(
    tenant_db:Database, model_class:Type, table_name:str, pk:str='id'
):

Create a table from a dataclass model if it doesn’t exist (atomic).

---

source

register_tables

def register_tables(
    tenant_db:Database, models:List
)->Dict:

Create multiple tables atomically (all succeed or all rollback).

Parameters: - models: List of tuples [(ModelClass, table_name, pk), ...]

Returns: Dict mapping table names to table objects {table_name: table_object}

Example:

class Transaction:
    id: str; amount: float; date: str

class Connection:
    id: str; provider: str; status: str

tables = register_tables(tenant_db, [
    (Transaction, "transactions", "id"),
    (Connection, "connections", "id"),
])

tables["transactions"].insert(...)

---

source

drop_table

def drop_table(
    tenant_db:Database, table_name:str
)->None:

Drop a table if it exists (atomic operation).

📇 Index Management

Function	Purpose
create_index	Create index with dialect-specific SQL
create_indexes	Create multiple indexes atomically
drop_index	Drop index if exists

Parameters for create_index:

Parameter	Description
table_name	Name of the table
columns	List of column names to index
unique	If True, creates UNIQUE index (default: False)
index_name	Custom name (auto-generates idx_{table}_{cols} if None)

Example:

# Simple index
create_index(tenant_db, "transactions", ["date"])

# Composite unique index
create_index(tenant_db, "transactions", ["account_id", "external_id"], unique=True)

# Custom name
create_index(tenant_db, "transactions", ["category"], index_name="idx_txn_cat")

---

source

create_index

def create_index(
    tenant_db:Database, table_name:str, columns:List, unique:bool=False, index_name:str=None
)->None:

Create an index on a table if it doesn’t exist (atomic operation).

---

source

create_indexes

def create_indexes(
    tenant_db:Database, indexes:List
)->None:

Create multiple indexes atomically (all succeed or all rollback).

Parameters: - indexes: List of tuples [(table_name, columns, unique, index_name), ...] - index_name can be None for auto-generated names

Example:

create_indexes(tenant_db, [
    ("transactions", ["date"], False, None),
    ("transactions", ["account_id", "external_id"], True, "idx_txn_unique"),
    ("connections", ["provider"], False, None),
])

---

source

drop_index

def drop_index(
    tenant_db:Database, index_name:str, table_name:str=None
)->None:

Drop an index if it exists (atomic operation).

🔍 Schema Introspection

Function	Purpose
table_exists	Check if a table exists

Example:

if not table_exists(tenant_db, "transactions"):
    transactions = register_table(tenant_db, Transaction, "transactions")

---

source

table_exists

def table_exists(
    tenant_db:Database, table_name:str
)->bool:

Check if a table exists in the database.

Usage Example

Complete workflow: Define dataclass models → Get tenant DB → Register tables → Create indexes