ninjavault-cdn

Typed Python client (sync and async) for the NinjaVault CDN Server: upload, search, download and share files.

Why ninjavault-cdn

• One client, every operation. Upload, list/search, metadata, usage summary, download, soft delete, public URLs, and single or batch presigned URLs.
• Sync and async, same surface. NinjaVaultCdnClient (on httpx.Client) and AsyncNinjaVaultCdnClient (on httpx.AsyncClient) have identical methods.
• No HTTP plumbing. Handles the X-Api-Key header, multipart fields, object-key encoding and the JSON success/error envelope for you.
• Typed errors. Every failure is a CdnApiError with the HTTP status, CDN error code, trace id, validation details and Retry-After. Network failures are CdnConnectionError / CdnTimeoutError.
• Typed models. Frozen dataclasses, py.typed, mypy --strict clean. A newer server never breaks an older SDK: unknown fields are ignored and unknown enum values are kept as strings.
• Streams large files. Downloads are streamed; download_to writes to disk chunk by chunk.
• Standalone, no opinions. Only depends on httpx. Logging, retries and correlation stay under your app's control.

Quick start

1. Install

pip install ninjavault-cdn

2. Configure (keep the key out of source control)

export NINJAVAULT_CDN_BASE_URL="https://cdn.example.com"
export NINJAVAULT_CDN_API_KEY="cdn_xxxxx"

3. Use it (sync)

from pathlib import Path
from ninjavault_cdn import NinjaVaultCdnClient

with NinjaVaultCdnClient() as cdn:                  # reads NINJAVAULT_CDN_* env vars
    access = cdn.get_access_context()               # verifies the key
    uploaded = cdn.upload("documents", 42, Path("invoice.pdf"), content_type="application/pdf")
    print(uploaded.bucket, uploaded.object_key)     # store these in your database
    link = cdn.create_presigned_url("documents", uploaded.object_key, expiry_seconds=600)
    print(link.url)                                 # safe to hand to a browser

...or async

import asyncio
from ninjavault_cdn import AsyncNinjaVaultCdnClient

async def main() -> None:
    async with AsyncNinjaVaultCdnClient() as cdn:
        uploaded = await cdn.upload("documents", 42, b"%PDF-1.7 ...", "invoice.pdf", "application/pdf")
        page = await cdn.list_files(bucket="documents", tenant_id=42, page_size=20)
        print(uploaded.object_key, page.total_count)

asyncio.run(main())

That's it. Everything below is optional.

Configuration reference

Pass keyword arguments, or leave them out to use environment variables:

cdn = NinjaVaultCdnClient(
    base_url="https://cdn.example.com",
    api_key=settings.cdn_api_key,
    public_base_url="https://static.example.com",   # optional
    timeout=30.0,                                   # optional
)

Explicit arguments win over environment variables. A missing base_url or api_key raises ValueError naming the argument and its environment variable. Create one client per process and reuse it: it holds a connection pool.

Verify the key at startup (recommended)

access = cdn.get_access_context()
# access.allowed_buckets / access.allowed_tenant_ids empty = unrestricted, not "no access"

Tenant ids are int when numeric (the server sends them as strings, e.g. ["1"]) and stay str for GUID tenants.

Usage

All examples use the sync client. With AsyncNinjaVaultCdnClient, await the same calls.

Upload

uploaded = cdn.upload(
    "documents",                 # bucket
    42,                          # tenant id: int or GUID string
    Path("report.docx"),         # bytes, a binary file object, or a pathlib.Path
    file_name="report.docx",     # defaults to the Path's name
    content_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
    owner_id=1001,               # optional
    folder_path="reports/2026",  # optional: up to 8 segments of [A-Za-z0-9_-]
)

# uploaded.bucket + uploaded.object_key identify the file from now on (not the original file name).

File objects are streamed, not read into memory. When content_type is omitted it is guessed from the file name. Allowed content types and the maximum size are set per CDN deployment (commonly PDF, JPEG, PNG, WEBP, DOC and DOCX, up to about 25 MB). Anything outside them fails with 40001.

Give a browser or mobile app access

logo_url = cdn.build_public_url("public-assets", object_key)

link = cdn.create_presigned_url("documents", object_key, expiry_seconds=600)
print(link.url, link.expires_at_utc)

batch = cdn.create_presigned_urls([("documents", key_a), ("documents", key_b)], expiry_seconds=600)
for failed in batch.failed:          # batch returns 200 even if some targets fail
    log.warning("%s: %s", failed.object_key, failed.reason)

Omit expiry_seconds to use the server default; the server clamps it to its configured maximum.

Never send the API key to a browser or mobile app. Give them a public or presigned URL instead.

Download (backend to backend)

with cdn.download("documents", object_key) as download:
    print(download.content_type, download.file_name, download.size_bytes)
    for chunk in download.iter_bytes():
        sink.write(chunk)

written = cdn.download_to("documents", object_key, "/tmp/report.pdf")  # path or binary file object

with cdn.download("documents", object_key) as download:                # small files only
    data = download.read()

Stream straight through a web framework without loading the file into memory, for example with FastAPI:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from ninjavault_cdn import AsyncNinjaVaultCdnClient

app = FastAPI()
cdn = AsyncNinjaVaultCdnClient()

