mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
40 lines
1.5 KiB
Python
40 lines
1.5 KiB
Python
from typing import Any
|
|
|
|
|
|
class S3Client:
|
|
def __init__(self, boto_s3_client: Any, bucket: str) -> None:
|
|
self._client = boto_s3_client
|
|
self._bucket = bucket
|
|
|
|
@property
|
|
def bucket(self) -> str:
|
|
return self._bucket
|
|
|
|
def get_object(self, key: str) -> bytes:
|
|
response: dict[str, Any] = self._client.get_object(
|
|
Bucket=self._bucket, Key=key
|
|
)
|
|
body: bytes = response["Body"].read()
|
|
return body
|
|
|
|
def put_object(self, key: str, body: bytes) -> str:
|
|
self._client.put_object(Bucket=self._bucket, Key=key, Body=body)
|
|
return f"s3://{self._bucket}/{key}"
|
|
|
|
def upload_file(self, local_path: str, key: str) -> str:
|
|
"""Upload a file from local disk, using boto's managed transfer so a
|
|
large object (a multi-GB Download Package ZIP, ADR-0060) is streamed in
|
|
multipart from ``/tmp`` rather than held in memory."""
|
|
self._client.upload_file(Filename=local_path, Bucket=self._bucket, Key=key)
|
|
return f"s3://{self._bucket}/{key}"
|
|
|
|
def generate_presigned_url(self, key: str, expires_in: int) -> str:
|
|
"""A time-limited URL that lets the holder GET this object without AWS
|
|
credentials (ADR-0060 — how a Download Package link is delivered).
|
|
``expires_in`` is the validity window in seconds."""
|
|
url: str = self._client.generate_presigned_url(
|
|
"get_object",
|
|
Params={"Bucket": self._bucket, "Key": key},
|
|
ExpiresIn=expires_in,
|
|
)
|
|
return url
|