primus.storage.graphql_artifact_store module

Application-authorized artifact transfer over GraphQL-issued HTTPS tickets.

This module deliberately knows nothing about dashboard authentication or object storage credentials. Callers provide a dashboard-client-compatible executor; the application authorizes an artifact transfer and the store transfers the bytes only through the resulting HTTPS URL.

exception primus.storage.graphql_artifact_store.ArtifactAuthorizationError

Bases: GraphQLArtifactStoreError

The signed URL rejected the authorized transfer.

exception primus.storage.graphql_artifact_store.ArtifactIntegrityError

Bases: GraphQLArtifactStoreError

Bytes do not match the declared size or SHA-256 digest.

exception primus.storage.graphql_artifact_store.ArtifactTicketError

Bases: GraphQLArtifactStoreError

The application did not return a valid transfer ticket.

exception primus.storage.graphql_artifact_store.ArtifactTransferError

Bases: GraphQLArtifactStoreError

The HTTP transfer did not complete successfully.

class primus.storage.graphql_artifact_store.ArtifactTransferRequest(operation: str, resource_type: str, resource_id: str, artifact_type: str, filename: str, content_type: str, size_bytes: int, sha256: str)

Bases: object

One artifact request in the frozen GraphQL ticket contract.

__init__(operation: str, resource_type: str, resource_id: str, artifact_type: str, filename: str, content_type: str, size_bytes: int, sha256: str) → None

artifact_type: str

as_graphql_input() → dict[str, Any]

content_type: str

filename: str

operation: str

resource_id: str

resource_type: str

sha256: str

size_bytes: int

class primus.storage.graphql_artifact_store.ArtifactTransferTicket(object_key: str, method: str, url: str, required_headers: Mapping[str, str], expires_at: datetime)

Bases: object

A short-lived HTTPS transfer authorization returned by GraphQL.

__init__(object_key: str, method: str, url: str, required_headers: Mapping[str, str], expires_at: datetime) → None

expires_at: datetime

classmethod from_graphql(ticket: Mapping[str, Any]) → ArtifactTransferTicket

is_expired(now: datetime) → bool

method: str

object_key: str

required_headers: Mapping[str, str]

url: str

class primus.storage.graphql_artifact_store.ArtifactUpload(request: ArtifactTransferRequest, content: bytes, existing_metadata: Mapping[str, Any] | None = None)

Bases: object

A verified write request and its bytes for one batched authorization.

__init__(request: ArtifactTransferRequest, content: bytes, existing_metadata: Mapping[str, Any] | None = None) → None

content: bytes

existing_metadata: Mapping[str, Any] | None = None

request: ArtifactTransferRequest

class primus.storage.graphql_artifact_store.GraphQLArtifactStore(executor: GraphQLExecutor, *, http_session: HTTPSession | None = None, timeout_seconds: float = 30.0, ca_bundle: str | None = None)

Bases: object

Transfer artifacts through application-authorized GraphQL tickets only.

__init__(executor: GraphQLExecutor, *, http_session: HTTPSession | None = None, timeout_seconds: float = 30.0, ca_bundle: str | None = None) → None

static build_metadata(*, existing_metadata: Mapping[str, Any] | None, object_key: str, sha256: str, size_bytes: int, content_type: str) → dict[str, Any]: Add integrity metadata while retaining the compatible _s3_key field.

download_batch(requests_to_download: Sequence[ArtifactTransferRequest]) → list[bytes]: Authorize up to twenty reads with one ticket request and verify each.

download_bytes(request: ArtifactTransferRequest) → bytes: Download and verify bytes; no unchecked or alternate source is returned.

request_tickets(requests_to_authorize: Sequence[ArtifactTransferRequest]) → list[ArtifactTransferTicket]: Request one to twenty tickets through the frozen GraphQL operation.

upload_batch(uploads: Sequence[ArtifactUpload]) → list[dict[str, Any]]

Authorize up to twenty verified writes with one ticket request.

An explicit signed-URL expiry or transient HTTPS failure may refresh only its individual WRITE ticket once; authorization, integrity, and other failures are never retried.

upload_bytes(request: ArtifactTransferRequest, content: bytes, *, existing_metadata: Mapping[str, Any] | None = None) → dict[str, Any]: Upload verified bytes, then return metadata for a caller to persist.

exception primus.storage.graphql_artifact_store.GraphQLArtifactStoreError

Bases: RuntimeError

Base class for artifact-store failures that must not fall back.

class primus.storage.graphql_artifact_store.GraphQLExecutor(*args, **kwargs)

Bases: Protocol

The stable subset supplied by PrimusDashboardClient.

__init__(*args, **kwargs)

execute(query: str, variables: dict[str, Any] | None = None, **kwargs: Any) → dict[str, Any]

class primus.storage.graphql_artifact_store.HTTPSession(*args, **kwargs)

Bases: Protocol

__init__(*args, **kwargs)

request(method: str, url: str, *, headers: Mapping[str, str], data: bytes | None = None, timeout: float | None = None) → Any