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 encoding and media processing capabilities through transforms (reusable job templates) and jobs (individual processing instances). This system enables video transcoding, audio processing, content analysis, thumbnail generation, and media format conversion with support for both built-in and custom encoding configurations.
Create and manage encoding transforms that serve as templates for media processing jobs.
def list(resource_group_name: str, account_name: str) -> Iterable[Transform]:
"""
List all transforms 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 Transform objects containing processing configurations
"""
def get(resource_group_name: str, account_name: str, transform_name: str) -> Transform:
"""
Get a specific transform with complete configuration.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- transform_name: Name of the transform (str)
Returns:
Transform object with preset and output configurations
"""
def create_or_update(
resource_group_name: str,
account_name: str,
transform_name: str,
parameters: Transform
) -> Transform:
"""
Create or update a transform with encoding presets.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- transform_name: Name of the transform (str)
- parameters: Transform configuration with presets (Transform)
Returns:
Created or updated Transform object
"""
def update(
resource_group_name: str,
account_name: str,
transform_name: str,
parameters: Transform
) -> Transform:
"""
Update an existing transform configuration.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- transform_name: Name of the transform (str)
- parameters: Updated transform configuration (Transform)
Returns:
Updated Transform object
"""
def delete(resource_group_name: str, account_name: str, transform_name: str) -> None:
"""
Delete a transform.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- transform_name: Name of the transform (str)
Returns:
None
"""Create, monitor, and manage encoding jobs based on transform templates.
def list(resource_group_name: str, account_name: str, transform_name: str) -> Iterable[Job]:
"""
List all jobs for a specific transform.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- transform_name: Name of the transform (str)
Returns:
Iterable of Job objects with processing status and details
"""
def get(
resource_group_name: str,
account_name: str,
transform_name: str,
job_name: str
) -> Job:
"""
Get a specific job with detailed status and progress information.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- transform_name: Name of the transform (str)
- job_name: Name of the job (str)
Returns:
Job object with current state, progress, and error information
"""
def create(
resource_group_name: str,
account_name: str,
transform_name: str,
job_name: str,
parameters: Job
) -> Job:
"""
Create a new encoding job from a transform.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- transform_name: Name of the transform (str)
- job_name: Name for the new job (str)
- parameters: Job configuration with input and output (Job)
Returns:
Created Job object with initial status
"""
def update(
resource_group_name: str,
account_name: str,
transform_name: str,
job_name: str,
parameters: Job
) -> Job:
"""
Update job properties such as description or priority.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- transform_name: Name of the transform (str)
- job_name: Name of the job (str)
- parameters: Updated job configuration (Job)
Returns:
Updated Job object
"""
def delete(
resource_group_name: str,
account_name: str,
transform_name: str,
job_name: str
) -> None:
"""
Delete a job.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- transform_name: Name of the transform (str)
- job_name: Name of the job (str)
Returns:
None
"""
def cancel_job(
resource_group_name: str,
account_name: str,
transform_name: str,
job_name: str
) -> None:
"""
Cancel a running job.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- transform_name: Name of the transform (str)
- job_name: Name of the job (str)
Returns:
None
"""class Transform:
"""Encoding transform template defining processing operations."""
name: str
description: str
outputs: List[TransformOutput]
created: str
last_modified: str
class TransformOutput:
"""Output configuration for a transform."""
preset: Preset
on_error: str # OnErrorType enum (StopProcessingJob, ContinueJob)
relative_priority: str # Priority enum (Low, Normal, High)
class Job:
"""Encoding or analysis job instance."""
name: str
description: str
input: JobInput
outputs: List[JobOutput]
priority: str # Priority enum
correlationData: dict
state: str # JobState enum
created: str
last_modified: str
start_time: str
end_time: str
class JobInput:
"""Base class for job inputs."""
pass
class JobInputAsset(JobInput):
"""Asset-based job input."""
asset_name: str
files: List[str]
start: str
end: str
label: str
class JobInputHttp(JobInput):
"""HTTP-based job input."""
base_uri: str
files: List[str]
start: str
end: str
label: str
class JobOutput:
"""Base class for job outputs."""
label: str
error: JobError
state: str # JobState enum
progress: int
class JobOutputAsset(JobOutput):
"""Asset-based job output."""
asset_name: str
# Encoding Presets
class Preset:
"""Base class for encoding presets."""
pass
class BuiltInStandardEncoderPreset(Preset):
"""Built-in encoding preset."""
preset_name: str # EncoderNamedPreset enum
configurations: PresetConfigurations
class StandardEncoderPreset(Preset):
"""Custom encoding preset."""
codecs: List[Codec]
formats: List[Format]
filters: Filters
class AudioAnalyzerPreset(Preset):
"""Audio analysis preset."""
audio_language: str
mode: str # AudioAnalysisMode enum
class VideoAnalyzerPreset(Preset):
"""Video analysis preset."""
audio_language: str
mode: str # VideoAnalysisMode enum
insights_to_extract: str # InsightsType enum
class FaceDetectorPreset(Preset):
"""Face detection preset."""
resolution: str # AnalysisResolution enum
mode: str # FaceDetectorMode enum
blur_type: str # BlurType enum
face_redactor_mode: str # FaceRedactorMode enum
# Codecs
class Codec:
"""Base class for codecs."""
label: str
class H264Video(Codec):
"""H.264 video codec."""
scene_change_detection: bool
complexity: str # H264Complexity enum
layers: List[H264Layer]
key_frame_interval: str
stretch_mode: str # StretchMode enum
sync_mode: str # VideoSyncMode enum
class H265Video(Codec):
"""H.265 video codec."""
scene_change_detection: bool
complexity: str # H265Complexity enum
layers: List[H265Layer]
key_frame_interval: str
stretch_mode: str # StretchMode enum
sync_mode: str # VideoSyncMode enum
class AacAudio(Codec):
"""AAC audio codec."""
channels: int
sampling_rate: int
bitrate: int
profile: str # AacAudioProfile enum
# Formats
class Format:
"""Base class for output formats."""
filename_pattern: str
class Mp4Format(Format):
"""MP4 container format."""
output_files: List[OutputFile]
class TransportStreamFormat(Format):
"""MPEG transport stream format."""
output_files: List[OutputFile]
class JpgFormat(Format):
"""JPEG image format."""
pass
class PngFormat(Format):
"""PNG image format."""
passfrom azure.mgmt.media import AzureMediaServices
from azure.mgmt.media.models import (
Transform, TransformOutput, BuiltInStandardEncoderPreset,
EncoderNamedPreset, Priority, OnErrorType
)
from azure.identity import DefaultAzureCredential
client = AzureMediaServices(
credential=DefaultAzureCredential(),
subscription_id="your-subscription-id"
)
# Create transform with built-in preset
preset = BuiltInStandardEncoderPreset(
preset_name=EncoderNamedPreset.ADAPTIVE_STREAMING
)
transform_output = TransformOutput(
preset=preset,
on_error=OnErrorType.STOP_PROCESSING_JOB,
relative_priority=Priority.NORMAL
)
transform = Transform(
description="Adaptive streaming transform",
outputs=[transform_output]
)
created_transform = client.transforms.create_or_update(
resource_group_name="my-resource-group",
account_name="my-media-service",
transform_name="adaptive-streaming-transform",
parameters=transform
)
print(f"Transform created: {created_transform.name}")from azure.mgmt.media.models import (
Job, JobInputAsset, JobOutputAsset, Priority
)
import time
# Create job from transform
job_input = JobInputAsset(asset_name="input-asset")
job_output = JobOutputAsset(asset_name="output-asset")
job = Job(
description="Encode video for adaptive streaming",
input=job_input,
outputs=[job_output],
priority=Priority.NORMAL
)
created_job = client.jobs.create(
resource_group_name="my-resource-group",
account_name="my-media-service",
transform_name="adaptive-streaming-transform",
job_name="encoding-job-001",
parameters=job
)
print(f"Job created: {created_job.name}")
print(f"Initial state: {created_job.state}")
# Monitor job progress
while True:
job_status = client.jobs.get(
resource_group_name="my-resource-group",
account_name="my-media-service",
transform_name="adaptive-streaming-transform",
job_name="encoding-job-001"
)
print(f"Job state: {job_status.state}")
if job_status.outputs:
for output in job_status.outputs:
print(f"Output progress: {output.progress}%")
if output.error:
print(f"Output error: {output.error.message}")
if job_status.state in ["Finished", "Error", "Canceled"]:
break
time.sleep(10)
print(f"Job completed with state: {job_status.state}")from azure.mgmt.media.models import (
StandardEncoderPreset, H264Video, H264Layer, AacAudio,
Mp4Format, H264Complexity, AacAudioProfile
)
# Define video codec with multiple layers
h264_layers = [
H264Layer(
bitrate=3400000,
width="1920",
height="1080",
label="HD"
),
H264Layer(
bitrate=2000000,
width="1280",
height="720",
label="SD"
),
H264Layer(
bitrate=1000000,
width="640",
height="360",
label="Mobile"
)
]
h264_codec = H264Video(
complexity=H264Complexity.BALANCED,
layers=h264_layers,
key_frame_interval="PT2S",
scene_change_detection=True
)
# Define audio codec
aac_codec = AacAudio(
channels=2,
sampling_rate=48000,
bitrate=128000,
profile=AacAudioProfile.AAC_LC
)
# Define output format
mp4_format = Mp4Format(
filename_pattern="{Basename}_{Label}_{Bitrate}.mp4"
)
# Create custom preset
custom_preset = StandardEncoderPreset(
codecs=[h264_codec, aac_codec],
formats=[mp4_format]
)
# Create transform with custom preset
custom_transform_output = TransformOutput(
preset=custom_preset,
on_error=OnErrorType.CONTINUE_JOB,
relative_priority=Priority.HIGH
)
custom_transform = Transform(
description="Custom multi-bitrate encoding",
outputs=[custom_transform_output]
)
client.transforms.create_or_update(
resource_group_name="my-resource-group",
account_name="my-media-service",
transform_name="custom-encoding-transform",
parameters=custom_transform
)
print("Custom encoding transform created")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