plycdn for Python

1Install the packages

Install only what you use. The two packages are independent: use either, or both in one application.

pip install "./plycdn_content_search-1.19.0-py3-none-any.whl[fastapi]"      # Content Search
pip install "./plycdn-1.11.0-py3-none-any.whl[fastapi,s3,azure]"            # storage, delivery and video

The extras add what you ask for and nothing else:

• fastapi: the routes for your frontend and the webhook receiver, for your FastAPI application.
• azure (both packages): an Azure Blob container as the store of your customers' files for Content Search uploads, and an Azure Blob client for plycdn storage.
• s3 (plycdn only): an S3 client for plycdn storage.

If your servers restrict outbound traffic, allow api.plycdn.com on port 443.

2Configure

Create the clients once, when your server starts. Keep the API key and the callback key in your secret store, never in source control and never in a browser bundle. Every option is a keyword argument.

import os

from plycdn.storage import PlycdnStorageApi
from plycdn_content_search import PlycdnContentApi

content = PlycdnContentApi(os.environ["PLYCDN_API_KEY"], callback_key=os.environ["PLYCDN_CALLBACK_KEY"])
storage = PlycdnStorageApi(os.environ["PLYCDN_API_KEY"])

Every method takes the organization it acts for as its first argument. Pass the plycdn organization id of the customer the call is for; a malformed id raises ValueError before anything is sent. Optional arguments are keyword-only, in snake case; a word that Python reserves gets a trailing underscore (from_ for from). Close a client with with (or async with) or close() (await aclose()) when your program ends.

What is retried

Reads, replacements and deletes are retried after a temporary failure (a network error, a timeout, a 408, 429 or 5xx), waiting longer each time and honouring Retry-After. A call that creates something is retried only when it carries an idempotency key, so a retry can never create a second copy: a create sent without a key is sent exactly once. These carry a key automatically: registrations, imports, re-indexing, transcript corrections, and creating a bucket, library, video, webhook, domain, live input or destination. Pass idempotency_key= to choose your own.

3Content Search from your server

import os
from datetime import datetime, timedelta, timezone

from plycdn_content_search import PlycdnContentApi, paginate

content = PlycdnContentApi(os.environ["PLYCDN_API_KEY"], callback_key=os.environ["PLYCDN_CALLBACK_KEY"])


