Database Admin
The Database Admin API provides Python bindings for multi-tenant database administration in PostgreSQL deployments.
These classes (DatabaseAdmin, TenantConfig, TenantCredentials) are
wheel-only and gated behind the wheel’s postgres Cargo feature
(crates/cloacina-python/src/lib.rs). The published PyPI wheel enables both
backends, so pip install cloaca includes them; they are absent from a
SQLite-only custom build and from the authoring surface inside packaged
workflows. At runtime the constructor additionally rejects non-PostgreSQL
URLs.
The main class for administrative operations on multi-tenant PostgreSQL databases.
DatabaseAdmin(database_url: str)
Parameters:
database_url(str): PostgreSQL connection string with administrative privileges. The URL must start withpostgres://orpostgresql://and include a database name in the path; anything else raisesRuntimeError.
Example:
import cloaca
admin = cloaca.DatabaseAdmin("postgresql://admin:password@localhost:5432/mydb")
create_tenant(config: TenantConfig) -> TenantCredentials
Creates a new tenant with dedicated schema and database user.
Parameters:
config(TenantConfig): Configuration for the new tenant
Returns:
TenantCredentials: Credentials and connection information for the new tenant
Raises: RuntimeError on failure (insufficient privileges, schema or
username already exists, connection issues, invalid schema/username).
Example:
config = cloaca.TenantConfig(
schema_name="tenant_acme",
username="acme_user",
# password omitted — auto-generate a secure password
)
credentials = admin.create_tenant(config)
print(f"Tenant created with schema: {credentials.schema_name}")
print(f"Connection string: {credentials.connection_string}")
remove_tenant(schema_name: str, username: str) -> None
Drops the tenant’s schema and database user.
Parameters:
schema_name(str): Schema of the tenant to removeusername(str): The tenant’s database username
Raises: RuntimeError on failure.
Example:
admin.remove_tenant("tenant_acme", "acme_user")
Configuration object for creating new tenants.
TenantConfig(schema_name: str, username: str, password: str | None = None)
Parameters:
schema_name(str): Name of the PostgreSQL schema for this tenantusername(str): Database username for this tenantpassword(str, optional): Password for the user. Omitted,None, or empty string means auto-generate a secure 32-character password atcreate_tenanttime
Example:
# With admin-provided password
config = cloaca.TenantConfig(
schema_name="tenant_acme",
username="acme_user",
password="secure_password123"
)
# With auto-generated password
config = cloaca.TenantConfig(
schema_name="tenant_acme",
username="acme_user",
)
schema_name(str): The schema name for the tenantusername(str): The database username for the tenantpassword(str): The password as configured (empty string when auto-generation was requested; the generated password is returned onTenantCredentials)
Returned credentials and connection information for a newly created tenant.
username(str): Database username for the tenantpassword(str): Database password for the tenantschema_name(str): PostgreSQL schema name for the tenantconnection_string(str): Complete PostgreSQL connection string for the tenant
Example:
credentials = admin.create_tenant(config)
# Access individual components
print(f"Username: {credentials.username}")
print(f"Password: {credentials.password}")
print(f"Schema: {credentials.schema_name}")
# Use connection string directly
runner = cloaca.DefaultRunner(credentials.connection_string)
import cloaca
# Set up admin connection
admin = cloaca.DatabaseAdmin("postgresql://admin:admin@localhost:5432/myapp")
# Create tenant with auto-generated password
config = cloaca.TenantConfig(
schema_name="tenant_customer123",
username="customer123_user",
password="" # Auto-generate
)
credentials = admin.create_tenant(config)
# Store credentials securely for the customer
# In production, you would save these to your user management system
class TenantManager:
def __init__(self, admin_db_url: str):
self.admin = cloaca.DatabaseAdmin(admin_db_url)
self.tenant_runners = {}
def onboard_customer(self, customer_id: str) -> dict:
"""Provision new customer tenant"""
config = cloaca.TenantConfig(
schema_name=f"tenant_{customer_id}",
username=f"{customer_id}_user",
password="" # Auto-generate secure password
)
credentials = self.admin.create_tenant(config)
# Create dedicated runner for this tenant
runner = cloaca.DefaultRunner(credentials.connection_string)
self.tenant_runners[customer_id] = runner
return {
"tenant_id": customer_id,
"schema": credentials.schema_name,
"username": credentials.username,
"connection_ready": True
}
def get_tenant_runner(self, customer_id: str):
"""Get workflow runner for specific tenant"""
return self.tenant_runners.get(customer_id)
When password is empty or not provided, the system generates a secure 32-character password using:
- Uppercase letters (A-Z)
- Lowercase letters (a-z)
- Numbers (0-9)
Special characters are excluded to avoid URL encoding issues in connection strings.
Created tenant users have:
- Full access to their dedicated schema
- No access to other tenants’ schemas
- No access to administrative functions
- No access to the public schema (by design)
Generated connection strings use unencoded passwords. The underlying database driver handles any necessary encoding automatically.
All admin failures surface as RuntimeError
(see Exceptions):
try:
credentials = admin.create_tenant(config)
except RuntimeError as e:
print(f"Failed to create tenant: {e}")
# Common causes:
# - Admin user lacks necessary privileges
# - Schema or username already exists
# - Database connection issues