Object Storage Service:Manage LiveChannels (Python SDK V1)

How it works

LiveChannel operations follow a dependency order. Complete them in sequence:

1. Create a LiveChannel — returns an ingest URL and a playback URL.

2. Sign the ingest URL — required unless the bucket ACL is public-read-write.

3. Ingest a stream — connect your streaming tool (for example, OBS Studio) to the signed ingest URL.

4. Check stream status — query the live ingest state while a client is connected.

5. Generate a VOD playlist — package recorded TS files into an M3U8 playlist for on-demand playback.

6. Retrieve ingest history — get the most recent ingest records for the channel.

Prerequisites

Before you begin, ensure that you have:

• Python 3.6 (compatible with Python 2.6, 2.7, 3.3, 3.4, and 3.5)

• aliyun-oss-python-sdk 2.9.0

• OBS Studio or another RTMP-capable streaming tool

• An IDE

Set the following environment variables before running any example:

Create a LiveChannel

Call bucket.create_live_channel() to create a LiveChannel. The response includes an ingest URL (publish_url) and a playback URL (play_url).

If a LiveChannel with the same name already exists, it is overwritten and reset to its default status and settings.

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET')
bucket_name = os.getenv('OSS_TEST_BUCKET')
endpoint = os.getenv('OSS_TEST_ENDPOINT')

# Create a Bucket instance.
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

channel_name = "test_rtmp_live"
playlist_name = "test.m3u8"

create_result = bucket.create_live_channel(
    channel_name,
    oss2.models.LiveChannelInfo(
        status='enabled',
        description='A test live channel',
        target=oss2.models.LiveChannelInfoTarget(
            playlist_name=playlist_name,
            frag_count=3,       # Number of TS files in the M3U8 playlist
            frag_duration=5     # Duration (seconds) of each TS file; actual duration depends on keyframes
        )
    )
)

# Retrieve the ingest URL and playback URL from the result.
publish_url = create_result.publish_url   # RTMP ingest URL
play_url = create_result.play_url         # M3U8 playback URL
print("Ingest URL:", publish_url)
print("Playback URL:", play_url)

Key parameters for LiveChannelInfoTarget:

Use the ingest URL or playback URL:

• Use publish_url to push an RTMP stream. If the bucket ACL is not public-read-write, sign the URL first (see Sign the ingest URL).

• Use play_url to play back the live stream. Share this URL with players or embed it in your application.

Sign the ingest URL

If the bucket ACL is not public-read-write, sign the ingest URL before streaming. This operation supports V1 signatures only. OSS only validates the signature when a new connection is established — an active stream is not interrupted if the URL expires.

Use channel_name and playlist_name from the Create a LiveChannel step.

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET')
bucket_name = os.getenv('OSS_TEST_BUCKET')
endpoint = os.getenv('OSS_TEST_ENDPOINT')

bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

channel_name = "test_rtmp_live"
playlist_name = "test.m3u8"

# Sign the ingest URL. expires is the validity period in seconds.
# All parameters are included in the signature.
signed_url = bucket.sign_rtmp_url(channel_name, playlist_name, expires=3600)
print("Signed ingest URL:", signed_url)

Set the status of a LiveChannel

Toggle a LiveChannel between enabled and disabled with bucket.put_live_channel_status(). A successful call returns no value.

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET')
bucket_name = os.getenv('OSS_TEST_BUCKET')
endpoint = os.getenv('OSS_TEST_ENDPOINT')

bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

channel_name = "test_rtmp_live"

bucket.put_live_channel_status(channel_name, 'enabled')   # Enable the channel
bucket.put_live_channel_status(channel_name, 'disabled')  # Disable the channel

Get the status of a LiveChannel

Call bucket.get_live_channel_stat() to check the current stream ingest status.

This call returns status data only while a client is actively ingesting a stream (that is, the channel is in the Live state). If no client is connected, the status fields are empty.

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET')
bucket_name = os.getenv('OSS_TEST_BUCKET')
endpoint = os.getenv('OSS_TEST_ENDPOINT')

bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

channel_name = "test_rtmp_live"