def demo(organization_id: str) -> None:
    # Register a file that already sits in your storage. The same file registered again is the same asset.
    content.register_asset(
        organization_id,
        {"externalId": organization_id + "/lecture-1.mp4", "filename": "lecture-1.mp4", "kind": "video",
         "sourceUrl": "https://files.example.com/lecture-1.mp4", "sourceVersion": "1"},
        {"url": "https://files.example.com/lecture-1.mp4?token=...",
         "expiresAt": (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()},
        collections=["week-1"], principals=["group:students"],
    )

    # Search by meaning and wording.
    found = content.search(organization_id, {"query": "how do I reset my password", "limit": 5})
    print(found["results"])

    # Written answer, as it is written. Stopping early is not charged.
    for event in content.stream_answer(organization_id, {"query": "what changed in the new policy?"}):
        if event["type"] == "delta":
            print(event.get("text", ""), end="")
        elif event["type"] == "citations":
            print(event["citations"])
        elif event["type"] == "error":
            raise RuntimeError("The answer stopped early")

    # Every document, page by page.
    for item in paginate(lambda cursor: content.find_documents(organization_id, cursor=cursor, limit=100)):
        print(item)

    print(content.get_usage(organization_id, year=2026, month=10))

Requests and answers are plain dictionaries with the same camelCase keys as the API reference, so its examples apply unchanged; only method and argument names are snake case. import_assets brings in many files at once (idempotency_key= makes a retry safe; import is a Python keyword, hence the name). answer returns the whole written answer in one piece; stream_answer yields delta, citations, done and error events as they arrive. The same object also covers the activity list, jobs, transcripts, metadata, permissions, collections, synonyms, boosts, pins, connectors, usage and settings; every method is named like its ASP.NET Core counterpart, without the Async and in snake case (Backend API reference).

The asynchronous form is the same call with await, and stream_answer with async for:

import os

from plycdn_content_search import AsyncPlycdnContentApi, PlycdnAccountApi


async def ask(organization_id: str) -> str:
    async with AsyncPlycdnContentApi(os.environ["PLYCDN_API_KEY"]) as content:
        reply = await content.answer(organization_id, {"query": "what changed in the new policy?"})
        return reply["answer"] or ""


def customers(account_id: str) -> list:
    # Sub-organizations, with an admin key.
    accounts = PlycdnAccountApi(os.environ["PLYCDN_ADMIN_KEY"])
    return accounts.list_sub_organizations(account_id)

4Routes for your own frontend

The Angular and React packages talk to your backend, not to plycdn. content_search_router is that backend: search, answers, activity, uploads and the callback plycdn makes to read your files. It is a router for your FastAPI application, and it uses the asynchronous API.

import os

from fastapi import FastAPI

from plycdn_content_search import AsyncPlycdnContentApi
from plycdn_content_search.azure import AzureBlobFileStorage
from plycdn_content_search.fastapi import content_search_router

api = AsyncPlycdnContentApi(os.environ["PLYCDN_API_KEY"], callback_key=os.environ["PLYCDN_CALLBACK_KEY"])
files = AzureBlobFileStorage.from_connection_string(os.environ["AZURE_STORAGE_CONNECTION_STRING"], "content")


def signed_in(request):
    return getattr(request.state, "user", None)


def resolve_organization(request):
    user = signed_in(request)
    return user.plycdn_organization_id if user else None


def authorize(request, access):
    user = signed_in(request)
    roles = user.roles if user else []
    if access == "read":
        return "reader" in roles or "manager" in roles
    return "manager" in roles


def resolve_principals(request):
    user = signed_in(request)
    return ["role:" + role for role in user.roles] if user else []


def add_content_search(app):
    app.include_router(
        content_search_router(
            api=api,
            resolve_organization=resolve_organization,
            authorize=authorize,
            resolve_principals=resolve_principals,
            resolve_user=lambda request: signed_in(request).id if signed_in(request) else None,
            storage=files,
            replay=files,
            callback_secret=os.environ["PLYCDN_CALLBACK_SECRET"],
        ),
        prefix="/api/plycdn-content/v1",
    )

Call add_content_search(app) on your application at start-up. Every hook may be a plain function or a coroutine.

How it decides, in plain terms:

• Every route is refused until authorize says yes. With no authorize, every route answers 403 forbidden and nothing is sent to plycdn. access is "read" (search, answers, status, viewing) or "manage" (uploads, re-indexing, cancelling, deleting, collections); a manage route needs both.
• authorize must return exactly True. 1, "yes", a non-empty list or an object are refusals. If the hook raises, that is your application's error (a 500), not an answer to the browser.
• The organization comes only from resolve_organization. A browser that sends X-Organization-Id, ?organizationId= or organizationId in a body still acts for the organization you resolved. Return an id string (or a uuid.UUID); anything else, or nothing, refuses the request with 403 and sends nothing. Likewise the principals used for permission-aware search come only from resolve_principals, which returns a list or tuple of strings (a single string is a mistake in your code and is answered 500).
• Uploads and the source callback need storage (where your customers' files live; an Azure Blob container is included, and you can supply your own object with create_upload, complete_upload and resolve_read), replay (an object with try_use(nonce, expires_at) that remembers callback requests so none is served twice; it must return exactly True) and a callback_secret of at least 32 characters. Without all three, those routes answer 404. The callback, /source-callback, needs no sign-in: it is checked by its signature instead, so it must be reachable without your session or CSRF middleware.
• Upload sessions are sealed so a browser cannot change them. They use a key derived from callback_secret; if you would rather not have that secret also seal sessions, pass a separate ticket_secret of at least 32 characters.
• Other options: public_read_base_url and public_write_base_url (https only), max_upload_bytes, allow_upload_relay, warn for the router's own warnings and storage_client (your own httpx.AsyncClient for the upload relay).

The router answers browsers with these codes: 403 forbidden (the hook said no, or answered something other than True), 403 organization_forbidden (no organization for this request, or an upload session that belongs to someone else or has expired), 400 invalid_request (a body that is missing, not application/json, not valid JSON, containing NaN or Infinity, nested more than 64 levels deep, a number longer than the allowed size, a malformed Idempotency-Key, or a number outside the allowed range), 413 for a body over the limit, 400 invalid_upload_session (a changed session), and what plycdn itself answers for a refused call (404, 429, ...), passed on as it is. A failure to reach plycdn is a 502 content_unavailable; a storage problem is 502 storage_unavailable or 409 storage_operation_incomplete. These are the same codes as the ASP.NET Core controller.

Body limits and mounting

The routers read each request body themselves, a piece at a time with a running size check, so an oversized body is refused without being held in memory: JSON up to 1 MiB, the callback up to 16 KiB, uploads up to max_upload_bytes. Add no middleware of your own that reads a whole body before the router sees it, and include the router with the prefix you gave plycdn: the callback address you registered in the dashboard is https://YOUR_BACKEND/api/plycdn-content/v1/source-callback.

Behind a load balancer, and cross-site

For a bucket that binds links to the viewer's address, the sign route signs for the address of the request, request.client.host. Behind a load balancer or reverse proxy, run uvicorn with proxy headers on (--proxy-headers) and --forwarded-allow-ips set to your balancer's address, so that address is the viewer's and not the proxy's. The address must be IPv4; with none, or an IPv6 one, the sign route answers 422 validation_error rather than issue a link the delivery network would refuse.

Cross-site protection is your application's job, as for any other route of yours. If a cookie signs your users in, set it SameSite=Lax (or Strict) and keep your usual CSRF protection on POST and DELETE; use SameSite=None only for a frontend on another site, together with that protection. A cross-origin frontend also needs your own CORS policy for these paths. Turn your application's debug mode off in production, so error pages show no stack traces.

5Storage and delivery

import os

from plycdn.storage import PlycdnLinkSigner, PlycdnStorageApi, PlycdnUploader

storage = PlycdnStorageApi(os.environ["PLYCDN_ADMIN_KEY"])
uploader = PlycdnUploader(storage)
signer = PlycdnLinkSigner(storage)


def publish(organization_id: str, path: str) -> str:
    # Admin key. A bucket's name is global, like an S3 bucket's: choose one that is yours.
    bucket = storage.create_bucket(organization_id, {"name": "acme-course-media", "region": "mumbai"})
    with open(path, "rb") as file:
        # Plans the parts, uploads them with retries, completes the session.
        stored = uploader.upload(organization_id, bucket["id"], "week-1/intro.mp4", file, os.path.getsize(path), "video/mp4")
    # Made in your process from the bucket's cached signing profile: no request to plycdn per link.
    return signer.sign(organization_id, bucket["id"], stored["key"], ttl_seconds=900)

Region codes come from storage.list_regions(organization_id); treat them as data. A bucket is creating until it is active (poll get_bucket). PlycdnUploader is for bytes your server holds (an import, a migration, a nightly job); when a browser uploads, it goes straight to the bucket with a short-lived ticket and your backend only relays the control calls through the routes below. An upload interrupted by a temporary failure keeps its session: the error carries upload_id, so you can resume or abort_upload it. An organization can hold at most 1,000 open uploads; one more fails with plan_limit_reached (403).

signer.sign caches each bucket's signing profile for 10 minutes (signing_profile_cache, in seconds, on the client). After you rotate a bucket's signing key, call signer.forget(organization_id, bucket_id). Options: ttl_seconds (60 seconds to 7 days), ip (for a bucket that binds links to the viewer), hostname and tags (delivery analytics). A custom domain's hostname must be passed in its ASCII (punycode) form, such as xn--...; a non-ASCII spelling is refused with validation_error. Delivery analytics, purges, domains and usage are methods of PlycdnStorageApi (get_delivery_analytics, purge_cache, create_domain, get_usage_by_resource, ...). The asynchronous forms are AsyncPlycdnStorageApi, AsyncPlycdnUploader (its content may also be an async iterable of chunks) and AsyncPlycdnLinkSigner.

6Video and live

import os

from plycdn.storage import PlycdnLinkSigner, PlycdnStorageApi
from plycdn.video import PlycdnVideoApi, PlycdnVideoSigner

storage = PlycdnStorageApi(os.environ["PLYCDN_ADMIN_KEY"])
video = PlycdnVideoApi(storage)
video_signer = PlycdnVideoSigner(PlycdnLinkSigner(storage), video)


def playback(organization_id: str, library_id: str, video_id: str) -> str:
    # A protected recording must be signed by plycdn: use sign_video_from_plycdn for it.
    link = video_signer.sign_video(organization_id, library_id, video_id, ttl_seconds=3600)
    return link["url"]


def go_live(organization_id: str, library_id: str) -> str:
    live_input = video.create_live_input(organization_id, library_id, {"name": "Weekly town hall"})
    token = video.create_publish_token(organization_id, live_input["id"], ttl_seconds=600)
    return token["token"]

create_video returns the upload to complete; videos play as soon as their lowest quality is encoded and at every quality once ready (Video). sign_video signs in your process; sign_video_from_plycdn asks plycdn to sign and is required for a protected recording; sign_live signs a live input's playback link (mode is "broadcast" or "realtime"). Live inputs, destinations, share links, watch pages and domains are methods of PlycdnVideoApi (Live streaming).

7Routes for your players and uploaders

plycdn_router is the backend for the storage and video packages of Angular and React: listing, uploading, signing and playback. It uses the asynchronous API.

import os

from fastapi import FastAPI

from plycdn.fastapi import plycdn_router
from plycdn.storage import AsyncPlycdnLinkSigner, AsyncPlycdnStorageApi, PlycdnOperation
from plycdn.video import AsyncPlycdnVideoApi

storage = AsyncPlycdnStorageApi(os.environ["PLYCDN_API_KEY"])
WATCHING = (PlycdnOperation.ListObjects, PlycdnOperation.GetObject, PlycdnOperation.Sign,
            PlycdnOperation.GetVideo, PlycdnOperation.WatchVideo)


def roles_of(request):
    user = getattr(request.state, "user", None)
    return user.roles if user else []


def authorize(request, operation, bucket_id, key):
    return ("learner" in roles_of(request)) if operation in WATCHING else ("instructor" in roles_of(request))


def authorize_go_live(request, live_input_id, library_id):
    return "broadcaster" in roles_of(request)


def add_plycdn(app):
    app.include_router(
        plycdn_router(
            storage=storage,
            signer=AsyncPlycdnLinkSigner(storage),
            video=AsyncPlycdnVideoApi(storage),
            resolve_organization=lambda request: getattr(getattr(request.state, "user", None), "plycdn_organization_id", None),
            authorize=authorize,
            authorize_go_live=authorize_go_live,
            max_browser_link_ttl_seconds=3600,
        ),
        prefix="/api/plycdn/v1",
    )

The video routes exist only when you pass video. As with the content router, every route is refused until authorize returns exactly True, the organization comes only from resolve_organization, and going live is decided apart by authorize_go_live (refused when absent): being allowed to watch never lets anyone broadcast.

For a route addressed by an upload, file, video or live input, the router first looks the item up with your server key, then asks your hook, so a missing item answers 404 and a forbidden one 403 without revealing anything else. The sign route takes one target (a file, a video or a live input), signs for the request's own address on a bucket that binds links to the viewer, never issues a link longer than max_browser_link_ttl_seconds (at least 60), and adds tags only when your tags_for(request, body) supplies them (or you pass accept_browser_tags=True).

Browsers get these answers: 403 forbidden, 404 for a missing item, 422 validation_error for a request that is not valid (a body that is not application/json, over 1 MiB, nested more than 64 levels, containing NaN, Infinity or a number that is absurdly long, an out-of-range number or lifetime, or no usable viewer address on an address-bound bucket), and 502/503 provider_unavailable when plycdn cannot be reached; other refusals from plycdn are passed on with their own code. An error raised by your own hooks is not one of these: it reaches your application as its 500. The same advice applies as for the content router: run uvicorn with proxy headers behind a load balancer, keep your cookies SameSite and your CSRF protection.

8Webhooks

plycdn signs every delivery with the endpoint's secret. The receiver checks the signature and calls you with the event:

import os

from fastapi import FastAPI

from plycdn.fastapi import plycdn_webhook_router

seen = set()


async def on_event(event, request):
    if event["id"] in seen:           # a delivery can arrive twice: act on each event id once
        return
    seen.add(event["id"])
    if event["type"] == "video.ready":
        print("ready", event["data"])


def add_webhooks(app):
    secrets = [os.environ["PLYCDN_WEBHOOK_SECRET"], os.environ.get("PLYCDN_WEBHOOK_SECRET_PREVIOUS", "")]
    app.include_router(plycdn_webhook_router("/webhooks/plycdn", secrets=[s for s in secrets if s], on_event=on_event))

Answers: 200 after your on_event returns, 401 for a missing, stale or forged signature, 400 for a body that is too large (over max_body_bytes, default 1 MiB) or not an event, and 500 when your own code raises, which makes plycdn deliver again. Rules the check applies: the timestamp must be within tolerance_seconds (default 300) of now, past or future; there must be exactly one timestamp made of digits and at least one well-formed signature; a secret that is empty never matches; and a bad setting (a tolerance that is negative or not a number, a non-positive max_body_bytes, a single string where a list of secrets belongs) is refused with an error when your application starts, never silently widened. During a secret rotation, list both secrets: either one verifies. secrets can also be a function (plain or coroutine), so you can read them from your store at each delivery. On another server, verify by hand: PlycdnWebhookVerifier.verify(raw_body, signature_header, secrets, tolerance_seconds=300) returns True or False. Event names are in PlycdnWebhookEvents.

9S3 and Azure clients

The S3 and Azure tools you already use work against plycdn storage. These helpers build a client for your region from the list plycdn publishes, so a region added later works without a new package. The credentials are a storage credential from Developers → Storage credentials in the dashboard, never your API key.

import os

from plycdn.azure import PlycdnBlob
from plycdn.s3 import PlycdnS3
from plycdn.storage import PlycdnStorageApi

storage = PlycdnStorageApi(os.environ["PLYCDN_API_KEY"])


def clients(organization_id: str):
    s3 = PlycdnS3.create_client_from_regions(storage, organization_id, "mumbai",
                                             os.environ["PLYCDN_STORAGE_KEY_ID"], os.environ["PLYCDN_STORAGE_SECRET"])
    blob = PlycdnBlob.create_service_client_from_regions(storage, organization_id, os.environ["PLYCDN_STORAGE_ACCOUNT"],
                                                         os.environ["PLYCDN_STORAGE_ACCOUNT_KEY"], region="mumbai")
    return s3, blob

See the compatibility guide for what is supported and the limits.

10Errors

Content Search calls raise PlycdnApiError; storage and video calls raise PlycdnError. Both carry the same attributes:

Failures to reach plycdn use codes that already exist: content_unavailable (502), content_timeout (504) and invalid_content_response (502) for Content Search, and provider_unavailable for storage and video. The routers' own browser-safe answers are listed above; the error table in the API reference lists every code.

11Troubleshooting

• Every route answers 403. There is no authorize, or it returns something other than the boolean True. Return True explicitly.
• A route answers 403 organization_forbidden, or forbidden with a signed-in user. resolve_organization returned nothing for this request. It must return the plycdn organization id of the signed-in user.
• The uploads and the source callback answer 404. Pass storage, replay and a callback_secret of at least 32 characters together.
• Webhooks answer 401. The secret is the wrong one, or the server clock is more than five minutes off.
• A signed link for an address-bound bucket answers 422. The request has no IPv4 viewer address. Run uvicorn with proxy headers behind a load balancer.
• A router raises a TypeError at start-up. The routers use the asynchronous API (AsyncPlycdnContentApi, AsyncPlycdnStorageApi), not the synchronous one.
• A call fails with admin_key_required. Creating, changing or deleting buckets, libraries, domains and webhooks needs an admin key.