derp.storage – File Storage

Storage wrapper for S3-compatible object storage.

Example usage:

from storage import Storage

async with Storage(

bucket_name=”my-bucket”, endpoint_url=”https://s3.amazonaws.com”, aws_access_key_id=”key”, aws_secret_access_key=”secret”,

) as storage:

# Upload a file await storage.upload_file(“path/to/local/file.txt”, “remote/file.txt”)

# Fetch a file content = await storage.fetch_file(“remote/file.txt”)

exception derp.storage.StorageAccessDeniedError[source]

Bases: StorageError

Raised when S3 rejects a call with AccessDenied / 403.

__init__(message='Access denied by storage backend')[source]
Parameters:

message (str)

exception derp.storage.StorageBackendError[source]

Bases: StorageError

Wraps an unrecognised backend failure (network, 5xx, unknown 4xx).

Always chains the original exception via raise ... from exc so the backend-specific type and traceback are preserved for debugging. The code attribute carries the S3 error code where available.

__init__(message='Storage backend error', code=None)[source]
Parameters:

  • message (str)

  • code (str | None)

exception derp.storage.StorageBucketNotFoundError[source]

Bases: StorageError

Raised when an operation targets a missing bucket.

__init__(bucket, message=None)[source]
Parameters:

  • bucket (str)

  • message (str | None)

class derp.storage.StorageClient[source]

Bases: object

S3-compatible storage client for uploading and fetching files.

Example:

config = StorageConfig(
    endpoint_url="https://s3.amazonaws.com",
    access_key_id="key",
    secret_access_key="secret",
)
storage = StorageClient(config)
await storage.connect()
await storage.upload_file("local.txt", "remote.txt")
await storage.disconnect()
__init__(config)[source]

Initialize Storage client.

Parameters:

config (StorageConfig) – Storage configuration.

async connect()[source]

Establish connection to S3.

Return type:

None

async disconnect()[source]

Close connection to S3.

Return type:

None

get_url(*, bucket, key)[source]

Gets the URL for a file in S3 (public or private).

Parameters:

  • bucket (str) – Name of the S3 bucket.

  • key (str) – S3 object key (path in bucket).

Returns:

URL for the file.

Raises:

ValueError – If no public endpoint URL is configured for this bucket and endpoint_url is not configured.

Return type:

str

async upload_file(*, bucket, key, data, content_type=None, metadata=None, extra_args=None)[source]

Upload a file to S3.

Parameters:

  • bucket (str) – Name of the S3 bucket.

  • key (str) – S3 object key (path in bucket).

  • data (bytes) – Bytes to upload.

  • content_type (str | None) – MIME type of the file.

  • metadata (dict[str, str] | None) – Metadata to attach to the object.

  • extra_args (dict[str, Any] | None) – Additional arguments to pass to put_object.

Return type:

None

Example:

await storage.upload_file(
    bucket="my-bucket",
    key="remote/file.txt",
    data=b"Hello, World!",
    content_type="text/plain",
    metadata={"author": "user123"},
)
async fetch_file(*, bucket, key)[source]

Fetch a file from S3.

Parameters:

  • bucket (str) – Name of the S3 bucket.

  • key (str) – S3 object key (path in bucket).

Returns:

File content as bytes.

Raises:
Return type:

bytes

Example

content = await storage.fetch_file(

bucket=”my-bucket”, key=”remote/file.txt”

)

async delete_file(*, bucket, key)[source]

Delete a file from S3.

S3 deletes are idempotent: calling this for a missing key succeeds silently rather than raising.

Parameters:

  • bucket (str) – Name of the S3 bucket.

  • key (str) – S3 object key (path in bucket).

Return type:

None

async delete_files(*, bucket, keys)[source]

Delete multiple files from S3 in a single request.

Parameters:

  • bucket (str) – Name of the S3 bucket.

  • keys (list[str]) – List of S3 object keys to delete.

Returns:

List of keys that were successfully deleted.

Raises:

StoragePartialDeleteError – If S3 reports per-key failures. The exception carries both the failed entries and the keys that succeeded in the same batch.

Return type:

list[str]

async copy_file(*, src_bucket, src_key, dst_bucket=None, dst_key)[source]

Copy an object between keys or buckets (server-side).

Parameters:

  • src_bucket (str) – Source bucket name.

  • src_key (str) – Source object key.

  • dst_bucket (str | None) – Destination bucket name. Defaults to src_bucket.

  • dst_key (str) – Destination object key.

