flip_api.utils.s3_client

Attributes

`_MULTIPART_OVERHEAD_BUFFER_BYTES`
`MAX_PUT_PRESIGNED_URL_TTL_SECONDS`

Classes

S3Client

S3 client wrapper for S3 operations.

Functions

`parse_s3_path`(→ tuple[str, str])	Parse an S3 path into bucket and key components.
`hash_s3_key`(→ str)	SHA-256 prefix of an S3 key, suitable for log correlation without leaking the key itself.

Module Contents

flip_api.utils.s3_client._MULTIPART_OVERHEAD_BUFFER_BYTES = 16384

flip_api.utils.s3_client.MAX_PUT_PRESIGNED_URL_TTL_SECONDS = 600

flip_api.utils.s3_client.parse_s3_path(s3_path: str) → tuple[str, str]

Parse an S3 path into bucket and key components.

Parameters:: s3_path – Full S3 path (e.g., s3://bucket-name/key)
Returns:: Tuple containing bucket name and key, e.g. (“bucket-name”, “key”)

flip_api.utils.s3_client.hash_s3_key(key: str) → str: SHA-256 prefix of an S3 key, suitable for log correlation without leaking the key itself.

class flip_api.utils.s3_client.S3Client

S3 client wrapper for S3 operations.

client

get_presigned_url(s3_path: str, expiration: int = 3600) → str

Generate a pre-signed URL for downloading a file from S3.

Parameters:

s3_path – Full S3 path (e.g., s3://bucket-name/key)
expiration – URL expiration time in seconds (default: 1 hour)

Returns:

Pre-signed URL string

Return type:

str

Raises:

Exception – If URL generation fails

get_put_presigned_post(s3_path: str, max_bytes: int, content_type: str | None = None, expiration: int = MAX_PUT_PRESIGNED_URL_TTL_SECONDS) → dict[str, Any]

Generate a pre-signed POST policy for uploading a file to S3 with explicit size and (optional) content-type constraints baked in.

S3 enforces the policy at the edge: multipart/form-data POSTs whose body exceeds max_bytes or whose Content-Type form field doesn’t match the policy are rejected before any bytes land in the bucket. The single-PUT URL produced by generate_presigned_url("put_object", ...) carries no such constraints, which is the whole point of using POST here instead.

Parameters:

s3_path (str) – Full S3 path (e.g., s3://bucket-name/key).
max_bytes (int) – Hard cap on the file size in bytes. The condition actually sent to S3 is max_bytes + _MULTIPART_OVERHEAD_BUFFER_BYTES because S3 measures the whole encoded request body, not just the file part — see the module-level comment for why.
content_type (str | None) – If provided, the policy locks Content-Type to this exact value. If None, any Content-Type is accepted but the size cap still applies.
expiration (int) – URL/policy expiration (seconds). Values above MAX_PUT_PRESIGNED_URL_TTL_SECONDS are silently clamped to the ceiling — a warning is logged so an over-limit caller leaves an audit trail. Silent clamping is deliberate: the ceiling is a hard security policy, never an error condition.

Returns:

{"url": ..., "fields": {...}} — pass through to the client as multipart/form-data POST.

Return type:

dict[str, Any]

Raises:

Exception – If policy generation fails.

delete_object(s3_path: str) → None

Delete an object from S3 bucket.

Parameters:: s3_path – Full S3 path (e.g., s3://bucket-name/key)
Raises:: Exception – If deletion fails

delete_objects(s3_paths: list[str]) → dict[str, Any]

Delete multiple objects from one or more S3 buckets in grouped batch requests.

Parameters:: s3_paths – List of full S3 paths (e.g., s3://bucket-name/key)
Returns:: Dictionary containing deletion results per bucket.
Return type:: dict[str, Any]
Raises:: Exception – If batch deletion fails for any bucket

get_object(s3_path: str) → dict[str, Any]

Get object from S3 bucket.

Parameters:: s3_path – Full S3 path (e.g., s3://bucket-name/key)
Returns:: Response containing object data.
Return type:: dict[str, Any]
Raises:: EndpointConnectionError – If connection to the S3 endpoint fails.

head_object(s3_path: str) → dict[str, Any]

Get object metadata from S3.

Parameters:: s3_path – Full S3 path (e.g., s3://bucket-name/key)
Returns:: Metadata of the object.
Return type:: dict[str, Any]
Raises:: Exception – If getting object metadata fails.

object_exists(s3_path: str) → bool

Check if an object exists in S3.

Parameters:: s3_path – Full S3 path (e.g., s3://bucket-name/key)
Returns:: True if the object exists, False otherwise.
Return type:: bool
Raises:: ClientError – If head_object fails with any error other than HTTP 404 (e.g., permissions issues).

copy_object(source_s3_path: str, dest_s3_path: str) → None

Copy object from source to destination bucket.

Parameters:

source_s3_path – Full S3 path of the source object (e.g., s3://source-bucket/key)
dest_s3_path – Full S3 path of the destination object (e.g., s3://dest-bucket/key)

Raises:

Exception – If copying the object fails.

list_objects(s3_path: str, delimiter: str = '') → list[str]

List object keys under a given S3 path (non-paginated) and return full S3 paths.

Parameters:

s3_path – Full S3 path (e.g., s3://bucket-name/prefix/)
delimiter – Character used to group keys (optional)

Returns:

//bucket/key)

Return type:

List of full S3 paths (e.g., s3

Raises:

HTTPException – If listing objects fails