Microsoft Azure RDBMS Management Client Library for Python providing management capabilities for MySQL, MariaDB, and PostgreSQL databases
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Shared types, enums, exceptions, and patterns used across all Azure RDBMS services (MySQL, PostgreSQL, MariaDB) in both single server and flexible server deployments.
Common base models used across all RDBMS services for consistent resource representation.
class Resource:
"""Base Azure resource representation."""
id: str
name: str
type: str
class TrackedResource(Resource):
"""Tracked Azure resource with location and tags."""
location: str
tags: Dict[str, str]
class ProxyResource(Resource):
"""Proxy resource without location (child resources)."""
pass
class SystemData:
"""System metadata for Azure resources."""
created_by: str
created_by_type: str # User, Application, ManagedIdentity, Key
created_at: datetime
last_modified_by: str
last_modified_by_type: str
last_modified_at: datetimeCommon patterns for server creation across all RDBMS services.
class ServerForCreate:
"""Base server creation model."""
location: str
tags: Dict[str, str]
properties: ServerPropertiesForCreate
class ServerPropertiesForCreate:
"""Base server properties for creation."""
pass
class ServerPropertiesForDefaultCreate(ServerPropertiesForCreate):
"""Server properties for default creation mode."""
administrator_login: str
administrator_login_password: str
version: str
ssl_enforcement: SslEnforcementEnum
minimal_tls_version: MinimalTlsVersionEnum
infrastructure_encryption: InfrastructureEncryption
public_network_access: PublicNetworkAccessEnum
storage_profile: StorageProfile
class ServerPropertiesForRestore(ServerPropertiesForCreate):
"""Server properties for restore creation mode."""
source_server_id: str
restore_point_in_time: datetime
class ServerPropertiesForGeoRestore(ServerPropertiesForCreate):
"""Server properties for geo-restore creation mode."""
source_server_id: str
class ServerPropertiesForReplica(ServerPropertiesForCreate):
"""Server properties for replica creation mode."""
source_server_id: strclass SslEnforcementEnum(str, Enum):
"""SSL enforcement options for database servers."""
ENABLED = "Enabled"
DISABLED = "Disabled"
class MinimalTlsVersionEnum(str, Enum):
"""Minimum TLS version requirements."""
TLS1_0 = "TLS1_0"
TLS1_1 = "TLS1_1"
TLS1_2 = "TLS1_2"
TLS_ENFORCEMENT_DISABLED = "TLSEnforcementDisabled"class ServerState(str, Enum):
"""Server operational states."""
READY = "Ready"
DROPPING = "Dropping"
DISABLED = "Disabled"
STARTING = "Starting"
STOPPING = "Stopping"
STOPPED = "Stopped"
UPDATING = "Updating"
INACCESSIBLE = "Inaccessible"
class ServerVersion(str, Enum):
"""Database server versions."""
# MySQL versions
MYSQL_5_6 = "5.6"
MYSQL_5_7 = "5.7"
MYSQL_8_0 = "8.0"
# PostgreSQL versions
POSTGRESQL_9_5 = "9.5"
POSTGRESQL_9_6 = "9.6"
POSTGRESQL_10 = "10"
POSTGRESQL_11 = "11"
POSTGRESQL_12 = "12"
POSTGRESQL_13 = "13"
POSTGRESQL_14 = "14"
POSTGRESQL_15 = "15"
# MariaDB versions
MARIADB_10_2 = "10.2"
MARIADB_10_3 = "10.3"
class PublicNetworkAccessEnum(str, Enum):
"""Public network access configuration."""
ENABLED = "Enabled"
DISABLED = "Disabled"
class InfrastructureEncryption(str, Enum):
"""Infrastructure encryption options."""
ENABLED = "Enabled"
DISABLED = "Disabled"class StorageAutogrow(str, Enum):
"""Storage auto-grow options."""
ENABLED = "Enabled"
DISABLED = "Disabled"
class GeoRedundantBackup(str, Enum):
"""Geo-redundant backup options."""
ENABLED = "Enabled"
DISABLED = "Disabled"
class BackupRetentionDays:
"""Backup retention validation."""
MIN_DAYS = 7
MAX_DAYS = 35Common storage models used across single server deployments.
class StorageProfile:
"""Storage configuration for database servers."""
backup_retention_days: int # 7-35 days
geo_redundant_backup: GeoRedundantBackup
storage_mb: int
storage_autogrow: StorageAutogrow
class StorageMB:
"""Storage size validation and common values."""
MIN_VALUE = 5120 # 5 GB
MAX_VALUE = 4194304 # 4 TB
# Common storage sizes
BASIC_5GB = 5120
BASIC_10GB = 10240
GENERAL_PURPOSE_100GB = 102400
GENERAL_PURPOSE_1TB = 1048576
MEMORY_OPTIMIZED_1TB = 1048576class PerformanceTierProperties:
"""Performance tier definition."""
id: str
max_backup_retention_days: int
min_backup_retention_days: int
max_storage_mb: int
min_large_storage_mb: int
max_large_storage_mb: int
min_storage_mb: int
class ServiceObjective:
"""Service objective for performance tiers."""
service_objective_name: str
id: str
unit: str # DTU, vCore
last_observed_time: datetime
metric_name: str
metric_display_name: str
metric_current_value: float
metric_limit: float
metric_unit: strclass FirewallRule(ProxyResource):
"""IP-based firewall rule for database servers."""
start_ip_address: str
end_ip_address: str
system_data: SystemData
class VirtualNetworkRule(ProxyResource):
"""Virtual network rule for database servers."""
virtual_network_subnet_id: str
ignore_missing_vnet_service_endpoint: bool
state: VirtualNetworkRuleState
class VirtualNetworkRuleState(str, Enum):
"""Virtual network rule states."""
INITIALIZING = "Initializing"
IN_PROGRESS = "InProgress"
READY = "Ready"
DELETING = "Deleting"
UNKNOWN = "Unknown"class ServerAdministratorResource(ProxyResource):
"""Azure Active Directory administrator."""
administrator_type: AdministratorType
login: str
sid: str
tenant_id: str
class AdministratorType(str, Enum):
"""Administrator types."""
ACTIVE_DIRECTORY = "ActiveDirectory"
class ServerKey(ProxyResource):
"""Customer-managed encryption key."""
kind: str
server_key_type: ServerKeyType
uri: str
creation_date: datetime
class ServerKeyType(str, Enum):
"""Server key types."""
AZURE_KEY_VAULT = "AzureKeyVault"class Configuration(ProxyResource):
"""Database server configuration parameter."""
value: str
description: str
default_value: str
data_type: str
allowed_values: str
source: str
is_dynamic: bool
is_read_only: bool
is_config_pending_restart: bool
system_data: SystemData
class ConfigurationSource(str, Enum):
"""Configuration parameter sources."""
SYSTEM_DEFAULT = "system-default"
USER_OVERRIDE = "user-override"class ErrorResponse:
"""Standard error response format."""
error: ErrorDefinition
class ErrorDefinition:
"""Error definition details."""
code: str
message: str
target: str
details: List[ErrorDefinition]
class CloudError(Exception):
"""Base cloud error for RDBMS operations."""
error: ErrorResponse
class ResourceNotFoundError(CloudError):
"""Resource not found error."""
pass
class BadRequestError(CloudError):
"""Bad request error."""
pass
class ConflictError(CloudError):
"""Resource conflict error."""
pass
class InternalServerError(CloudError):
"""Internal server error."""
passclass LROPoller:
"""Long running operation poller."""
def result(self, timeout: int = None):
"""Wait for operation completion and return result."""
pass
def status(self) -> str:
"""Get current operation status."""
pass
def done(self) -> bool:
"""Check if operation is completed."""
pass
def wait(self, timeout: int = None):
"""Wait for operation completion without returning result."""
pass
class OperationStatus(str, Enum):
"""Operation status values."""
IN_PROGRESS = "InProgress"
SUCCEEDED = "Succeeded"
FAILED = "Failed"
CANCELED = "Canceled"class PagedResult:
"""Paged result collection."""
value: List[Any]
next_link: str
class ItemPaged:
"""Iterable paged collection."""
def __iter__(self):
"""Iterate through all items across pages."""
pass
def by_page(self):
"""Iterate page by page."""
passclass NameAvailabilityRequest:
"""Name availability check request."""
name: str
type: str # Microsoft.DBforMySQL/servers, Microsoft.DBforPostgreSQL/servers
class NameAvailability:
"""Name availability check result."""
name_available: bool
reason: UnavailabilityReason
message: str
class UnavailabilityReason(str, Enum):
"""Reasons for name unavailability."""
INVALID = "Invalid"
ALREADY_EXISTS = "AlreadyExists"class FlexibleServerStorage:
"""Enhanced storage for flexible servers."""
storage_size_gb: int
iops: int
auto_grow: StorageAutogrow
storage_sku: str # Premium_LRS, Standard_LRS
tier: str # P4, P6, P10, etc.
class FlexibleServerBackup:
"""Enhanced backup configuration for flexible servers."""
backup_retention_days: int
geo_redundant_backup: GeoRedundantBackup
earliest_restore_date: datetime
class HighAvailability:
"""High availability configuration for flexible servers."""
mode: HighAvailabilityMode
state: HighAvailabilityState
standby_availability_zone: str
class HighAvailabilityMode(str, Enum):
"""High availability modes."""
DISABLED = "Disabled"
ZONE_REDUNDANT = "ZoneRedundant"
SAME_ZONE = "SameZone"
class HighAvailabilityState(str, Enum):
"""High availability states."""
NOT_ENABLED = "NotEnabled"
CREATING_STANDBY = "CreatingStandby"
REPLICATING_DATA = "ReplicatingData"
FAILING_OVER = "FailingOver"
HEALTHY = "Healthy"
REMOVING_STANDBY = "RemovingStandby"class Network:
"""Network configuration for flexible servers."""
public_network_access: PublicNetworkAccessEnum
delegated_subnet_resource_id: str
private_dns_zone_resource_id: str
class MaintenanceWindow:
"""Maintenance window configuration."""
custom_window: str # Enabled, Disabled
start_hour: int # 0-23
start_minute: int # 0, 30
day_of_week: int # 0-6 (Sunday-Saturday)from azure.identity import DefaultAzureCredential, ClientSecretCredential
# Using default credential chain
credential = DefaultAzureCredential()
# Using service principal
credential = ClientSecretCredential(
tenant_id="tenant-id",
client_id="client-id",
client_secret="client-secret"
)from azure.core.exceptions import HttpResponseError, ResourceNotFoundError
try:
server = client.servers.get("resource-group", "server-name")
except ResourceNotFoundError:
print("Server not found")
except HttpResponseError as e:
print(f"HTTP error: {e.status_code} - {e.message}")
except Exception as e:
print(f"Unexpected error: {e}")# Async operation with polling
operation = client.servers.begin_create("rg", "server", parameters)
# Wait for completion
try:
server = operation.result(timeout=1800) # 30 minutes
print(f"Server created: {server.name}")
except Exception as e:
print(f"Operation failed: {e}")
# Check status without waiting
if operation.done():
result = operation.result()
else:
print(f"Operation status: {operation.status()}")# Iterate through all items
for server in client.servers.list():
print(f"Server: {server.name}")
# Page-by-page iteration
paged_result = client.servers.list()
for page in paged_result.by_page():
for server in page:
print(f"Server: {server.name}")Install with Tessl CLI
npx tessl i tessl/pypi-azure-mgmt-rdbms