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
Complete management capabilities for MySQL single server deployments on Azure. This service provides server lifecycle management, database operations, security configuration, performance monitoring, and advisory recommendations.
Main client for MySQL single server operations with comprehensive resource management capabilities.
class MySQLManagementClient:
"""
MySQL management client for single server deployments.
Parameters:
- credential: TokenCredential for Azure authentication
- subscription_id: str, Azure subscription ID
- base_url: str, management endpoint (default: https://management.azure.com)
- polling_interval: int, LRO polling interval in seconds
Attributes:
- servers: Server lifecycle operations
- databases: Database management operations
- firewall_rules: Firewall configuration operations
- virtual_network_rules: Virtual network rule operations
- configurations: Server parameter operations
- server_parameters: Server parameter operations (legacy)
- log_files: Log file access operations
- server_administrators: Azure AD administrator operations
- recoverable_servers: Disaster recovery operations
- replicas: Read replica operations
- server_based_performance_tier: Performance tier operations (server-specific)
- location_based_performance_tier: Performance tier operations (location-based)
- check_name_availability: Name availability validation
- operations: Available API operations listing
- server_security_alert_policies: Threat detection policies
- query_texts: Performance query text operations
- top_query_statistics: Query performance statistics
- wait_statistics: Wait event statistics
- advisors: Performance advisory operations
- recommended_actions: Performance recommendations
- location_based_recommended_action_sessions_operation_status: Recommendation session status
- location_based_recommended_action_sessions_result: Recommendation session results
- private_endpoint_connections: Private endpoint connections
- private_link_resources: Private link resource operations
- server_keys: Customer-managed encryption keys
"""
def __init__(
self,
credential,
subscription_id: str,
base_url: str = "https://management.azure.com",
**kwargs
): ...
def close(self): ...
def __enter__(self): ...
def __exit__(self, *exc_details): ...Core server lifecycle management including creation, updates, deletion, restart, and upgrade operations.
class ServersOperations:
"""Server lifecycle and management operations."""
def begin_create(self, resource_group_name: str, server_name: str, parameters, **kwargs):
"""
Creates a new MySQL server or updates an existing server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name (must be globally unique)
- parameters: ServerForCreate, server creation parameters
Returns:
LROPoller[Server] for server creation operation
"""
def begin_update(self, resource_group_name: str, server_name: str, parameters, **kwargs):
"""
Updates an existing MySQL server configuration.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- parameters: ServerUpdateParameters, server update parameters
Returns:
LROPoller[Server] for server update operation
"""
def begin_delete(self, resource_group_name: str, server_name: str, **kwargs):
"""
Deletes a MySQL server and all its resources.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
Returns:
LROPoller[None] for server deletion operation
"""
def begin_restart(self, resource_group_name: str, server_name: str, **kwargs):
"""
Restarts a MySQL server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
Returns:
LROPoller[None] for server restart operation
"""
def begin_start(self, resource_group_name: str, server_name: str, **kwargs):
"""
Starts a stopped MySQL server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
Returns:
LROPoller[None] for server start operation
"""
def begin_stop(self, resource_group_name: str, server_name: str, **kwargs):
"""
Stops a running MySQL server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
Returns:
LROPoller[None] for server stop operation
"""
def begin_upgrade(self, resource_group_name: str, server_name: str, parameters, **kwargs):
"""
Upgrades a MySQL server version.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- parameters: ServerUpgradeParameters, upgrade parameters
Returns:
LROPoller[None] for server upgrade operation
"""
def get(self, resource_group_name: str, server_name: str, **kwargs):
"""
Gets details of a MySQL server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
Returns:
Server object with current properties
"""
def list(self, **kwargs):
"""
Lists all MySQL servers in the subscription.
Returns:
Paged collection of Server objects
"""
def list_by_resource_group(self, resource_group_name: str, **kwargs):
"""
Lists MySQL servers in a resource group.
Parameters:
- resource_group_name: str, resource group name
Returns:
Paged collection of Server objects
"""Database management within MySQL servers including creation, deletion, and property queries.
class DatabasesOperations:
"""Database management operations within MySQL servers."""
def begin_create_or_update(self, resource_group_name: str, server_name: str, database_name: str, parameters, **kwargs):
"""
Creates or updates a database within a MySQL server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- database_name: str, database name
- parameters: Database, database parameters (charset, collation)
Returns:
LROPoller[Database] for database operation
"""
def begin_delete(self, resource_group_name: str, server_name: str, database_name: str, **kwargs):
"""
Deletes a database from a MySQL server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- database_name: str, database name
Returns:
LROPoller[None] for deletion operation
"""
def get(self, resource_group_name: str, server_name: str, database_name: str, **kwargs):
"""
Gets database properties.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- database_name: str, database name
Returns:
Database object with properties
"""
def list_by_server(self, resource_group_name: str, server_name: str, **kwargs):
"""
Lists databases in a MySQL server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
Returns:
Paged collection of Database objects
"""IP-based access control configuration for MySQL servers.
class FirewallRulesOperations:
"""Firewall rule management for MySQL servers."""
def create_or_update(self, resource_group_name: str, server_name: str, firewall_rule_name: str, parameters, **kwargs):
"""
Creates or updates a firewall rule.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- firewall_rule_name: str, firewall rule name
- parameters: Firewall rule parameters (start_ip_address, end_ip_address)
Returns:
LROPoller for firewall rule operation
"""
def delete(self, resource_group_name: str, server_name: str, firewall_rule_name: str, **kwargs):
"""
Deletes a firewall rule.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- firewall_rule_name: str, firewall rule name
Returns:
LROPoller for deletion operation
"""
def get(self, resource_group_name: str, server_name: str, firewall_rule_name: str, **kwargs):
"""
Gets a firewall rule.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- firewall_rule_name: str, firewall rule name
Returns:
FirewallRule object
"""
def list_by_server(self, resource_group_name: str, server_name: str, **kwargs):
"""
Lists firewall rules for a MySQL server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
Returns:
Paged collection of FirewallRule objects
"""VNet-based access control configuration for MySQL servers.
class VirtualNetworkRulesOperations:
"""Virtual network rule management for MySQL servers."""
def create_or_update(self, resource_group_name: str, server_name: str, virtual_network_rule_name: str, parameters, **kwargs):
"""
Creates or updates a virtual network rule.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- virtual_network_rule_name: str, virtual network rule name
- parameters: VNet rule parameters (virtual_network_subnet_id, ignore_missing_vnet_service_endpoint)
Returns:
LROPoller for virtual network rule operation
"""
def delete(self, resource_group_name: str, server_name: str, virtual_network_rule_name: str, **kwargs):
"""
Deletes a virtual network rule.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- virtual_network_rule_name: str, virtual network rule name
Returns:
LROPoller for deletion operation
"""
def get(self, resource_group_name: str, server_name: str, virtual_network_rule_name: str, **kwargs):
"""
Gets a virtual network rule.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- virtual_network_rule_name: str, virtual network rule name
Returns:
VirtualNetworkRule object
"""
def list_by_server(self, resource_group_name: str, server_name: str, **kwargs):
"""
Lists virtual network rules for a MySQL server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
Returns:
Paged collection of VirtualNetworkRule objects
"""Server parameter and configuration management for MySQL instances.
class ConfigurationsOperations:
"""MySQL server configuration and parameter management."""
def create_or_update(self, resource_group_name: str, server_name: str, configuration_name: str, parameters, **kwargs):
"""
Updates a configuration parameter.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- configuration_name: str, parameter name
- parameters: Configuration value and properties
Returns:
LROPoller for configuration update
"""
def get(self, resource_group_name: str, server_name: str, configuration_name: str, **kwargs):
"""
Gets a configuration parameter value.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- configuration_name: str, parameter name
Returns:
Configuration object with current value
"""
def list_by_server(self, resource_group_name: str, server_name: str, **kwargs):
"""
Lists all configuration parameters for a server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
Returns:
Paged collection of Configuration objects
"""MySQL-specific performance insight capabilities including query analysis and performance recommendations.
class QueryTextsOperations:
"""Query text retrieval for performance analysis."""
def get(self, resource_group_name: str, server_name: str, query_id: str, **kwargs):
"""
Gets query text by query ID.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- query_id: str, query identifier
Returns:
QueryText object with query details
"""
def list_by_server(self, resource_group_name: str, server_name: str, query_ids: List[str], **kwargs):
"""
Lists query texts for specified query IDs.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- query_ids: List[str], query identifiers
Returns:
Paged collection of QueryText objects
"""
class TopQueryStatisticsOperations:
"""Top query performance statistics operations."""
def get(self, resource_group_name: str, server_name: str, query_statistic_id: str, **kwargs):
"""
Gets query performance statistics.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- query_statistic_id: str, statistic identifier
Returns:
QueryStatistic object with performance metrics
"""
def list_by_server(self, resource_group_name: str, server_name: str, parameters, **kwargs):
"""
Lists top query statistics for a server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- parameters: Query parameters (aggregation window, observed metrics)
Returns:
Paged collection of QueryStatistic objects
"""
class AdvisorsOperations:
"""Performance advisor operations for MySQL servers."""
def get(self, resource_group_name: str, server_name: str, advisor_name: str, **kwargs):
"""
Gets performance advisor details.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- advisor_name: str, advisor name
Returns:
Advisor object with recommendations
"""
def list_by_server(self, resource_group_name: str, server_name: str, **kwargs):
"""
Lists performance advisors for a server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
Returns:
Paged collection of Advisor objects
"""
class RecommendedActionsOperations:
"""Recommended action operations for MySQL servers."""
def get(self, resource_group_name: str, server_name: str, advisor_name: str, recommended_action_name: str, **kwargs):
"""
Gets a recommended action for performance improvement.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- advisor_name: str, advisor name
- recommended_action_name: str, recommended action name
Returns:
RecommendedAction object with implementation details
"""
def list_by_server(self, resource_group_name: str, server_name: str, advisor_name: str, **kwargs):
"""
Lists recommended actions for a specific advisor.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- advisor_name: str, advisor name
Returns:
Paged collection of RecommendedAction objects
"""
class PrivateEndpointConnectionsOperations:
"""Private endpoint connection operations for MySQL servers."""
def get(self, resource_group_name: str, server_name: str, private_endpoint_connection_name: str, **kwargs):
"""
Gets a private endpoint connection.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- private_endpoint_connection_name: str, connection name
Returns:
PrivateEndpointConnection object
"""
def create_or_update(self, resource_group_name: str, server_name: str, private_endpoint_connection_name: str, parameters, **kwargs):
"""
Creates or updates a private endpoint connection.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- private_endpoint_connection_name: str, connection name
- parameters: Connection parameters including private link service connection state
Returns:
LROPoller for private endpoint connection operation
"""
def delete(self, resource_group_name: str, server_name: str, private_endpoint_connection_name: str, **kwargs):
"""
Deletes a private endpoint connection.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- private_endpoint_connection_name: str, connection name
Returns:
LROPoller for deletion operation
"""
def list_by_server(self, resource_group_name: str, server_name: str, **kwargs):
"""
Lists private endpoint connections for a server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
Returns:
Paged collection of PrivateEndpointConnection objects
"""
class ServerKeysOperations:
"""Customer-managed encryption key operations for MySQL servers."""
def get(self, resource_group_name: str, server_name: str, key_name: str, **kwargs):
"""
Gets a server key used for encryption.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- key_name: str, key name
Returns:
ServerKey object with key details
"""
def create_or_update(self, resource_group_name: str, server_name: str, key_name: str, parameters, **kwargs):
"""
Creates or updates a server key for encryption.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- key_name: str, key name
- parameters: Key parameters including server key type and URI
Returns:
LROPoller for server key operation
"""
def delete(self, resource_group_name: str, server_name: str, key_name: str, **kwargs):
"""
Deletes a server key.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
- key_name: str, key name
Returns:
LROPoller for deletion operation
"""
def list(self, resource_group_name: str, server_name: str, **kwargs):
"""
Lists server keys for a MySQL server.
Parameters:
- resource_group_name: str, resource group name
- server_name: str, server name
Returns:
Paged collection of ServerKey objects
"""class Server:
"""MySQL server resource representation."""
id: str
name: str
type: str
location: str
tags: Dict[str, str]
administrator_login: str
version: str # MySQL version (5.6, 5.7, 8.0)
ssl_enforcement: str # Enabled, Disabled
user_visible_state: str # Ready, Dropping, Disabled, etc.
fully_qualified_domain_name: str
earliest_restore_date: datetime
storage_profile: StorageProfile
replication_role: str # None, Master, Replica
master_server_id: str
replica_capacity: int
class Database:
"""Database within MySQL server."""
id: str
name: str
type: str
charset: str
collation: str
class FirewallRule:
"""IP-based access rule for MySQL server."""
id: str
name: str
type: str
start_ip_address: str
end_ip_address: str
class Configuration:
"""MySQL server configuration parameter."""
id: str
name: str
type: str
value: str
description: str
default_value: str
data_type: str
allowed_values: str
source: str
class StorageProfile:
"""Storage configuration for MySQL server."""
backup_retention_days: int
geo_redundant_backup: str # Enabled, Disabled
storage_mb: int
storage_autogrow: str # Enabled, Disabled
class VirtualNetworkRule:
"""VNet-based access rule for MySQL server."""
id: str
name: str
type: str
virtual_network_subnet_id: str
ignore_missing_vnet_service_endpoint: bool
state: str # Ready, Deleting, Unknown, etc.
class PrivateEndpointConnection:
"""Private endpoint connection for MySQL server."""
id: str
name: str
type: str
private_endpoint: PrivateEndpointProperty
private_link_service_connection_state: PrivateLinkServiceConnectionStateProperty
provisioning_state: str
class ServerKey:
"""Customer-managed encryption key for MySQL server."""
id: str
name: str
type: str
kind: str
server_key_type: str # AzureKeyVault
uri: str
creation_date: datetime
class QueryText:
"""Query text for performance analysis."""
id: str
name: str
type: str
query_id: str
query_text: str
class QueryStatistic:
"""Query performance statistics."""
id: str
name: str
type: str
query_id: str
start_time: datetime
end_time: datetime
aggregation_function: str
database_names: List[str]
query_execution_count: int
metric_name: str
metric_display_name: str
metric_value: float
metric_value_unit: str
class Advisor:
"""Performance advisor for MySQL server."""
id: str
name: str
type: str
properties: Dict[str, Any]
class RecommendedAction:
"""Performance recommendation for MySQL server."""
id: str
name: str
type: str
advisor_name: str
session_id: str
action_id: int
created_time: datetime
expiration_time: datetime
reason: str
recommendation_type: str
details: Dict[str, Any]from azure.identity import DefaultAzureCredential
from azure.mgmt.rdbms.mysql import MySQLManagementClient
from azure.mgmt.rdbms.mysql.models import ServerForCreate, ServerPropertiesForDefaultCreate, StorageProfile
credential = DefaultAzureCredential()
client = MySQLManagementClient(credential, "subscription-id")
# Define server parameters
server_creation_parameters = ServerForCreate(
properties=ServerPropertiesForDefaultCreate(
administrator_login="myadmin",
administrator_login_password="SecurePassword123!",
version="8.0",
ssl_enforcement="Enabled",
storage_profile=StorageProfile(
backup_retention_days=7,
geo_redundant_backup="Disabled",
storage_mb=5120,
storage_autogrow="Enabled"
)
),
location="East US",
tags={"environment": "production"}
)
# Create server (async operation)
server_operation = client.servers.begin_create(
"my-resource-group",
"my-mysql-server",
server_creation_parameters
)
server = server_operation.result() # Wait for completion
print(f"Server created: {server.name}")# Create a database
database_params = {"charset": "utf8mb4", "collation": "utf8mb4_general_ci"}
db_operation = client.databases.begin_create_or_update(
"my-resource-group",
"my-mysql-server",
"my-application-db",
database_params
)
database = db_operation.result()
# Add firewall rule for office network
firewall_params = {
"start_ip_address": "203.0.113.0",
"end_ip_address": "203.0.113.255"
}
fw_operation = client.firewall_rules.begin_create_or_update(
"my-resource-group",
"my-mysql-server",
"office-network",
firewall_params
)
firewall_rule = fw_operation.result()
# Configure server parameter
config_params = {"value": "3600"}
config_operation = client.configurations.begin_create_or_update(
"my-resource-group",
"my-mysql-server",
"wait_timeout",
config_params
)
configuration = config_operation.result()Install with Tessl CLI
npx tessl i tessl/pypi-azure-mgmt-rdbms