Microsoft Azure Batch Management Client Library for Python
npx @tessl/cli install tessl/pypi-azure-mgmt-batch@18.0.0A comprehensive Python library for managing Azure Batch resources through the Azure Resource Manager APIs. This client library provides programmatic access to create, configure, monitor, and manage Azure Batch accounts, pools, applications, certificates, and related resources.
pip install azure-mgmt-batch azure-identityfrom azure.mgmt.batch import BatchManagementClient
from azure.identity import DefaultAzureCredential
from typing import Union, Optional, IOimport os
from azure.identity import DefaultAzureCredential
from azure.mgmt.batch import BatchManagementClient
# Set up authentication
credential = DefaultAzureCredential()
subscription_id = os.getenv("AZURE_SUBSCRIPTION_ID")
# Create the client
client = BatchManagementClient(credential=credential, subscription_id=subscription_id)
# List all Batch accounts in a resource group
accounts = client.batch_account.list_by_resource_group("my-resource-group")
for account in accounts:
print(f"Account: {account.name}, Location: {account.location}")
# Get details of a specific Batch account
account = client.batch_account.get("my-resource-group", "my-batch-account")
print(f"Account ID: {account.id}")
print(f"Provisioning State: {account.provisioning_state}")
# Create a new application
app_params = {
"display_name": "My Application",
"allow_updates": True
}
application = client.application.create(
"my-resource-group",
"my-batch-account",
"my-app",
app_params
)Authentication requires Azure Active Directory credentials. The recommended approach uses environment variables:
AZURE_CLIENT_ID: Azure client IDAZURE_TENANT_ID: Azure tenant IDAZURE_CLIENT_SECRET: Azure client secretAZURE_SUBSCRIPTION_ID: Azure subscription IDThe BatchManagementClient provides access to Azure Batch management operations through specialized operation classes:
Complete lifecycle management of Azure Batch accounts including creation, configuration, key management, and network security settings.
def begin_create(resource_group_name: str, account_name: str, parameters: Union[BatchAccountCreateParameters, IO[bytes]], **kwargs) -> LROPoller[BatchAccount]
def get(resource_group_name: str, account_name: str, **kwargs) -> BatchAccount
def update(resource_group_name: str, account_name: str, parameters: Union[BatchAccountUpdateParameters, IO[bytes]], **kwargs) -> BatchAccount
def begin_delete(resource_group_name: str, account_name: str, **kwargs) -> LROPoller[None]
def list(subscription_id: str, **kwargs) -> ItemPaged[BatchAccount]
def list_by_resource_group(resource_group_name: str, **kwargs) -> ItemPaged[BatchAccount]
def get_keys(resource_group_name: str, account_name: str, **kwargs) -> BatchAccountKeys
def regenerate_key(resource_group_name: str, account_name: str, parameters: Union[BatchAccountRegenerateKeyParameters, IO[bytes]], **kwargs) -> BatchAccountKeys
def synchronize_auto_storage_keys(resource_group_name: str, account_name: str, **kwargs) -> None
def list_detectors(resource_group_name: str, account_name: str, **kwargs) -> ItemPaged[DetectorResponse]
def get_detector(resource_group_name: str, account_name: str, detector_id: str, **kwargs) -> DetectorResponse
def list_outbound_network_dependencies_endpoints(resource_group_name: str, account_name: str, **kwargs) -> ItemPaged[OutboundEnvironmentEndpoint]Creation and management of compute pools for batch workloads, including virtual machine configuration, scaling policies, and network settings.
def create(resource_group_name: str, account_name: str, pool_name: str, parameters: Union[Pool, IO[bytes]], if_match: Optional[str] = None, if_none_match: Optional[str] = None, **kwargs) -> Pool
def get(resource_group_name: str, account_name: str, pool_name: str, **kwargs) -> Pool
def update(resource_group_name: str, account_name: str, pool_name: str, parameters: Union[Pool, IO[bytes]], if_match: Optional[str] = None, **kwargs) -> Pool
def begin_delete(resource_group_name: str, account_name: str, pool_name: str, **kwargs) -> LROPoller[None]
def list_by_batch_account(resource_group_name: str, account_name: str, maxresults: Optional[int] = None, select: Optional[str] = None, filter: Optional[str] = None, **kwargs) -> ItemPaged[Pool]
def disable_auto_scale(resource_group_name: str, account_name: str, pool_name: str, **kwargs) -> Pool
def stop_resize(resource_group_name: str, account_name: str, pool_name: str, **kwargs) -> PoolDeployment and version management for applications that run on Batch compute nodes, including application packages.
# Application Operations
def create(resource_group_name: str, account_name: str, application_name: str, parameters: Optional[Union[Application, IO[bytes]]] = None, **kwargs) -> Application
def get(resource_group_name: str, account_name: str, application_name: str, **kwargs) -> Application
def update(resource_group_name: str, account_name: str, application_name: str, parameters: Union[Application, IO[bytes]], **kwargs) -> Application
def delete(resource_group_name: str, account_name: str, application_name: str, **kwargs) -> None
def list(resource_group_name: str, account_name: str, maxresults: Optional[int] = None, **kwargs) -> ItemPaged[Application]
# Application Package Operations
def create_package(resource_group_name: str, account_name: str, application_name: str, version_name: str, parameters: Optional[Union[ApplicationPackage, IO[bytes]]] = None, **kwargs) -> ApplicationPackage
def activate_package(resource_group_name: str, account_name: str, application_name: str, version_name: str, parameters: Union[ActivateApplicationPackageParameters, IO[bytes]], **kwargs) -> ApplicationPackage
def get_package(resource_group_name: str, account_name: str, application_name: str, version_name: str, **kwargs) -> ApplicationPackage
def delete_package(resource_group_name: str, account_name: str, application_name: str, version_name: str, **kwargs) -> None
def list_packages(resource_group_name: str, account_name: str, application_name: str, maxresults: Optional[int] = None, **kwargs) -> ItemPaged[ApplicationPackage]Management of security certificates for secure communication and authentication in Batch environments.
def list_by_batch_account(resource_group_name: str, account_name: str, maxresults: Optional[int] = None, select: Optional[str] = None, filter: Optional[str] = None, **kwargs) -> ItemPaged[Certificate]
def create(resource_group_name: str, account_name: str, certificate_name: str, parameters: Union[CertificateCreateOrUpdateParameters, IO[bytes]], if_match: Optional[str] = None, if_none_match: Optional[str] = None, **kwargs) -> Certificate
def get(resource_group_name: str, account_name: str, certificate_name: str, **kwargs) -> Certificate
def update(resource_group_name: str, account_name: str, certificate_name: str, parameters: Union[CertificateCreateOrUpdateParameters, IO[bytes]], if_match: Optional[str] = None, **kwargs) -> Certificate
def begin_delete(resource_group_name: str, account_name: str, certificate_name: str, **kwargs) -> LROPoller[None]
def cancel_deletion(resource_group_name: str, account_name: str, certificate_name: str, **kwargs) -> CertificateRegional quota management and virtual machine SKU availability queries.
def get_quotas(location_name: str, **kwargs) -> BatchLocationQuota
def list_supported_virtual_machine_skus(location_name: str, maxresults: Optional[int] = None, filter: Optional[str] = None, **kwargs) -> ItemPaged[SupportedSku]
def check_name_availability(location_name: str, parameters: Union[CheckNameAvailabilityParameters, IO[bytes]], **kwargs) -> CheckNameAvailabilityResultPrivate endpoint connections and network security perimeter configuration for secure network access.
def list_by_batch_account(resource_group_name: str, account_name: str, maxresults: Optional[int] = None, **kwargs) -> ItemPaged[PrivateEndpointConnection]
def get(resource_group_name: str, account_name: str, private_endpoint_connection_name: str, **kwargs) -> PrivateEndpointConnection
def begin_update(resource_group_name: str, account_name: str, private_endpoint_connection_name: str, parameters: Union[PrivateEndpointConnection, IO[bytes]], if_match: Optional[str] = None, **kwargs) -> LROPoller[PrivateEndpointConnection]
def begin_delete(resource_group_name: str, account_name: str, private_endpoint_connection_name: str, **kwargs) -> LROPoller[None]class BatchManagementClient:
def __init__(
self,
credential: TokenCredential,
subscription_id: str,
base_url: str = "https://management.azure.com",
**kwargs: Any
) -> None
# Operation attributes
batch_account: BatchAccountOperations
application_package: ApplicationPackageOperations
application: ApplicationOperations
location: LocationOperations
operations: Operations
certificate: CertificateOperations
private_link_resource: PrivateLinkResourceOperations
private_endpoint_connection: PrivateEndpointConnectionOperations
pool: PoolOperations
network_security_perimeter: NetworkSecurityPerimeterOperations
def close() -> None
def _send_request(request: HttpRequest, *, stream: bool = False, **kwargs: Any) -> HttpResponseclass BatchAccount:
id: str
name: str
type: str
location: str
tags: dict
identity: BatchAccountIdentity
provisioning_state: ProvisioningState
account_endpoint: str
node_management_endpoint: str
auto_storage: AutoStorageProperties
encryption: EncryptionProperties
dedicated_core_quota: int
low_priority_core_quota: int
dedicated_core_quota_per_vm_family: list
pool_allocation_mode: PoolAllocationMode
public_network_access: PublicNetworkAccessType
network_profile: NetworkProfile
class Pool:
id: str
name: str
type: str
etag: str
identity: BatchPoolIdentity
display_name: str
vm_size: str
deployment_configuration: DeploymentConfiguration
scale_settings: ScaleSettings
provisioning_state: PoolProvisioningState
allocation_state: AllocationState
creation_time: datetime
provisioning_state_transition_time: datetime
allocation_state_transition_time: datetime
current_dedicated_nodes: int
current_low_priority_nodes: int
network_configuration: NetworkConfiguration
class Application:
id: str
name: str
type: str
etag: str
display_name: str
allow_updates: bool
default_version: str