Examples¶
Complete programs from examples/. Run them from the repository root, e.g. uv run python examples/basic.py; each module docstring lists curl commands to try. Every example is type-checked and covered by the test suite.
Basic¶
Standalone connector configured from a JSON file.
"""Standalone connector configured from a JSON file.
Run from the repository root::
uv run python examples/basic.py
curl "http://localhost:8081/?action=files"
"""
from pathlib import Path
import uvicorn
from fastapi import FastAPI
from jcpy import create_app
CONFIG = Path(__file__).parent / "config" / "basic.json"
def build_app() -> FastAPI:
"""Build the application.
Returns:
Connector application; every request gets ``defaultRole``.
"""
return create_app(CONFIG)
if __name__ == "__main__":
uvicorn.run(build_app(), host="127.0.0.1", port=8081)
Cookie authentication¶
Role taken from a cookie on every request.
"""Role taken from a cookie on every request.
The cookie is set by the client, so this only suits trusted
environments; sign it or keep the role server side (see
``session_auth.py``) in production.
Run from the repository root::
uv run python examples/cookie_auth.py
curl "http://localhost:8081/?action=permissions&source=uploads"
curl -H "Cookie: userRole=editor" \\
"http://localhost:8081/?action=permissions&source=uploads"
curl -X POST -H "Cookie: userRole=admin" -F "files[0]=@README.md" \\
"http://localhost:8081/?action=fileUpload&source=uploads"
"""
from pathlib import Path
import uvicorn
from fastapi import FastAPI, Request
from jcpy import create_app
CONFIG = Path(__file__).parent / "config" / "roles.json"
ROLE_COOKIE = "userRole"
def role_from_cookie(request: Request) -> str:
"""Read the role of the user from the ``userRole`` cookie.
Args:
request: Incoming request.
Returns:
Cookie value, ``guest`` when the cookie is missing.
"""
return request.cookies.get(ROLE_COOKIE) or "guest"
def build_app() -> FastAPI:
"""Build the application.
Returns:
Connector application authenticating by cookie.
"""
return create_app(CONFIG, check_authentication=role_from_cookie)
if __name__ == "__main__":
uvicorn.run(build_app(), host="127.0.0.1", port=8081)
{
"defaultRole": "guest",
"accessControl": [
{"role": "guest", "FILES": true, "FILE_UPLOAD": false, "FILE_REMOVE": false},
{"role": "editor", "FILES": true, "FILE_UPLOAD": true, "FILE_REMOVE": false},
{"role": "admin", "FILES": true, "FILE_UPLOAD": true, "FILE_REMOVE": true}
],
"sources": {
"uploads": {
"title": "Uploads",
"root": "./files/uploads",
"baseurl": "http://localhost:8080/files/uploads/"
}
}
}
JWT authentication¶
Role from a signed JWT (PyJWT); bad tokens get 401.
"""Role taken from a signed JWT in the ``Authorization`` header.
Tokens are HS256-signed with ``JWT_SECRET`` and must carry ``role``
and ``exp`` claims. Requests without a token are guests; a bad token
is answered with ``401``.
Run from the repository root (prints demo tokens)::
JWT_SECRET=$(openssl rand -hex 32) uv run python examples/jwt_auth.py
curl -H "Authorization: Bearer <token>" \\
"http://localhost:8081/?action=permissions&source=uploads"
"""
import os
import time
from http import HTTPStatus
from pathlib import Path
import jwt
import uvicorn
from fastapi import FastAPI, Request
from jcpy import HttpError, create_app
CONFIG = Path(__file__).parent / "config" / "roles.json"
SECRET = os.environ.get("JWT_SECRET", "change-me-to-a-long-random-secret")
ALGORITHM = "HS256"
TOKEN_LIFETIME = 3600
def make_token(
role: str, username: str, lifetime: int = TOKEN_LIFETIME
) -> str:
"""Issue a token, as a login endpoint would.
Args:
role: Role of the user.
username: Name of the user.
lifetime: Seconds until the token expires.
Returns:
Signed token.
"""
payload = {
"role": role,
"username": username,
"exp": int(time.time()) + lifetime,
}
return jwt.encode(payload, SECRET, algorithm=ALGORITHM)
def role_from_jwt(request: Request) -> str:
"""Verify the bearer token and read the role from it.
Args:
request: Incoming request.
Returns:
``role`` claim of the token, ``guest`` without a token.
Raises:
HttpError: ``401`` for a malformed, forged or expired token.
"""
header = request.headers.get("authorization")
if not header:
return "guest"
scheme, _, token = header.partition(" ")
unauthorized = HttpError(
HTTPStatus.UNAUTHORIZED, "Invalid or expired token"
)
if scheme.lower() != "bearer" or not token:
raise unauthorized
try:
payload = jwt.decode(
token,
SECRET,
algorithms=[ALGORITHM],
options={"require": ["exp", "role"]},
)
except jwt.InvalidTokenError as error:
raise unauthorized from error
role = payload["role"]
if not isinstance(role, str):
raise unauthorized
return role
def build_app() -> FastAPI:
"""Build the application.
Returns:
Connector application authenticating by JWT.
"""
return create_app(CONFIG, check_authentication=role_from_jwt)
if __name__ == "__main__":
for demo_role in ("editor", "admin"):
print(f"{demo_role}: {make_token(demo_role, demo_role)}")
uvicorn.run(build_app(), host="127.0.0.1", port=8081)
{
"defaultRole": "guest",
"accessControl": [
{"role": "guest", "FILES": true, "FILE_UPLOAD": false, "FILE_REMOVE": false},
{"role": "editor", "FILES": true, "FILE_UPLOAD": true, "FILE_REMOVE": false},
{"role": "admin", "FILES": true, "FILE_UPLOAD": true, "FILE_REMOVE": true}
],
"sources": {
"uploads": {
"title": "Uploads",
"root": "./files/uploads",
"baseurl": "http://localhost:8080/files/uploads/"
}
}
}
Session authentication¶
Role kept in a signed Starlette session, with login and logout routes.
"""Role kept in a server-signed session, like PHP ``$_SESSION``.
The connector is mounted into an application that owns the session
middleware and the login routes. The login route below trusts its
argument and only stands in for a real sign-in.
Run from the repository root::
SESSION_SECRET=change-me uv run python examples/session_auth.py
curl -c cookies.txt "http://localhost:8081/login/editor"
curl -b cookies.txt \\
"http://localhost:8081/?action=permissions&source=uploads"
curl -b cookies.txt "http://localhost:8081/logout"
"""
import os
from pathlib import Path
from typing import Literal
import uvicorn
from fastapi import FastAPI, Request
from starlette.middleware.sessions import SessionMiddleware
from jcpy import create_router
CONFIG = Path(__file__).parent / "config" / "roles.json"
SECRET = os.environ.get("SESSION_SECRET", "change-me")
SESSION_MAX_AGE = 24 * 60 * 60
def role_from_session(request: Request) -> str:
"""Read the role stored in the session.
Args:
request: Incoming request.
Returns:
Stored role, ``guest`` before login.
"""
role = request.session.get("userRole")
return role if isinstance(role, str) else "guest"
def build_app() -> FastAPI:
"""Build the application.
Returns:
Application with session routes and the connector at ``/``.
"""
app = FastAPI(openapi_url=None, docs_url=None, redoc_url=None)
app.add_middleware(
SessionMiddleware, secret_key=SECRET, max_age=SESSION_MAX_AGE
)
@app.get("/login/{role}")
def login(
request: Request, role: Literal["guest", "editor", "admin"]
) -> dict[str, str]:
request.session["userRole"] = role
return {"role": role}
@app.get("/logout")
def logout(request: Request) -> dict[str, str]:
request.session.clear()
return {"role": "guest"}
@app.get("/whoami")
def whoami(request: Request) -> dict[str, str]:
return {"role": role_from_session(request)}
# Own routes first: the connector serves every other /{action}.
app.include_router(
create_router(CONFIG, check_authentication=role_from_session)
)
return app
if __name__ == "__main__":
uvicorn.run(build_app(), host="127.0.0.1", port=8081)
{
"defaultRole": "guest",
"accessControl": [
{"role": "guest", "FILES": true, "FILE_UPLOAD": false, "FILE_REMOVE": false},
{"role": "editor", "FILES": true, "FILE_UPLOAD": true, "FILE_REMOVE": false},
{"role": "admin", "FILES": true, "FILE_UPLOAD": true, "FILE_REMOVE": true}
],
"sources": {
"uploads": {
"title": "Uploads",
"root": "./files/uploads",
"baseurl": "http://localhost:8080/files/uploads/"
}
}
}
Custom SVG icons¶
Thumbnails of folders and non-image files drawn by your function.
"""Custom SVG thumbnails for folders and non-image files.
Run from the repository root::
uv run python examples/custom_svg.py
curl "http://localhost:8081/?action=files"
"""
from pathlib import Path, PurePosixPath
from typing import TYPE_CHECKING
from xml.sax.saxutils import escape
import uvicorn
from fastapi import FastAPI
from jcpy import create_app
if TYPE_CHECKING:
from jcpy.storage.base import StatEntry
CONFIG = Path(__file__).parent / "config" / "svg.json"
COLORS = {
".pdf": "#e74c3c",
".doc": "#3498db",
".docx": "#3498db",
".txt": "#95a5a6",
".zip": "#f39c12",
".tar": "#f39c12",
".gz": "#f39c12",
".json": "#9b59b6",
".xml": "#9b59b6",
}
MAX_NAME = 12
def colored_icon(entry: StatEntry, width: int, height: int) -> str:
"""Render a tile colored by file type with the extension and name.
Args:
entry: Folder or file the icon stands for.
width: Icon width in pixels.
height: Icon height in pixels.
Returns:
SVG document.
"""
path = PurePosixPath(entry.path)
ext = path.suffix.lower()
color = "#2ecc71" if entry.is_directory else COLORS.get(ext, "#7f8c8d")
label = "DIR" if entry.is_directory else ext.removeprefix(".").upper()
name = path.name
if len(name) > MAX_NAME:
name = f"{name[:MAX_NAME]}..."
# File names are user input: escape them inside the markup.
return (
f'<svg width="{width}" height="{height}" viewBox="0 0 100 100" '
'xmlns="http://www.w3.org/2000/svg">'
f'<rect width="100" height="100" fill="{color}" rx="8"/>'
'<text x="50" y="40" text-anchor="middle" fill="white" '
'font-family="Arial" font-size="20" font-weight="bold">'
f"{escape(label)}</text>"
'<text x="50" y="70" text-anchor="middle" fill="white" '
'font-family="Arial" font-size="10" opacity="0.8">'
f"{escape(name)}</text></svg>"
)
def build_app() -> FastAPI:
"""Build the application.
Returns:
Connector application with the custom icon generator.
"""
return create_app(CONFIG, svg_generator=colored_icon)
if __name__ == "__main__":
uvicorn.run(build_app(), host="127.0.0.1", port=8081)
Several instances¶
A public read-only connector and an admin one in the same application.
"""Two independent connectors in one application.
``/public`` lets everyone browse and nothing else; ``/admin`` needs the
``X-Admin-Token`` header. Each instance has its own configuration,
authentication and caches.
Run from the repository root::
ADMIN_TOKEN=secret uv run python examples/multi_instance.py
curl "http://localhost:8081/public/files"
curl -H "X-Admin-Token: secret" "http://localhost:8081/admin/files"
"""
import os
import secrets
from pathlib import Path
import uvicorn
from fastapi import FastAPI, Request
from jcpy import create_router
CONFIG_DIR = Path(__file__).parent / "config"
ADMIN_TOKEN = os.environ.get("ADMIN_TOKEN", "change-me")
def public_role(_request: Request) -> str:
"""Treat every visitor of the public connector as a guest.
Returns:
``guest``.
"""
return "guest"
def admin_role(request: Request) -> str:
"""Grant ``admin`` to requests with the right token.
Args:
request: Incoming request.
Returns:
``admin`` or ``anonymous``.
"""
token = request.headers.get("x-admin-token", "")
if secrets.compare_digest(token.encode(), ADMIN_TOKEN.encode()):
return "admin"
return "anonymous"
def build_app() -> FastAPI:
"""Build the application.
Returns:
Application serving both connectors.
"""
app = FastAPI(openapi_url=None, docs_url=None, redoc_url=None)
app.include_router(
create_router(
CONFIG_DIR / "public.json", check_authentication=public_role
),
prefix="/public",
)
app.include_router(
create_router(
CONFIG_DIR / "admin.json", check_authentication=admin_role
),
prefix="/admin",
)
return app
if __name__ == "__main__":
uvicorn.run(build_app(), host="127.0.0.1", port=8081)
S3¶
Files in an S3 bucket.
"""Files stored in an S3 bucket.
Credentials come from the usual AWS sources (environment variables,
shared config, instance role). For MinIO or another S3-compatible
service set ``endpoint`` and ``forcePathStyle`` in the ``s3`` block.
Run from the repository root::
AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... \\
uv run python examples/s3.py
"""
from pathlib import Path
import uvicorn
from fastapi import FastAPI
from jcpy import create_app
CONFIG = Path(__file__).parent / "config" / "s3.json"
def build_app() -> FastAPI:
"""Build the application.
Returns:
Connector application backed by S3.
"""
return create_app(CONFIG)
if __name__ == "__main__":
uvicorn.run(build_app(), host="127.0.0.1", port=8081)
{
"debug": false,
"allowCrossOrigin": true,
"createThumb": true,
"thumbSize": 250,
"quality": 90,
"sources": {
"media": {
"title": "Media library",
"baseurl": "https://my-bucket.s3.eu-central-1.amazonaws.com/media/",
"storageAdapter": "s3",
"s3": {
"bucket": "my-bucket",
"region": "eu-central-1",
"prefix": "media"
}
}
}
}
Multi-tenant¶
Sources picked per request from a header.
"""Sources picked per request (one set of folders per tenant).
The tenant comes from the ``X-Tenant`` header; in a real application
it would come from the authenticated user, and the settings from a
database. Requests without a tenant use the configured ``shared``
source. Built sources are cached per tenant id (see
``dynamicSourcesCache``); putting the settings version into the id
rebuilds them after a change.
Run from the repository root::
uv run python examples/multi_tenant.py
curl -H "X-Tenant: acme" "http://localhost:8081/?action=files"
"""
from dataclasses import dataclass
from pathlib import Path
import uvicorn
from fastapi import FastAPI, Request
from jcpy import HttpError, ResolvedSources, create_app
CONFIG = Path(__file__).parent / "config" / "tenants.json"
@dataclass(frozen=True, slots=True)
class Tenant:
"""Stored tenant settings.
Attributes:
title: Title of the tenant's source.
version: Changes whenever the settings change.
"""
title: str
version: int
TENANTS = {
"acme": Tenant("ACME files", 1),
"globex": Tenant("Globex files", 3),
}
def tenant_sources(request: Request) -> ResolvedSources | None:
"""Pick the sources of the request's tenant.
Args:
request: Incoming request.
Returns:
Tenant sources, ``None`` for requests without a tenant.
Raises:
HttpError: ``403`` for an unknown tenant.
"""
name = request.headers.get("x-tenant")
if name is None:
return None
tenant = TENANTS.get(name)
if tenant is None:
raise HttpError.forbidden("Unknown tenant")
return ResolvedSources(
id=f"{name}:{tenant.version}",
sources={
"files": {
"title": tenant.title,
"root": f"./files/tenants/{name}",
"baseurl": f"http://localhost:8080/files/tenants/{name}/",
}
},
)
def build_app() -> FastAPI:
"""Build the application.
Returns:
Connector application with per-tenant sources.
"""
return create_app(CONFIG, resolve_sources=tenant_sources)
if __name__ == "__main__":
uvicorn.run(build_app(), host="127.0.0.1", port=8081)
Custom storage adapter¶
An in-memory adapter registered by name.
"""Custom storage adapter, registered by name.
Files live in process memory, so they vanish on restart; the adapter
shows the interface a real backend (a database, another object store)
implements. Sources select it with ``"storageAdapter": "memory"``.
Run from the repository root::
uv run python examples/custom_storage.py
curl -F "files[0]=@README.md" "http://localhost:8081/fileUpload"
curl "http://localhost:8081/files"
"""
import time
from pathlib import Path
from typing import TYPE_CHECKING
import uvicorn
from fastapi import FastAPI
from jcpy import create_app, register_storage_adapter
from jcpy.storage.base import FileWasNotFoundError, StatEntry, StorageError
if TYPE_CHECKING:
from collections.abc import AsyncIterator
from jcpy import SourceConfig
CONFIG = Path(__file__).parent / "config" / "memory.json"
class MemoryStorageAdapter:
"""Keep files in a dict; paths are relative and use ``/``."""
def __init__(self) -> None:
self.files: dict[str, tuple[bytes, float]] = {}
self.directories: set[str] = {""}
def _add_parents(self, path: str) -> None:
parts = path.split("/")
for index in range(1, len(parts)):
self.directories.add("/".join(parts[:index]))
async def write(self, path: str, contents: bytes) -> None:
"""Create or replace a file."""
self._add_parents(path)
self.files[path] = (contents, time.time() * 1000)
async def read(self, path: str) -> bytes:
"""Read a whole file."""
if path not in self.files:
raise FileWasNotFoundError(path)
return self.files[path][0]
async def delete_file(self, path: str) -> None:
"""Delete a file; a missing file is not an error."""
self.files.pop(path, None)
async def create_directory(self, path: str) -> None:
"""Create a directory with its parents."""
self._add_parents(path)
self.directories.add(path)
async def delete_directory(self, path: str) -> None:
"""Delete a directory recursively."""
prefix = f"{path}/"
self.files = {
key: value
for key, value in self.files.items()
if not key.startswith(prefix)
}
self.directories = {
item
for item in self.directories
if item != path and not item.startswith(prefix)
}
async def stat(self, path: str) -> StatEntry:
"""Read size and modification time."""
if path in self.files:
contents, modified = self.files[path]
return StatEntry(path, True, len(contents), modified)
if path in self.directories:
return StatEntry(path, False)
msg = f"Unable to get stat. Reason: {path} does not exist"
raise StorageError(msg)
async def list(self, path: str, *, deep: bool) -> AsyncIterator[StatEntry]:
"""List a directory (and its sub-directories when ``deep``)."""
prefix = f"{path}/" if path else ""
entries = [
StatEntry(item, is_file=False)
for item in sorted(self.directories)
if item and item.startswith(prefix)
] + [
StatEntry(item, is_file=True)
for item in sorted(self.files)
if item.startswith(prefix)
]
for entry in entries:
if deep or "/" not in entry.path.removeprefix(prefix):
yield entry
async def file_exists(self, path: str) -> bool:
"""Tell whether a file exists."""
return path in self.files
async def directory_exists(self, path: str) -> bool:
"""Tell whether a directory exists."""
return path in self.directories
async def copy_file(self, source: str, destination: str) -> None:
"""Copy a file."""
await self.write(destination, await self.read(source))
async def move_file(self, source: str, destination: str) -> None:
"""Move a file or a whole directory."""
if source in self.files:
self._add_parents(destination)
self.files[destination] = self.files.pop(source)
return
prefix = f"{source}/"
for key in [key for key in self.files if key.startswith(prefix)]:
await self.write(
f"{destination}/{key.removeprefix(prefix)}",
self.files.pop(key)[0],
)
for item in sorted(self.directories):
if item == source or item.startswith(prefix):
self.directories.discard(item)
await self.create_directory(
destination + item.removeprefix(source)
)
_STORES: dict[str, MemoryStorageAdapter] = {}
def memory_adapter(source: SourceConfig) -> MemoryStorageAdapter:
"""Give every source its own store, kept between requests.
Args:
source: Settings of the source using the adapter.
Returns:
Store of the source.
"""
return _STORES.setdefault(source.name, MemoryStorageAdapter())
def build_app() -> FastAPI:
"""Build the application.
Returns:
Connector application with in-memory sources.
"""
register_storage_adapter("memory", memory_adapter)
return create_app(CONFIG)
if __name__ == "__main__":
uvicorn.run(build_app(), host="127.0.0.1", port=8081)