cloacina-python::bindings::admin

Rust

Python bindings for database administration functionality.

This module provides Python access to admin operations for managing multi-tenant PostgreSQL deployments.

Structs

class TenantConfig

Binding

Python API: cloaca.TenantConfig

Python wrapper for TenantConfig

Fields

Name	Type	Description
inner	TenantConfig

Methods

new

fn new(
    schema_name: String,
    username: String,
    password: Option<String>,
) -> Self

Python API: cloaca.TenantConfig.new

Source

    pub fn new(schema_name: String, username: String, password: Option<String>) -> Self {
        Self {
            inner: TenantConfig {
                schema_name,
                username,
                password: password.unwrap_or_default(),
            },
        }
    }

schema_​name

fn schema_name(&self) -> String

Python API: cloaca.TenantConfig.schema_name

Source

    pub fn schema_name(&self) -> String {
        self.inner.schema_name.clone()
    }

username

fn username(&self) -> String

Python API: cloaca.TenantConfig.username

Source

    pub fn username(&self) -> String {
        self.inner.username.clone()
    }

password

fn password(&self) -> String

Python API: cloaca.TenantConfig.password

Source

    pub fn password(&self) -> String {
        self.inner.password.clone()
    }

_​_​repr_​_​

fn __repr__(&self) -> String

Python API: cloaca.TenantConfig.__repr__

Source

    pub fn __repr__(&self) -> String {
        format!(
            "TenantConfig(schema_name='{}', username='{}', password='***')",
            self.inner.schema_name, self.inner.username
        )
    }

class TenantCredentials

Binding

Python API: cloaca.TenantCredentials

Python wrapper for TenantCredentials

Fields

Name	Type	Description
inner	TenantCredentials

Methods

username

fn username(&self) -> String

Python API: cloaca.TenantCredentials.username

Source

    pub fn username(&self) -> String {
        self.inner.username.clone()
    }

password

fn password(&self) -> String

Python API: cloaca.TenantCredentials.password

Source

    pub fn password(&self) -> String {
        self.inner.password.clone()
    }

schema_​name

fn schema_name(&self) -> String

Python API: cloaca.TenantCredentials.schema_name

Source

    pub fn schema_name(&self) -> String {
        self.inner.schema_name.clone()
    }

connection_​string

fn connection_string(&self) -> String

Python API: cloaca.TenantCredentials.connection_string

Source

    pub fn connection_string(&self) -> String {
        self.inner.connection_string.clone()
    }

_​_​repr_​_​

fn __repr__(&self) -> String

Python API: cloaca.TenantCredentials.__repr__

Source

    pub fn __repr__(&self) -> String {
        format!(
            "TenantCredentials(username='{}', schema_name='{}', password='***', connection_string='***')",
            self.inner.username, self.inner.schema_name
        )
    }

class DatabaseAdmin

Binding

Python API: cloaca.DatabaseAdmin

Python wrapper for DatabaseAdmin

Note: This class is only functional with PostgreSQL databases. SQLite does not support database schemas or user management.

Fields

Name	Type	Description
inner	DatabaseAdmin

Methods

new

fn new(database_url: String) -> PyResult<Self>

Python API: cloaca.DatabaseAdmin.new

Source

    pub fn new(database_url: String) -> PyResult<Self> {
        // Runtime check for PostgreSQL
        if !is_postgres_url(&database_url) {
            return Err(PyRuntimeError::new_err(
                "DatabaseAdmin requires a PostgreSQL connection. \
                 SQLite does not support database schemas or user management. \
                 Use a PostgreSQL URL like 'postgres://user:pass@host/db'",
            ));
        }

        // Parse the database URL to extract components
        let url = url::Url::parse(&database_url)
            .map_err(|e| PyRuntimeError::new_err(format!("Invalid database URL: {}", e)))?;

        let database_name = url.path().trim_start_matches('/');
        if database_name.is_empty() {
            return Err(PyRuntimeError::new_err(
                "Database name is required in URL path",
            ));
        }

        // Build connection string with all components
        let username = url.username();
        let password = url.password().unwrap_or("");
        let host = url.host_str().unwrap_or("localhost");
        let port = url.port().unwrap_or(5432);

        let connection_string = if password.is_empty() {
            format!("{}://{}@{}:{}", url.scheme(), username, host, port)
        } else {
            format!(
                "{}://{}:{}@{}:{}",
                url.scheme(),
                username,
                password,
                host,
                port
            )
        };

        let database = Database::new(&connection_string, database_name, 10);
        let admin = DatabaseAdmin::new(database);
        Ok(Self { inner: admin })
    }

create_​tenant

fn create_tenant(
    &self,
    py: Python<'_>,
    config: &PyTenantConfig,
) -> PyResult<PyTenantCredentials>

Python API: cloaca.DatabaseAdmin.create_tenant

Source

    pub fn create_tenant(
        &self,
        py: Python<'_>,
        config: &PyTenantConfig,
    ) -> PyResult<PyTenantCredentials> {
        let tenant_config = TenantConfig {
            schema_name: config.inner.schema_name.clone(),
            username: config.inner.username.clone(),
            password: config.inner.password.clone(),
        };

        // CLOACI-I-0136: drive the async admin call through the one GIL-safe
        // bridge (releases the GIL before blocking) — previously this held the
        // GIL across the await on a fresh runtime.
        let credentials =
            crate::gil::py_block_on(py, async { self.inner.create_tenant(tenant_config).await })
                .map_err(|e: AdminError| {
                    PyRuntimeError::new_err(format!("Failed to create tenant: {}", e))
                })?;

        Ok(PyTenantCredentials { inner: credentials })
    }

remove_​tenant

fn remove_tenant(
    &self,
    py: Python<'_>,
    schema_name: String,
    username: String,
) -> PyResult<()>

Python API: cloaca.DatabaseAdmin.remove_tenant

Source

    pub fn remove_tenant(
        &self,
        py: Python<'_>,
        schema_name: String,
        username: String,
    ) -> PyResult<()> {
        // CLOACI-I-0136: GIL-safe bridge (see create_tenant).
        crate::gil::py_block_on(py, async {
            self.inner.remove_tenant(&schema_name, &username).await
        })
        .map_err(|e: AdminError| {
            PyRuntimeError::new_err(format!("Failed to remove tenant: {}", e))
        })?;

        Ok(())
    }

_​_​repr_​_​

fn __repr__(&self) -> String

Python API: cloaca.DatabaseAdmin.__repr__

Source

    pub fn __repr__(&self) -> String {
        "DatabaseAdmin()".to_string()
    }