API Reference

Complete API reference for the Cloudflare SaaS Platform library.

Core Classes

CloudflareSaaSPlatform

class cloudflare_saas.CloudflareSaaSPlatform(config: Config, storage: StorageAdapter | None = None)

Bases: LoggerMixin

Main platform orchestrator with pluggable storage.

This class coordinates R2 storage, Cloudflare custom hostnames, DNS verification, and tenant management with configurable persistence.

__init__(config: Config, storage: StorageAdapter | None = None)

async create_tenant(name: str, slug: str, owner_id: str | None = None, metadata: Dict | None = None) → Tenant

Create a new tenant.

async get_tenant(tenant_id: str) → Tenant

Get tenant by ID.

async list_tenants(limit: int = 100, offset: int = 0) → List[Tenant]

List tenants with pagination.

async delete_tenant(tenant_id: str) → None

Delete tenant and all associated resources.

async resolve_tenant_from_host(host: str) → str | None

Resolve tenant ID from hostname.

async deploy_tenant_site(tenant_id: str, local_path: str, base_prefix: str = '') → DeploymentResult

Deploy static site for tenant.

async get_deployment_status(tenant_id: str) → Dict

Get deployment status for tenant.

async add_custom_domain(tenant_id: str, domain: str, verification_method: VerificationMethod = VerificationMethod.HTTP) → HostnameVerificationInstructions

Start custom domain onboarding process.

async get_domain_status(domain: str) → CustomDomain

Get status of custom domain provisioning.

async remove_custom_domain(domain: str) → None

Remove custom domain.

async list_tenant_domains(tenant_id: str) → List[CustomDomain]

List all domains for a tenant.

Config

class cloudflare_saas.Config(*, cloudflare_api_token: str, cloudflare_account_id: str, cloudflare_zone_id: str, r2_access_key_id: str, r2_secret_access_key: str, r2_bucket_name: str, r2_endpoint: str | None = None, r2_jurisdiction: str | None = None, d1_database_id: str | None = None, d1_jurisdiction: str | None = None, platform_domain: str, worker_script_name: str = 'site-router', worker_script_template_name: str | None = None, internal_api_key: str | None = None, log_level: str = 'INFO', log_format: str = 'detailed', log_file: str | None = None, enable_console_logging: bool = True, enable_custom_hostnames: bool = True, default_cache_ttl: int = 604800)

Bases: BaseModel

Configuration for Cloudflare SaaS platform.

cloudflare_api_token: str

cloudflare_account_id: str

cloudflare_zone_id: str

r2_access_key_id: str

r2_secret_access_key: str

r2_bucket_name: str

r2_endpoint: str | None

r2_jurisdiction: str | None

d1_database_id: str | None

d1_jurisdiction: str | None

platform_domain: str

worker_script_name: str

worker_script_template_name: str | None

internal_api_key: str | None

log_level: str

log_format: str

log_file: str | None

enable_console_logging: bool

enable_custom_hostnames: bool

default_cache_ttl: int

classmethod set_r2_endpoint(v, values)

classmethod from_env() → Config

Load configuration from environment variables.

class Config

Bases: object

validate_assignment = True

