Python SDK for content moderation

Official Python SDK for VideoCensor API. Analyze and censor text, audio, and video — profanity, hate speech, extremism, drugs, sexual content. Sync + async clients, typed models, auto-retry with exponential backoff.

pip install videocensor SDK docs

Sync + async clients

VideoCensor (synchronous, httpx) and AsyncVideoCensor (async/await). Same API, same models. Pick by framework — FastAPI, Django, Flask, Celery.

Typed models

Pydantic-like dataclasses for every endpoint. Full mypy/pyright support. IDE autocomplete out of the box.

Auto-retry with backoff

Retries on 429/5xx with exponential backoff. max_retries in constructor. Retry-After header respected.

File upload from path or stream

analyze_media accepts a file path, BufferedReader, or BytesIO. No need to explicitly open the file — the SDK handles it.

wait_for_job helper

Helper for polling until job completion: client.wait_for_job(job_id, timeout=600, poll_interval=2). Supports progress callback.

Context manager

with VideoCensor(api_key=...) as client — automatic httpx client close. Works async too: async with AsyncVideoCensor(...).

Quick start

Sync — text

import os
from videocensor import VideoCensor

with VideoCensor(api_key=os.environ["VIDEOCENSOR_API_KEY"]) as client:
    result = client.analyze_text("text to check")
    print(result.flagged_count, result.categories)

Async — video

import os
import asyncio
from videocensor import AsyncVideoCensor

async def main():
    async with AsyncVideoCensor(api_key=os.environ["VIDEOCENSOR_API_KEY"]) as client:
        job = await client.analyze_media("/path/to/video.mp4")
        done = await client.wait_for_job(job.id)
        result = await client.get_job_result(done.id)
        print(result.censored_count)

asyncio.run(main())

Censor media

import os
from videocensor import VideoCensor

with VideoCensor(api_key=os.environ["VIDEOCENSOR_API_KEY"]) as client:
    job = client.censor_media("/path/to/video.mp4", way_of_blocking="beep")
    done = client.wait_for_job(job.id)
    client.download_job(done.id, "./clean.mp4")

Error handling

from videocensor import VideoCensor, RateLimitError, AuthenticationError

try:
    result = client.analyze_text("test")
except RateLimitError as e:
    print(f"retry after {e.retry_after}s")
except AuthenticationError:
    print("invalid api key")

What the SDK can do

FastAPI backend
async endpoint → AsyncVideoCensor → return job_id → webhook. Or wait_for_job for sync response.
Django admin
Sync VideoCensor inside an admin action for manual moderation of uploaded files.
Celery tasks
wait_for_job inside a Celery task — great for batch processing of long files.
Jupyter notebook
Data-science analysis of text/video corpus — all through sync API, inline results.
CLI tool
Script for batch file processing: glob + analyze_media + write CSV results.
Airflow DAG
DAG for auto-moderation of new files from S3/GCS — task_instance calls VideoCensor.

Python SDK — FAQ

Which Python version is supported?+

Python 3.10+. Uses modern type hints (X | None) and dataclasses.

What dependencies does the SDK have?+

Just httpx >= 0.25. No heavy dependencies. Package size ~30 KB.

Does it work on AWS Lambda / Cloud Functions?+

Yes. No subprocesses, long-lived processes, or local files required. Use webhooks instead of wait_for_job for long tasks.

Is there built-in telemetry?+

No. The SDK sends no telemetry — only direct calls to videocensor.ru/api/v1.

How do I update the SDK?+

pip install --upgrade videocensor. We follow SemVer — breaking changes only in major versions, with 3-month deprecation notice.

Where's the source code?+

Open source on GitHub — github.com/dzhumaevn/videocensor/tree/main/sdk/python. MIT license.

Install the SDK in seconds

pip install videocensor — a typed client with sync and async versions.

pip install videocensor SDK docs

Quick start

Sync — text

import os
from videocensor import VideoCensor

with VideoCensor(api_key=os.environ["VIDEOCENSOR_API_KEY"]) as client:
    result = client.analyze_text("text to check")
    print(result.flagged_count, result.categories)

Async — video

import os
import asyncio
from videocensor import AsyncVideoCensor

async def main():
    async with AsyncVideoCensor(api_key=os.environ["VIDEOCENSOR_API_KEY"]) as client:
        job = await client.analyze_media("/path/to/video.mp4")
        done = await client.wait_for_job(job.id)
        result = await client.get_job_result(done.id)
        print(result.censored_count)

asyncio.run(main())

Censor media

import os
from videocensor import VideoCensor

with VideoCensor(api_key=os.environ["VIDEOCENSOR_API_KEY"]) as client:
    job = client.censor_media("/path/to/video.mp4", way_of_blocking="beep")
    done = client.wait_for_job(job.id)
    client.download_job(done.id, "./clean.mp4")

Error handling

from videocensor import VideoCensor, RateLimitError, AuthenticationError

try:
    result = client.analyze_text("test")
except RateLimitError as e:
    print(f"retry after {e.retry_after}s")
except AuthenticationError:
    print("invalid api key")

What the SDK can do

FastAPI backend

async endpoint → AsyncVideoCensor → return job_id → webhook. Or wait_for_job for sync response.

Django admin

Sync VideoCensor inside an admin action for manual moderation of uploaded files.

Celery tasks

wait_for_job inside a Celery task — great for batch processing of long files.

Jupyter notebook

Data-science analysis of text/video corpus — all through sync API, inline results.

CLI tool

Script for batch file processing: glob + analyze_media + write CSV results.

Airflow DAG

DAG for auto-moderation of new files from S3/GCS — task_instance calls VideoCensor.