@app.get("/files/{object_key:path}")
async def get_file(object_key: str) -> StreamingResponse:
    download = await cdn.download("documents", object_key)

    async def body():
        try:
            async for chunk in download.aiter_bytes():
                yield chunk
        finally:
            await download.aclose()

    return StreamingResponse(
        body(),
        media_type=download.content_type or "application/octet-stream",
        headers={"Content-Disposition": f'attachment; filename="{download.file_name or "file"}"'},
    )

With the async client, use either async with cdn.download(...) as download: or download = await cdn.download(...) and close it yourself with await download.aclose().

Search and list

from datetime import datetime, timedelta, timezone
from ninjavault_cdn import FileCategory, FileListQuery, FileSortBy

page = cdn.list_files(FileListQuery(
    bucket="documents",
    tenant_id=42,
    owner_id=1001,
    object_key_prefix="reports/2026",
    file_name_contains="invoice",
    category=FileCategory.DOCUMENT,
    created_from_utc=datetime.now(timezone.utc) - timedelta(days=30),
    sort_by=FileSortBy.CREATED_AT_UTC,
    sort_descending=True,
    page=1,
    page_size=50,                    # max 200
))

page = cdn.list_files(bucket="documents", page=2)   # keyword filters work too

for item in page.items:
    print(item.object_key, item.size_bytes, item.created_at_utc)

Metadata, usage, buckets, delete

info = cdn.get_metadata("documents", object_key)   # checksum, category, thumbnail...
usage = cdn.get_summary("documents")               # totals per category (omit the bucket for all)
buckets = cdn.list_buckets()                       # visibility + counters
cdn.delete("documents", object_key)                # soft delete

Error handling

import time
from ninjavault_cdn import CdnApiError, CdnConnectionError, ErrorCode

try:
    cdn.upload("documents", 42, data, "report.pdf")
except CdnApiError as ex:
    if ex.error_code == ErrorCode.TOO_MANY_REQUESTS and ex.retry_after is not None:
        time.sleep(ex.retry_after)            # back off exactly as long as the server asks
    else:
        log.warning("CDN %s %s: %s (trace %s)", ex.status_code, ex.error_code, ex.message, ex.correlation_id)
        for field, errors in (ex.details or {}).items():
            log.warning("%s: %s", field, ", ".join(errors))
except CdnConnectionError:                    # DNS, TLS, refused, timeouts (CdnTimeoutError)
    log.exception("CDN unreachable")

All SDK exceptions derive from NinjaVaultCdnError. CdnConnectionError and CdnTimeoutError wrap the original httpx exception (available as __cause__). Invalid arguments (blank bucket or object key, missing file name) raise ValueError / TypeError before any request is sent.

Logging, correlation and retries

The SDK logs one DEBUG line per request on the ninjavault_cdn logger (method, URL, status, elapsed ms). It never logs the API key, headers or file bytes, and it adds no handlers: whether and where these lines go is your app's choice.

import logging
logging.getLogger("ninjavault_cdn").setLevel(logging.DEBUG)

For your own request logging, metrics or an X-Correlation-Id header, use httpx event hooks:

import uuid

def add_correlation_id(request):
    request.headers["X-Correlation-Id"] = current_correlation_id() or str(uuid.uuid4())

def log_response(response):
    log.info("CDN %s %s -> %s", response.request.method, response.request.url, response.status_code)

cdn = NinjaVaultCdnClient(event_hooks={"request": [add_correlation_id], "response": [log_response]})

With AsyncNinjaVaultCdnClient, hooks must be async def. If you pass your own http_client, configure hooks, proxies, retries (for example httpx.HTTPTransport(retries=3)) on it instead; passing both http_client and event_hooks raises ValueError.

Every request carries the X-Api-Key header. If your hook logs headers, redact it first.

Retries

The SDK never retries on its own, so a retry can't upload a file twice behind your back. Two easy options:

import httpx
from ninjavault_cdn import NinjaVaultCdnClient

# Retry failed connections (DNS, refused, TLS) up to 3 times
cdn = NinjaVaultCdnClient(http_client=httpx.Client(transport=httpx.HTTPTransport(retries=3), timeout=30))

For rate limits (42901), wait ex.retry_after seconds and try again, as in the error-handling example above.

Unit testing your code

Swap the transport with httpx.MockTransport to exercise your code without a network:

import httpx
from ninjavault_cdn import NinjaVaultCdnClient

def handler(request: httpx.Request) -> httpx.Response:
    assert request.headers["X-Api-Key"] == "cdn_test"
    return httpx.Response(200, json={"success": True, "data": [], "error": None})

cdn = NinjaVaultCdnClient(
    base_url="https://cdn.test",
    api_key="cdn_test",
    http_client=httpx.Client(transport=httpx.MockTransport(handler)),
)
assert cdn.list_buckets() == []

Or depend on the client in your own code and mock it:

from unittest.mock import create_autospec
from ninjavault_cdn import NinjaVaultCdnClient

cdn = create_autospec(NinjaVaultCdnClient, instance=True)
cdn.build_public_url.return_value = "https://cdn.test/public/public-assets/logo.png"
service = BrandingService(cdn)

Troubleshooting

Compatibility

NinjaVault SDKs

The same CDN client in every language, with the same features and the same version number (for example 100.42.1 everywhere):

Links

• Source and issues: https://github.com/alisaivi786/NinjaVault-Python
• Release notes (change-sets): https://github.com/alisaivi786/NinjaVault-Python/tree/main/changesets
• Releasing: docs/RELEASING.md
• Security: SECURITY.md
• License: MIT