model_config = {'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Clients

CloudflareClient

class cloudflare_saas.CloudflareClient(config: Config)

Bases: LoggerMixin

Async Cloudflare API client.

__init__(config: Config)

async create_custom_hostname(hostname: str, verification_method: VerificationMethod = VerificationMethod.HTTP) → Dict[str, Any]

Create custom hostname in Cloudflare for SaaS.

async get_custom_hostname(hostname_id: str) → Dict[str, Any]

Get custom hostname status.

async delete_custom_hostname(hostname_id: str, hostname: str | None = None) → None

Delete custom hostname.

async list_custom_hostnames(hostname: str | None = None) → list

List custom hostnames.

R2Client

class cloudflare_saas.R2Client(config: Config)

Bases: LoggerMixin

Async R2 client for object operations.

__init__(config: Config)

async upload_file(tenant_id: str, local_path: Path, object_key: str, content_type: str | None = None) → str

Upload a file to R2 under tenant namespace.

async upload_directory(tenant_id: str, local_dir: Path, base_prefix: str = '') → List[str]

Upload entire directory to R2.

async delete_object(tenant_id: str, object_key: str) → None

Delete an object from R2.

async delete_tenant_objects(tenant_id: str) → int

Delete all objects for a tenant.

async list_tenant_objects(tenant_id: str, max_keys: int = 1000) → List[dict]

List objects for a tenant.

async object_exists(tenant_id: str, object_key: str) → bool

Check if object exists.

DNSVerifier

class cloudflare_saas.DNSVerifier

Bases: object

Async DNS verification.

__init__()

async verify_cname(domain: str, expected_target: str) → bool

Verify CNAME record points to expected target.

async get_cname_records(domain: str) → List[str]

Get all CNAME records for a domain.

async verify_txt(domain: str, expected_value: str) → bool

Verify TXT record contains expected value.

async wait_for_cname(domain: str, expected_target: str, max_attempts: int = 30, delay_seconds: int = 10) → bool

Poll for CNAME verification with retries.

TerraformDeployer

class cloudflare_saas.TerraformDeployer(config: Config, working_dir: str = './terraform')

Bases: object

Automate Terraform deployment for Workers and R2.

__init__(config: Config, working_dir: str = './terraform')

generate_terraform_config(worker_script_path: Path) → None

Generate Terraform configuration files.

async deploy(worker_script_path: Path, auto_approve: bool = False) → Dict[str, Any]

Deploy infrastructure with Terraform.

async destroy(auto_approve: bool = False) → Dict[str, Any]

Destroy infrastructure.

Storage Adapters

StorageAdapter

class cloudflare_saas.StorageAdapter

Bases: ABC

Abstract base class for storage implementations.

abstractmethod async save_tenant(tenant: Tenant) → None

Save tenant to storage.

abstractmethod async get_tenant(tenant_id: str) → Tenant | None

Retrieve tenant by ID.

abstractmethod async delete_tenant(tenant_id: str) → None

Delete tenant from storage.

abstractmethod async list_tenants(limit: int = 100, offset: int = 0) → List[Tenant]

List all tenants with pagination.

abstractmethod async save_domain(domain: CustomDomain) → None

Save custom domain to storage.

abstractmethod async get_domain(domain: str) → CustomDomain | None

Retrieve custom domain by hostname.

abstractmethod async delete_domain(domain: str) → None

Delete custom domain from storage.

abstractmethod async list_tenant_domains(tenant_id: str) → List[CustomDomain]

List all domains for a tenant.

abstractmethod async get_domain_by_tenant(domain: str) → str | None

Get tenant ID for active domain.

InMemoryStorageAdapter

class cloudflare_saas.InMemoryStorageAdapter

Bases: StorageAdapter

In-memory storage implementation (for testing/development).

__init__()

async save_tenant(tenant: Tenant) → None

Save tenant to storage.

async get_tenant(tenant_id: str) → Tenant | None

Retrieve tenant by ID.

async delete_tenant(tenant_id: str) → None

Delete tenant from storage.

async list_tenants(limit: int = 100, offset: int = 0) → List[Tenant]

List all tenants with pagination.

async save_domain(domain: CustomDomain) → None

Save custom domain to storage.

async get_domain(domain: str) → CustomDomain | None

Retrieve custom domain by hostname.

async delete_domain(domain: str) → None

Delete custom domain from storage.

async list_tenant_domains(tenant_id: str) → List[CustomDomain]

List all domains for a tenant.

async get_domain_by_tenant(domain: str) → str | None

Get tenant ID for active domain.

PostgresStorageAdapter

class cloudflare_saas.PostgresStorageAdapter(connection_string: str)

Bases: StorageAdapter

PostgreSQL storage implementation.

__init__(connection_string: str)

async initialize() → None

Initialize connection pool and create tables.

async close() → None

Close connection pool.

async save_tenant(tenant: Tenant) → None

Save tenant to storage.

async get_tenant(tenant_id: str) → Tenant | None

Retrieve tenant by ID.

async delete_tenant(tenant_id: str) → None

Delete tenant from storage.

async list_tenants(limit: int = 100, offset: int = 0) → List[Tenant]

List all tenants with pagination.

async save_domain(domain: CustomDomain) → None

Save custom domain to storage.

async get_domain(domain: str) → CustomDomain | None

Retrieve custom domain by hostname.

async delete_domain(domain: str) → None

Delete custom domain from storage.

async list_tenant_domains(tenant_id: str) → List[CustomDomain]

List all domains for a tenant.

async get_domain_by_tenant(domain: str) → str | None

Get tenant ID for active domain.

Data Models

Tenant

class cloudflare_saas.Tenant(*, tenant_id: str, name: str, slug: str, subdomain: str, owner_id: str | None = None, created_at: datetime = <factory>, metadata: Dict[str, ~typing.Any]=<factory>)

Bases: BaseModel

Tenant model.

tenant_id: str

name: str

slug: str

subdomain: str

owner_id: str | None

created_at: datetime

metadata: Dict[str, Any]

classmethod set_tenant_id(v, values)

classmethod set_subdomain(v, values)

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

CustomDomain

class cloudflare_saas.CustomDomain(*, domain: str, tenant_id: str, status: DomainStatus = DomainStatus.PENDING, verification_method: VerificationMethod = VerificationMethod.HTTP, verification_token: str | None = None, cname_target: str | None = None, cloudflare_hostname_id: str | None = None, ssl_status: str | None = None, created_at: datetime = <factory>, verified_at: datetime | None = None, error_message: str | None = None)

Bases: BaseModel

Custom domain model.

domain: str

tenant_id: str

status: DomainStatus

verification_method: VerificationMethod

verification_token: str | None

cname_target: str | None

cloudflare_hostname_id: str | None

ssl_status: str | None

created_at: datetime

verified_at: datetime | None

error_message: str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

DomainStatus

class cloudflare_saas.DomainStatus(value)

Bases: str, Enum

Status of custom domain provisioning.

PENDING = 'pending'

VERIFYING = 'verifying'

VERIFIED = 'verified'

ACTIVE = 'active'

FAILED = 'failed'

DeploymentResult

class cloudflare_saas.DeploymentResult(*, tenant_id: str, files_uploaded: int, total_size_bytes: int, deployment_time_seconds: float, success: bool, error_message: str | None = None, uploaded_paths: List[str] = <factory>)

Bases: BaseModel

Result of site deployment.

tenant_id: str

files_uploaded: int

total_size_bytes: int

deployment_time_seconds: float

success: bool

error_message: str | None

uploaded_paths: List[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

VerificationMethod

class cloudflare_saas.VerificationMethod(value)

Bases: str, Enum

DNS verification method.

HTTP = 'http'

TXT = 'txt'

EMAIL = 'email'

HostnameVerificationInstructions

class cloudflare_saas.HostnameVerificationInstructions(*, domain: str, cname_target: str, verification_method: VerificationMethod, http_verification_url: str | None = None, http_verification_token: str | None = None, txt_record_name: str | None = None, txt_record_value: str | None = None, instructions: str)

Bases: BaseModel

Instructions for domain verification.

domain: str

cname_target: str

verification_method: VerificationMethod

http_verification_url: str | None

http_verification_token: str | None

txt_record_name: str | None

txt_record_value: str | None

instructions: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Exceptions

CloudflareSaaSException

class cloudflare_saas.CloudflareSaaSException

Bases: Exception

Base exception for all Cloudflare SaaS operations.

TenantNotFoundError

class cloudflare_saas.TenantNotFoundError

Bases: CloudflareSaaSException

Raised when tenant is not found.

DomainVerificationError

class cloudflare_saas.DomainVerificationError

Bases: CloudflareSaaSException

Raised when domain verification fails.

DeploymentError

class cloudflare_saas.DeploymentError

Bases: CloudflareSaaSException

Raised when site deployment fails.

R2OperationError

class cloudflare_saas.R2OperationError

Bases: CloudflareSaaSException

Raised when R2 operations fail.

CustomHostnameError

class cloudflare_saas.CustomHostnameError

Bases: CloudflareSaaSException

Raised when custom hostname operations fail.

DNSError

class cloudflare_saas.DNSError

Bases: CloudflareSaaSException

Raised when DNS operations fail.

Logging

configure_logging

cloudflare_saas.configure_logging(level: LogLevel = LogLevel.INFO, log_format: LogFormat = LogFormat.DETAILED, log_file: str | None = None, enable_console: bool = True) → None

Configure logging for the application.

Parameters:

• level – The logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)

• log_format – The format of log messages (simple, detailed, json)

• log_file – Optional file path to write logs to

• enable_console – Whether to enable console output

Examples

>>> from cloudflare_saas.logging_config import configure_logging, LogLevel, LogFormat
>>> configure_logging(level=LogLevel.DEBUG, log_format=LogFormat.JSON)
>>> configure_logging(level=LogLevel.INFO, log_file="app.log")

get_logger

cloudflare_saas.get_logger(name: str) → Logger

Get a logger instance.

Parameters:

name – The name of the logger (typically __name__)

Returns:

A configured logger instance

Examples

>>> from cloudflare_saas.logging_config import get_logger
>>> logger = get_logger(__name__)
>>> logger.info("Starting operation")

LogLevel

class cloudflare_saas.LogLevel(value)

Logging levels.

DEBUG = 'DEBUG'

INFO = 'INFO'

WARNING = 'WARNING'

ERROR = 'ERROR'

CRITICAL = 'CRITICAL'

LogFormat

class cloudflare_saas.LogFormat(value)

Log output formats.

SIMPLE = 'simple'

DETAILED = 'detailed'

JSON = 'json'

LoggerMixin

class cloudflare_saas.LoggerMixin

Mixin class to add logging capability to any class.

Examples

>>> class MyService(LoggerMixin):
...     def do_work(self):
...         self.logger.info("Doing work")

property logger: Logger

Get logger for this class.