Raises:
Return type:

None

async file_exists(*, bucket, key)[source]

Check if a file exists in S3.

Returns False for a missing key; only true backend errors (denied, network, 5xx) raise.

Parameters:

  • bucket (str) – Name of the S3 bucket.

  • key (str) – S3 object key (path in bucket).

Returns:

True if file exists, False otherwise.

Return type:

bool

async head_object(*, bucket, key)[source]

Get object metadata without downloading the content.

Parameters:

  • bucket (str) – Name of the S3 bucket.

  • key (str) – S3 object key (path in bucket).

Returns:

Dict with ‘content_type’, ‘content_length’, ‘last_modified’, ‘etag’, and ‘metadata’ keys.

Raises:

StorageObjectNotFoundError – If key does not exist.

Return type:

dict[str, Any]

async list_files(*, bucket, prefix='', max_keys=None)[source]

List files in S3 bucket.

Parameters:

  • bucket (str) – Name of the S3 bucket.

  • prefix (str) – Prefix to filter files by.

  • max_keys (int | None) – Maximum number of keys to return.

Returns:

List of object keys.

Return type:

list[str]

Example

files = await storage.list_files(bucket=”my-bucket”, prefix=”folder/”)

async signed_download_url(*, bucket, key, expires_in=3600, response_content_disposition=None, response_content_type=None)[source]

Generate a presigned URL for downloading (GET) an object.

Parameters:

  • bucket (str) – Name of the S3 bucket.

  • key (str) – S3 object key (path in bucket).

  • expires_in (int) – URL expiry in seconds (default 3600).

  • response_content_disposition (str | None) – Value for the Content-Disposition response header the storage backend returns for this URL, e.g. 'attachment; filename="report.pdf"' to force a download. The value is signed into the URL, so it can’t be tampered with.

  • response_content_type (str | None) – Value for the Content-Type response header the storage backend returns for this URL.

Returns:

Presigned URL string.

Return type:

str

async signed_upload_url(*, bucket, key, expires_in=3600, content_type=None)[source]

Generate a presigned URL for uploading (PUT) an object.

Parameters:

  • bucket (str) – Name of the S3 bucket.

  • key (str) – S3 object key (path in bucket).

  • expires_in (int) – URL expiry in seconds (default 3600).

  • content_type (str | None) – MIME type the uploader must use.

Returns:

Presigned URL string.

Return type:

str

async list_buckets()[source]

List all S3 buckets.

Returns:

List of dicts with ‘name’ and ‘creation_date’ keys.

Return type:

list[dict[str, Any]]

async list_objects(*, bucket, prefix='', max_keys=1000)[source]

List objects and common prefixes in a bucket.

Uses Delimiter='/' for folder-like browsing.

Parameters:

  • bucket (str) – Name of the S3 bucket.

  • prefix (str) – Prefix to filter by (use trailing / for folders).

  • max_keys (int) – Maximum number of keys to return.

Returns:

Dict with ‘objects’ (list of object metadata dicts) and ‘prefixes’ (list of prefix strings representing folders).

Return type:

dict[str, Any]

class derp.storage.StorageConfig[source]

Bases: _StrictModel

Storage configuration.

endpoint_url: str | None
public_urls: dict[str, str]
service_name: str
access_key_id: str | None
secret_access_key: str | None
session_token: str | None
region: str
use_ssl: bool
verify: bool | str
model_config = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

exception derp.storage.StorageError[source]

Bases: Exception

Base exception for all storage errors.

__init__(message, code=None)[source]
Parameters:

  • message (str)

  • code (str | None)

exception derp.storage.StorageNotConnectedError[source]

Bases: StorageError

Raised when storage client is used before connect().

__init__(message='Storage not connected. Call connect() first.')[source]
Parameters:

message (str)

exception derp.storage.StorageObjectNotFoundError[source]

Bases: StorageError

Raised when an S3 object lookup targets a missing key.

The bucket and key are exposed as attributes for logging.

__init__(bucket, key, message=None)[source]
Parameters:

  • bucket (str)

  • key (str)

  • message (str | None)

exception derp.storage.StoragePartialDeleteError[source]

Bases: StorageError

Raised when delete_objects reports per-key failures.

The errors attribute is a list of {"key", "code", "message"} dicts mirroring the S3 response. The deleted attribute lists keys that succeeded in the same batch.

__init__(errors, deleted)[source]
Parameters: