granite_assets.models

Data transfer objects (DTOs) and configuration dataclasses used throughout the library.

Domain models and configuration dataclasses for granite-assets.

All models use slots=True for memory efficiency and to prevent accidental attribute creation. frozen=True is applied to value objects that should be immutable after construction; mutable builder-style objects stay unfrozen.

class granite_assets.models.AssetSaveRequest(source: BinaryIO | bytes, content_type: str, key: str | None = None, visibility: AssetVisibility = AssetVisibility.PRIVATE, filename: str | None = None, content_length: int | None = None, checksum: str | None = None, metadata: dict[str, str]=<factory>, overwrite: bool = True)[source]

Bases: object

All the information needed to persist a new asset.

Variables:

source (BinaryIO | bytes) – Readable binary stream with the asset content.
content_type (str) – MIME type (e.g. "application/pdf").
key (str | None) – Logical key that uniquely identifies the asset within the repository (e.g. "invoices/2024/inv-001.pdf"). It must not contain a leading slash. When None (default) the repository auto-generates a namespaced key of the form <uuid>/<uuid>.<ext>, where the extension is derived from filename. The resolved key is always returned in AssetSaveResult.
visibility (granite_assets.enums.AssetVisibility) – Whether the asset will be publicly accessible or private.
filename (str | None) – Original human-readable filename, stored as metadata. Required when key is omitted (used to derive the file extension for the auto-generated key).
content_length (int | None) – Byte size of the asset when known; used to set Content-Length headers on upload.
checksum (str | None) – Optional integrity hash (e.g. "md5:abc123").
metadata (dict[str, str]) – Arbitrary key/value pairs forwarded to the backend (e.g. S3 object metadata or extended file attributes).
overwrite (bool) – If False and the key already exists the backend should raise AssetError. Implementations may expose a per-request override even when the global config differs.

source: BinaryIO | bytes

content_type: str

key: str | None

visibility: AssetVisibility

filename: str | None

content_length: int | None

checksum: str | None

metadata: dict[str, str]

overwrite: bool

open_source() → BinaryIO[source]: Return a readable binary stream regardless of whether source is already a stream or a raw bytes object.

__init__(source: BinaryIO | bytes, content_type: str, key: str | None = None, visibility: AssetVisibility = AssetVisibility.PRIVATE, filename: str | None = None, content_length: int | None = None, checksum: str | None = None, metadata: dict[str, str]=<factory>, overwrite: bool = True) → None

class granite_assets.models.AssetSaveResult(key: str, backend_ref: str, content_length: int | None = None, checksum: str | None = None, visibility: AssetVisibility = AssetVisibility.PRIVATE)[source]

Bases: object

Returned by IAssetRepository.save() after a successful write.

Variables:

key (str) – The logical key under which the asset was stored.
backend_ref (str) – Backend-specific identifier (e.g. S3 ETag, local path).
content_length (int | None) – Byte size as recorded by the backend.
checksum (str | None) – Integrity hash as returned or calculated by the backend.
visibility (granite_assets.enums.AssetVisibility) – Visibility at the time of saving.

key: str

backend_ref: str

content_length: int | None

checksum: str | None

visibility: AssetVisibility

__init__(key: str, backend_ref: str, content_length: int | None = None, checksum: str | None = None, visibility: AssetVisibility = AssetVisibility.PRIVATE) → None

class granite_assets.models.AssetDescriptor(key: str, content_type: str | None = None, content_length: int | None = None, visibility: AssetVisibility = AssetVisibility.PRIVATE, last_modified: datetime | None = None, checksum: str | None = None, metadata: dict[str, str]=<factory>)[source]

Bases: object

Metadata about an asset that already exists in the repository.

Variables:

key (str) – Logical key.
content_type (str | None) – MIME type.
content_length (int | None) – Size in bytes.
visibility (granite_assets.enums.AssetVisibility) – Current visibility.
last_modified (datetime.datetime | None) – Last modification timestamp (backend-provided).
checksum (str | None) – Integrity hash when available.
metadata (dict[str, str]) – Backend-provided key/value metadata.

key: str

content_type: str | None

content_length: int | None

visibility: AssetVisibility

last_modified: datetime | None

checksum: str | None

metadata: dict[str, str]

__init__(key: str, content_type: str | None = None, content_length: int | None = None, visibility: AssetVisibility = AssetVisibility.PRIVATE, last_modified: datetime | None = None, checksum: str | None = None, metadata: dict[str, str]=<factory>) → None

class granite_assets.models.AssetAccessUrl(url: str, expires_at: datetime | None = None)[source]

Bases: object

A resolved URL for accessing an asset.

Variables:

url (str) – The full URL string.
expires_at (datetime.datetime | None) – Expiry timestamp for signed URLs; None for permanent URLs.

url: str

expires_at: datetime | None

property is_permanent: bool: True when the URL has no expiry (i.e. public permanent URL).

__init__(url: str, expires_at: datetime | None = None) → None

class granite_assets.models.UploadUrlResult(url: str, method: str, headers: dict[str, str], expires_at: datetime, key: str)[source]

Bases: object

Result of a pre-signed or tus upload URL request.

Backends may use different upload protocols:

S3 / S3-compatible — presigned PUT (single-part, up to 5 GB). The client sends the file body directly as a plain HTTP PUT to url.
tusd (tus protocol) — the client sends an HTTP POST to url to create the upload resource, then PATCH chunks until complete. The required tus headers (Tus-Resumable, Upload-Metadata) are included in headers. This supports arbitrarily large files and resumable uploads.

Variables:

url (str) – The URL the client should target for the first request.
method (str) – HTTP method for the first request ("PUT" for S3, "POST" for tus).
headers (dict[str, str]) – Headers that the client must include in the request.
expires_at (datetime.datetime) – When the upload authorisation token expires.
key (str) – The logical key that will be created after a successful upload.

url: str

method: str

headers: dict[str, str]

expires_at: datetime

key: str

__init__(url: str, method: str, headers: dict[str, str], expires_at: datetime, key: str) → None

class granite_assets.models.CfSignedCookies(policy: str, signature: str, key_pair_id: str, expires_at: datetime)[source]

Bases: object

CloudFront signed-cookie values to be set on the response.

All three cookies are required by CloudFront. Set them with HttpOnly; Secure; SameSite=None so the browser sends them on cross-origin CloudFront requests.

Variables:

policy (str) – Base64-encoded custom policy (CloudFront-Policy cookie).
signature (str) – RSA-SHA1 signature of the policy (CloudFront-Signature).
key_pair_id (str) – CloudFront key-pair ID (CloudFront-Key-Pair-Id).
expires_at (datetime.datetime) – When the authorisation expires.

policy: str

signature: str

key_pair_id: str

expires_at: datetime

as_cookie_header_values() → dict[str, str][source]: Return a mapping of cookie name → value ready for Set-Cookie.

__init__(policy: str, signature: str, key_pair_id: str, expires_at: datetime) → None

class granite_assets.models.LocalNginxAssetRepositoryConfig(storage_path: str, base_url: str, public_prefix: str = 'public', private_prefix: str = 'private', overwrite: bool = True, create_directories: bool = True, secure_link_secret: str | None = None, secure_link_ttl_seconds: int = 3600, tusd_url: str | None = None, upload_secret: str | None = None, upload_ttl_seconds: int = 3600)[source]

Bases: object

Configuration for LocalNginxAssetRepository.

Variables:

storage_path (str) – Absolute path on disk where assets are written.
base_url (str) –
Root URL at which Nginx (or any static server) serves the storage_path directory. Example:
```
"http://localhost:8080/assets"
```
public_prefix (str) – Sub-directory name (and URL path segment) used for publicly accessible assets. Default: "public". Files are placed at {storage_path}/{public_prefix}/{key} and served at {base_url}/{public_prefix}/{key}.
private_prefix (str) – Sub-directory name for private assets. Default: "private". Files are placed at {storage_path}/{private_prefix}/{key} and are only served when Nginx signed-URL validation passes (requires secure_link_secret to be set) or when the operator configures auth_request / internal directives on their own.
overwrite (bool) – Global overwrite default applied when AssetSaveRequest.overwrite is None. Defaults to True.
create_directories (bool) – Automatically create missing parent directories when saving assets. Disable if you want strict control over the directory layout. Default: True.
secure_link_secret (str | None) –
Shared secret used to generate and validate Nginx secure_link tokens for private assets. When set, build_download_url() and resolve_access() will return time-limited signed URLs instead of raising AssetAccessNotSupportedError.

The secret must match the value configured in the Nginx directive:
```
secure_link_md5 "$secure_link_expires$uri SECRET";
```
Keep this value secret. Anyone who knows it can forge download tokens. Use an environment variable or a secrets manager — never hardcode it.

When None (default), private asset access via URL is unsupported and the caller must proxy downloads through the application layer.
secure_link_ttl_seconds (int) – Default lifetime (in seconds) of signed URLs generated for private assets. Default: 3600 (1 hour). Individual calls to build_download_url() can override this value via the ttl_seconds parameter.
tusd_url (str | None) –
Base URL of the tusd server used to receive file uploads (e.g. "http://localhost:1080"). When set, build_upload_url() generates a tus creation request targeting {tusd_url}/files/.

tusd must be configured to call back to your application’s hook endpoint so it can:
1. Validate the upload-token in the upload metadata (pre-create hook).
2. Move the finished upload to the correct directory under storage_path (post-finish hook).
When None (default), build_upload_url() raises AssetAccessNotSupportedError.
upload_secret (str | None) –
HMAC-SHA256 secret used to sign upload tokens embedded in the tus Upload-Metadata header. The same secret must be available to the hook endpoint that validates incoming uploads.

Keep this value secret — it authorises writes to your storage directory.

