Admin
Python bindings for database administration functionality.
This module provides Python access to admin operations for managing multi-tenant PostgreSQL deployments.
Rust Implementation: cloacina::python::bindings::admin::PyTenantConfig
Python wrapper for TenantConfig
new(schema_name: str, username: str, password: Optional[str]) -> Self
Rust Implementation: cloacina::python::bindings::admin::PyTenantConfig::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() -> str
Rust Implementation: cloacina::python::bindings::admin::PyTenantConfig::schema_name
Source
pub fn schema_name(&self) -> String {
self.inner.schema_name.clone()
}
username() -> str
Rust Implementation: cloacina::python::bindings::admin::PyTenantConfig::username
Source
pub fn username(&self) -> String {
self.inner.username.clone()
}
password() -> str
Rust Implementation: cloacina::python::bindings::admin::PyTenantConfig::password
Source
pub fn password(&self) -> String {
self.inner.password.clone()
}
__repr__() -> str
Rust Implementation: cloacina::python::bindings::admin::PyTenantConfig::repr
Source
pub fn __repr__(&self) -> String {
format!(
"TenantConfig(schema_name='{}', username='{}', password='***')",
self.inner.schema_name, self.inner.username
)
}
Rust Implementation: cloacina::python::bindings::admin::PyTenantCredentials
Python wrapper for TenantCredentials
username() -> str
Rust Implementation: cloacina::python::bindings::admin::PyTenantCredentials::username
Source
pub fn username(&self) -> String {
self.inner.username.clone()
}
password() -> str
Rust Implementation: cloacina::python::bindings::admin::PyTenantCredentials::password
Source
pub fn password(&self) -> String {
self.inner.password.clone()
}
schema_name() -> str
Rust Implementation: cloacina::python::bindings::admin::PyTenantCredentials::schema_name
Source
pub fn schema_name(&self) -> String {
self.inner.schema_name.clone()
}
connection_string() -> str
Rust Implementation: cloacina::python::bindings::admin::PyTenantCredentials::connection_string
Source
pub fn connection_string(&self) -> String {
self.inner.connection_string.clone()
}
__repr__() -> str
Rust Implementation: cloacina::python::bindings::admin::PyTenantCredentials::repr
Source
pub fn __repr__(&self) -> String {
format!(
"TenantCredentials(username='{}', schema_name='{}', password='***', connection_string='***')",
self.inner.username, self.inner.schema_name
)
}
Rust Implementation: cloacina::python::bindings::admin::PyDatabaseAdmin
Python wrapper for DatabaseAdmin
Note: This class is only functional with PostgreSQL databases. SQLite does not support database schemas or user management.
new(database_url: str) -> Self
Rust Implementation: cloacina::python::bindings::admin::PyDatabaseAdmin::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(config: PyTenantConfig) -> PyTenantCredentials
Rust Implementation: cloacina::python::bindings::admin::PyDatabaseAdmin::create_tenant
Source
pub fn create_tenant(&self, config: &PyTenantConfig) -> PyResult<PyTenantCredentials> {
let rt = tokio::runtime::Runtime::new()
.map_err(|e| PyRuntimeError::new_err(format!("Failed to create runtime: {}", e)))?;
let tenant_config = TenantConfig {
schema_name: config.inner.schema_name.clone(),
username: config.inner.username.clone(),
password: config.inner.password.clone(),
};
let credentials = rt
.block_on(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(schema_name: str, username: str) -> None
Rust Implementation: cloacina::python::bindings::admin::PyDatabaseAdmin::remove_tenant
Source
pub fn remove_tenant(&self, schema_name: String, username: String) -> PyResult<()> {
let rt = tokio::runtime::Runtime::new()
.map_err(|e| PyRuntimeError::new_err(format!("Failed to create runtime: {}", e)))?;
rt.block_on(async { self.inner.remove_tenant(&schema_name, &username).await })
.map_err(|e: AdminError| {
PyRuntimeError::new_err(format!("Failed to remove tenant: {}", e))
})?;
Ok(())
}
__repr__() -> str
Rust Implementation: cloacina::python::bindings::admin::PyDatabaseAdmin::repr
Source
pub fn __repr__(&self) -> String {
"DatabaseAdmin()".to_string()
}