get_status = bucket.get_live_channel_stat(channel_name)
print("Ingest status:", get_status.status)
print("Client IP:", get_status.remote_addr)
print("Connection time:", get_status.connected_time)

Get the configuration of a LiveChannel

Call bucket.get_live_channel() to retrieve the stored configuration for a channel.

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET')
bucket_name = os.getenv('OSS_TEST_BUCKET')
endpoint = os.getenv('OSS_TEST_ENDPOINT')

bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

channel_name = "test_rtmp_live"

get_result = bucket.get_live_channel(channel_name)
print("Description:", get_result.description)
print("Status:", get_result.status)
print("Target type:", get_result.target.type)
print("Fragment count:", get_result.target.frag_count)
print("Fragment duration:", get_result.target.frag_duration)
print("Playlist name:", get_result.target.playlist_name)

Generate and view a VOD playlist

The PostVodPlaylist operation queries the TS files recorded by the LiveChannel within a time range and combines them into an M3U8 playlist for VOD playback.

import os
import oss2
import time

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET')
bucket_name = os.getenv('OSS_TEST_BUCKET')
endpoint = os.getenv('OSS_TEST_ENDPOINT')

bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

channel_name = "test_rtmp_live"

# Set the time range. Both start_time and end_time are Unix timestamps (integers).
end_time = int(time.time())
start_time = end_time - 3600   # 1 hour ago

# Generate a VOD playlist for the specified time range.
# This creates "my_vod_list.m3u8" in the bucket.
bucket.post_vod_playlist(
    channel_name,
    "my_vod_list.m3u8",
    start_time=start_time,
    end_time=end_time
)

# Retrieve and display the playlist content.
result = bucket.get_vod_playlist(channel_name, start_time=start_time, end_time=end_time)
print("Playlist:", result.playlist)

post_vod_playlist can only generate a playlist from TS files that have already been uploaded. If you call it immediately after stopping a stream, some files may not yet be available.

List and delete LiveChannels

Use oss2.LiveChannelIterator to list channels, then delete them with bucket.delete_live_channel().

If multiple LiveChannels exist, this operation deletes only the latest LiveChannel that matches the prefix. To delete a specific channel, set prefix to the full channel name. If prefix is not specified, all channels in the bucket are listed. A successful deletion returns no value. If no channel matches the prefix, the call returns an error.

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET')
bucket_name = os.getenv('OSS_TEST_BUCKET')
endpoint = os.getenv('OSS_TEST_ENDPOINT')

bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# List channels whose names start with "test".
for info in oss2.LiveChannelIterator(bucket, prefix="test"):
    print(info.name)

# Delete the last channel returned by the iterator.
bucket.delete_live_channel(info.name)

Get ingest history

GetLiveChannelHistory returns up to 10 of the most recent ingest records for a channel, including start time, end time, and the client's remote address.

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET')
bucket_name = os.getenv('OSS_TEST_BUCKET')
endpoint = os.getenv('OSS_TEST_ENDPOINT')

bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

channel_name = "test_rtmp_live"

history_result = bucket.get_live_channel_history(channel_name)
print("Number of ingest records:", len(history_result.records))

This operation returns a maximum of 10 records. For each record, you can access the start time, end time, and remote address from history_result.records.

FAQ

Why does `get_live_channel_stat` return empty status fields?

get_live_channel_stat returns data only while a client is actively ingesting a stream. Connect your streaming tool to the signed ingest URL first, then call get_live_channel_stat.

I get `'Code': 'InvalidArgument', 'Message': 'No ts file found in specified time span.'`

OSS generates the VOD playlist from TS files that have already been uploaded. Call post_vod_playlist only after the stream files are available in the bucket.

What data types are required for `end_time` and `start_time` in `post_vod_playlist`?

Both are integers (Unix timestamps). Use int(time.time()) to get the current time.

What is the data type returned for channel names by `LiveChannelIterator`?

String. For the full schema, see ListLiveChannel.

What's next

• LiveChannel operations — full API reference

• live_channel.py — SDK example file

• api.py — SDK source

• GetLiveChannelHistory

• PostVodPlaylist