When None and tusd_url is set, any upload is accepted without verification (only appropriate behind a trusted network).
upload_ttl_seconds (int) – Default lifetime (in seconds) of upload tokens. Default: 3600 (1 hour). Individual calls to build_upload_url() can override this value.

storage_path: str

base_url: str

public_prefix: str

private_prefix: str

overwrite: bool

create_directories: bool

secure_link_secret: str | None

secure_link_ttl_seconds: int

tusd_url: str | None

upload_secret: str | None

upload_ttl_seconds: int

__init__(storage_path: str, base_url: str, public_prefix: str = 'public', private_prefix: str = 'private', overwrite: bool = True, create_directories: bool = True, secure_link_secret: str | None = None, secure_link_ttl_seconds: int = 3600, tusd_url: str | None = None, upload_secret: str | None = None, upload_ttl_seconds: int = 3600) → None

class granite_assets.models.S3AssetRepositoryConfig(bucket: str, region: str, public_base_url: str | None = None, key_prefix: str = '', presign_ttl_seconds: int = 3600, endpoint_url: str | None = None, access_key_id: str | None = None, secret_access_key: str | None = None, session_token: str | None = None, cf_key_id: str | None = None, cf_private_key: str | None = None, cf_unsigned_urls: bool = False, cf_signing_method: CfSigningMethod = CfSigningMethod.URL, use_object_acl: bool = True)[source]

Bases: object

Configuration for S3AssetRepository.

Variables:

bucket (str) – S3 bucket name.
region (str) – AWS region (e.g. "eu-west-1").
public_base_url (str | None) – Optional CDN or custom-domain base URL for public assets. When set, build_public_url will use it instead of the native S3 endpoint.
key_prefix (str) – Optional prefix prepended to every logical key before writing to S3 (e.g. "uploads/").
presign_ttl_seconds (int) – Default TTL for presigned URLs.
endpoint_url (str | None) – Custom endpoint for S3-compatible stores (MinIO, etc.).
access_key_id (str | None) – Explicit AWS credentials (optional; falls back to the standard boto3 credential chain).
secret_access_key (str | None) – Explicit AWS credentials.
session_token (str | None) – STS session token when using temporary credentials.

bucket: str

region: str

public_base_url: str | None

key_prefix: str

presign_ttl_seconds: int

endpoint_url: str | None

access_key_id: str | None

secret_access_key: str | None

session_token: str | None

cf_key_id: str | None: CloudFront key-pair ID (KXXXXXXXXXXXXX). When set together with cf_private_key, build_download_url() generates a CloudFront signed URL instead of an S3 presigned URL.

cf_private_key: str | None: PEM-encoded RSA private key matching cf_key_id. May be the raw PEM string (-----BEGIN RSA PRIVATE KEY-----\n...) or a path to a file (not recommended; prefer injecting the value from a secret manager).

cf_unsigned_urls: bool

When True and public_base_url is set, build_download_url() returns a plain CloudFront URL (no signature, no expiry) instead of an S3 presigned URL. Use this when the CloudFront distribution has no Restrict Viewer Access policy and the bucket is only accessible via CloudFront OAC. The S3 bucket remains private; the URL is permanent (expires_at=None).

Security trade-off: the URL does not expire — anyone who obtains it can access the asset indefinitely. For time-limited access set cf_key_id and cf_private_key instead.

has_cf_signing() takes precedence: if both signing keys and this flag are set, CloudFront signed URLs are used.

cf_signing_method: CfSigningMethod

Controls whether access credentials are embedded in the URL (default) or issued as signed cookies. See CfSigningMethod.

URL – build_download_url() returns a CloudFront signed URL (query-param credentials). Works everywhere, no cookie handling needed.
COOKIE – build_download_url() returns a plain (unsigned) CloudFront URL. The caller must first obtain cookies via build_signed_cookies() and set them on the browser.

__init__(bucket: str, region: str, public_base_url: str | None = None, key_prefix: str = '', presign_ttl_seconds: int = 3600, endpoint_url: str | None = None, access_key_id: str | None = None, secret_access_key: str | None = None, session_token: str | None = None, cf_key_id: str | None = None, cf_private_key: str | None = None, cf_unsigned_urls: bool = False, cf_signing_method: CfSigningMethod = CfSigningMethod.URL, use_object_acl: bool = True) → None

use_object_acl: bool: When True (default), PutObject requests include an S3 object ACL (public-read for public assets, none for private). Set to False when the bucket has Object Ownership = BucketOwnerEnforced and ACLs are disabled — visibility is then expressed solely via key prefix and bucket policy / CloudFront OAC rather than per-object ACLs.

presign_ttl() → timedelta[source]: Convenience accessor returning the TTL as a timedelta.

has_cf_signing() → bool[source]: Return True when both CloudFront signing fields are present.

has_cf_unsigned_url() → bool[source]

Return True when plain (unsigned) CloudFront URL delivery is active.

Requires cf_unsigned_urls=True and public_base_url to be set. has_cf_signing() takes precedence over this flag.