Microsoft Azure Media Services Client Library for Python - A management library for Azure Media Services that provides programmatic access to media processing and streaming capabilities in the cloud.
83
Quality
Pending
Does it follow best practices?
Impact
83%
1.09xAverage score across 10 eval scenarios
Comprehensive content delivery capabilities through streaming endpoints, locators, and policies for delivering media content to end users. Supports adaptive bitrate streaming, multiple delivery protocols (HLS, DASH, Smooth Streaming), global CDN distribution, and dynamic packaging with scale-out streaming infrastructure.
Manage streaming endpoints that serve as the delivery infrastructure for media content with CDN integration and scaling capabilities.
def list(resource_group_name: str, account_name: str) -> Iterable[StreamingEndpoint]:
"""
List all streaming endpoints in a media service account.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
Returns:
Iterable of StreamingEndpoint objects with configuration and status
"""
def get(
resource_group_name: str,
account_name: str,
streaming_endpoint_name: str
) -> StreamingEndpoint:
"""
Get a specific streaming endpoint with complete configuration.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_endpoint_name: Name of the streaming endpoint (str)
Returns:
StreamingEndpoint object with current state and configuration
"""
def begin_create(
resource_group_name: str,
account_name: str,
streaming_endpoint_name: str,
parameters: StreamingEndpoint
) -> LROPoller[StreamingEndpoint]:
"""
Create a new streaming endpoint.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_endpoint_name: Name for the new streaming endpoint (str)
- parameters: Streaming endpoint configuration (StreamingEndpoint)
Returns:
LROPoller for tracking the creation operation
"""
def begin_update(
resource_group_name: str,
account_name: str,
streaming_endpoint_name: str,
parameters: StreamingEndpoint
) -> LROPoller[StreamingEndpoint]:
"""
Update an existing streaming endpoint configuration.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_endpoint_name: Name of the streaming endpoint (str)
- parameters: Updated streaming endpoint configuration (StreamingEndpoint)
Returns:
LROPoller for tracking the update operation
"""
def begin_delete(
resource_group_name: str,
account_name: str,
streaming_endpoint_name: str
) -> LROPoller[None]:
"""
Delete a streaming endpoint.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_endpoint_name: Name of the streaming endpoint (str)
Returns:
LROPoller for tracking the deletion operation
"""Control streaming endpoint lifecycle including start, stop, and scaling operations with long-running operation support.
def begin_start(
resource_group_name: str,
account_name: str,
streaming_endpoint_name: str
) -> LROPoller[None]:
"""
Start a streaming endpoint to begin serving content.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_endpoint_name: Name of the streaming endpoint (str)
Returns:
LROPoller for tracking the start operation
"""
def begin_stop(
resource_group_name: str,
account_name: str,
streaming_endpoint_name: str
) -> LROPoller[None]:
"""
Stop a streaming endpoint to cease content delivery.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_endpoint_name: Name of the streaming endpoint (str)
Returns:
LROPoller for tracking the stop operation
"""
def begin_scale(
resource_group_name: str,
account_name: str,
streaming_endpoint_name: str,
parameters: StreamingEndpointScaleUnits
) -> LROPoller[None]:
"""
Scale a streaming endpoint to handle increased traffic.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_endpoint_name: Name of the streaming endpoint (str)
- parameters: Scaling configuration (StreamingEndpointScaleUnits)
Returns:
LROPoller for tracking the scaling operation
"""Retrieve streaming endpoint capabilities and pricing information.
def skus(
resource_group_name: str,
account_name: str,
streaming_endpoint_name: str
) -> StreamingEndpointSkuInfoListResult:
"""
List available SKUs and pricing tiers for streaming endpoints.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_endpoint_name: Name of the streaming endpoint (str)
Returns:
StreamingEndpointSkuInfoListResult with available SKU information
"""Manage streaming locators that define how assets are published and accessed by end users.
def list(resource_group_name: str, account_name: str) -> Iterable[StreamingLocator]:
"""
List all streaming locators in a media service account.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
Returns:
Iterable of StreamingLocator objects with publishing configuration
"""
def get(
resource_group_name: str,
account_name: str,
streaming_locator_name: str
) -> StreamingLocator:
"""
Get a specific streaming locator with complete configuration.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_locator_name: Name of the streaming locator (str)
Returns:
StreamingLocator object with publishing details
"""
def create(
resource_group_name: str,
account_name: str,
streaming_locator_name: str,
parameters: StreamingLocator
) -> StreamingLocator:
"""
Create a streaming locator to publish an asset.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_locator_name: Name for the new streaming locator (str)
- parameters: Streaming locator configuration (StreamingLocator)
Returns:
Created StreamingLocator object
"""
def delete(
resource_group_name: str,
account_name: str,
streaming_locator_name: str
) -> None:
"""
Delete a streaming locator.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_locator_name: Name of the streaming locator (str)
Returns:
None
"""Generate streaming URLs and retrieve content keys for published assets.
def list_paths(
resource_group_name: str,
account_name: str,
streaming_locator_name: str
) -> ListPathsResponse:
"""
List streaming URLs and download paths for a streaming locator.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_locator_name: Name of the streaming locator (str)
Returns:
ListPathsResponse containing streaming and download URLs
"""
def list_content_keys(
resource_group_name: str,
account_name: str,
streaming_locator_name: str
) -> ListContentKeysResponse:
"""
List content keys for a streaming locator with encryption.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_locator_name: Name of the streaming locator (str)
Returns:
ListContentKeysResponse containing content encryption keys
"""Manage streaming policies that define delivery protocols and encryption settings.
def list(resource_group_name: str, account_name: str) -> Iterable[StreamingPolicy]:
"""
List all streaming policies in a media service account.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
Returns:
Iterable of StreamingPolicy objects with delivery configuration
"""
def get(
resource_group_name: str,
account_name: str,
streaming_policy_name: str
) -> StreamingPolicy:
"""
Get a specific streaming policy with complete configuration.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_policy_name: Name of the streaming policy (str)
Returns:
StreamingPolicy object with delivery and encryption settings
"""
def create(
resource_group_name: str,
account_name: str,
streaming_policy_name: str,
parameters: StreamingPolicy
) -> StreamingPolicy:
"""
Create a streaming policy with delivery and encryption configuration.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_policy_name: Name for the new streaming policy (str)
- parameters: Streaming policy configuration (StreamingPolicy)
Returns:
Created StreamingPolicy object
"""
def delete(
resource_group_name: str,
account_name: str,
streaming_policy_name: str
) -> None:
"""
Delete a streaming policy.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- streaming_policy_name: Name of the streaming policy (str)
Returns:
None
"""class StreamingEndpoint:
"""Streaming endpoint for content delivery."""
name: str
description: str
location: str
scale_units: int
availability_set_name: str
access_control: StreamingEndpointAccessControl
max_cache_age: int
custom_host_names: List[str]
host_name: str
cdn_enabled: bool
cdn_provider: str
cdn_profile: str
provisioning_state: str
resource_state: str # StreamingEndpointResourceState enum
cross_site_access_policies: CrossSiteAccessPolicies
free_trial_end_time: str
created: str
last_modified: str
class StreamingEndpointAccessControl:
"""Access control configuration for streaming endpoints."""
akamai_signature_header_authentication_key_list: List[AkamaiSignatureHeaderAuthenticationKey]
ip: IPAccessControl
class StreamingEndpointScaleUnits:
"""Scaling configuration for streaming endpoints."""
scale_unit: int
class StreamingEndpointSkuInfoListResult:
"""Available SKU information for streaming endpoints."""
value: List[StreamingEndpointSkuInfo]
class StreamingEndpointSkuInfo:
"""SKU information for streaming endpoints."""
resource_type: str
sku: StreamingEndpointSku
capacity: StreamingEndpointSkuCapacity
class StreamingLocator:
"""Configuration for publishing assets to streaming endpoints."""
name: str
asset_name: str
start_time: str
end_time: str
streaming_locator_id: str
streaming_policy_name: str
default_content_key_policy_name: str
content_keys: List[StreamingLocatorContentKey]
alternative_media_id: str
filters: List[str]
created: str
class StreamingLocatorContentKey:
"""Content key configuration for streaming locators."""
id: str
type: str # StreamingLocatorContentKeyType enum
label_reference_in_streaming_policy: str
value: str
policy_name: str
tracks: List[TrackSelection]
class ListPathsResponse:
"""Response containing streaming URLs and download paths."""
streaming_paths: List[StreamingPath]
download_paths: List[str]
class StreamingPath:
"""Streaming path information."""
streaming_protocol: str # StreamingPolicyStreamingProtocol enum
encryption_scheme: str # EncryptionScheme enum
paths: List[str]
class ListContentKeysResponse:
"""Response containing content encryption keys."""
content_keys: List[StreamingLocatorContentKey]
class StreamingPolicy:
"""Policy defining streaming delivery protocols and encryption."""
name: str
default_content_key_policy_name: str
envelope_encryption: EnvelopeEncryption
common_encryption_cenc: CommonEncryptionCenc
common_encryption_cbcs: CommonEncryptionCbcs
no_encryption: NoEncryption
created: str
class EnvelopeEncryption:
"""Envelope encryption configuration."""
enabled_protocols: EnabledProtocols
clear_tracks: List[TrackSelection]
content_keys: StreamingPolicyContentKeys
custom_key_acquisition_url_template: str
class CommonEncryptionCenc:
"""Common encryption CENC configuration."""
enabled_protocols: EnabledProtocols
clear_tracks: List[TrackSelection]
content_keys: StreamingPolicyContentKeys
drm: CencDrmConfiguration
class CommonEncryptionCbcs:
"""Common encryption CBCS configuration."""
enabled_protocols: EnabledProtocols
clear_tracks: List[TrackSelection]
content_keys: StreamingPolicyContentKeys
drm: CbcsDrmConfiguration
class NoEncryption:
"""Clear streaming configuration without encryption."""
enabled_protocols: EnabledProtocols
class EnabledProtocols:
"""Enabled streaming protocols."""
download: bool
dash: bool
hls: bool
smooth_streaming: boolfrom azure.mgmt.media import AzureMediaServices
from azure.mgmt.media.models import StreamingEndpoint, IPAccessControl, IPRange
from azure.identity import DefaultAzureCredential
client = AzureMediaServices(
credential=DefaultAzureCredential(),
subscription_id="your-subscription-id"
)
# Configure access control (allow all - use restrictive rules in production)
ip_range = IPRange(
name="AllowAll",
address="0.0.0.0",
subnet_prefix_length=0
)
access_control = StreamingEndpointAccessControl(
ip=IPAccessControl(allow=[ip_range])
)
# Create streaming endpoint
streaming_endpoint = StreamingEndpoint(
location="East US",
description="Primary streaming endpoint for content delivery",
scale_units=1,
cdn_enabled=True,
access_control=access_control,
max_cache_age=300
)
# Start creation
create_operation = client.streaming_endpoints.begin_create(
resource_group_name="my-resource-group",
account_name="my-media-service",
streaming_endpoint_name="default",
parameters=streaming_endpoint
)
created_endpoint = create_operation.result()
print(f"Streaming endpoint created: {created_endpoint.name}")
# Start the streaming endpoint
start_operation = client.streaming_endpoints.begin_start(
resource_group_name="my-resource-group",
account_name="my-media-service",
streaming_endpoint_name="default"
)
start_operation.result()
print("Streaming endpoint started and ready to serve content")from azure.mgmt.media.models import StreamingLocator
from datetime import datetime, timedelta
# Create streaming locator
streaming_locator = StreamingLocator(
asset_name="encoded-asset",
streaming_policy_name="Predefined_ClearStreamingOnly",
start_time=datetime.utcnow().isoformat() + "Z",
end_time=(datetime.utcnow() + timedelta(days=365)).isoformat() + "Z"
)
created_locator = client.streaming_locators.create(
resource_group_name="my-resource-group",
account_name="my-media-service",
streaming_locator_name="clear-streaming-locator",
parameters=streaming_locator
)
print(f"Streaming locator created: {created_locator.name}")
# Get streaming URLs
paths_response = client.streaming_locators.list_paths(
resource_group_name="my-resource-group",
account_name="my-media-service",
streaming_locator_name="clear-streaming-locator"
)
# Get streaming endpoint hostname
streaming_endpoint = client.streaming_endpoints.get(
resource_group_name="my-resource-group",
account_name="my-media-service",
streaming_endpoint_name="default"
)
base_url = f"https://{streaming_endpoint.host_name}"
print("Streaming URLs:")
for streaming_path in paths_response.streaming_paths:
protocol = streaming_path.streaming_protocol
print(f"\n{protocol} URLs:")
for path in streaming_path.paths:
full_url = f"{base_url}{path}"
print(f" {full_url}")
if paths_response.download_paths:
print("\nDownload URLs:")
for path in paths_response.download_paths:
full_url = f"{base_url}{path}"
print(f" {full_url}")from azure.mgmt.media.models import (
StreamingPolicy, NoEncryption, EnabledProtocols
)
# Configure clear streaming policy
enabled_protocols = EnabledProtocols(
download=False,
dash=True,
hls=True,
smooth_streaming=True
)
no_encryption = NoEncryption(
enabled_protocols=enabled_protocols
)
# Create streaming policy
streaming_policy = StreamingPolicy(
no_encryption=no_encryption
)
created_policy = client.streaming_policies.create(
resource_group_name="my-resource-group",
account_name="my-media-service",
streaming_policy_name="custom-clear-policy",
parameters=streaming_policy
)
print(f"Streaming policy created: {created_policy.name}")from azure.mgmt.media.models import StreamingEndpointScaleUnits
# Check current streaming endpoint status
endpoint = client.streaming_endpoints.get(
resource_group_name="my-resource-group",
account_name="my-media-service",
streaming_endpoint_name="default"
)
print(f"Current scale units: {endpoint.scale_units}")
print(f"Current state: {endpoint.resource_state}")
# Scale up to handle more traffic
scale_params = StreamingEndpointScaleUnits(scale_unit=3)
scale_operation = client.streaming_endpoints.begin_scale(
resource_group_name="my-resource-group",
account_name="my-media-service",
streaming_endpoint_name="default",
parameters=scale_params
)
scale_operation.result()
print("Streaming endpoint scaled to 3 units")
# Verify scaling
scaled_endpoint = client.streaming_endpoints.get(
resource_group_name="my-resource-group",
account_name="my-media-service",
streaming_endpoint_name="default"
)
print(f"New scale units: {scaled_endpoint.scale_units}")Install with Tessl CLI
npx tessl i tessl/pypi-azure-mgmt-mediadocs
evals
scenario-1
scenario-2
scenario-3
scenario-4
scenario-5
scenario-6
scenario-7
scenario-8
scenario-9
scenario-10