66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
"""S3 (Garage) helpers shared by the web layer and Celery workers."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import boto3
|
|
from botocore.config import Config
|
|
|
|
from .config import settings
|
|
|
|
_EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)
|
|
|
|
|
|
def s3_client():
|
|
"""Garage S3 client (S3v4 signatures)."""
|
|
return boto3.client(
|
|
"s3",
|
|
endpoint_url=settings.S3_ENDPOINT_URL,
|
|
region_name=settings.S3_REGION or None,
|
|
aws_access_key_id=settings.S3_ACCESS_KEY or None,
|
|
aws_secret_access_key=settings.S3_SECRET_KEY or None,
|
|
config=Config(signature_version="s3v4"),
|
|
)
|
|
|
|
|
|
def s3_bucket_name(username: str) -> str:
|
|
"""Valid S3 bucket name for a user.
|
|
|
|
S3 names must be >= 3 chars, lowercase, only letters/numbers/dots/hyphens.
|
|
Short usernames (e.g. "jo") get a suffix so the bucket is legal.
|
|
"""
|
|
u = (username or "").strip().lower()
|
|
if not u:
|
|
return "shared-media"
|
|
return u if len(u) >= 3 else f"{u}-media"
|
|
|
|
|
|
def list_recent(bucket: str, max_keys: int = 20) -> list[dict]:
|
|
"""Return the newest objects in a bucket, newest first.
|
|
|
|
Returns an empty list on any error (unreachable/misconfigured/empty), so the
|
|
UI can degrade gracefully.
|
|
"""
|
|
try:
|
|
resp = s3_client().list_objects_v2(Bucket=bucket, MaxKeys=max_keys)
|
|
except Exception: # noqa: BLE001 - degrade to "no files"
|
|
return []
|
|
objects = [
|
|
{
|
|
"key": o.get("Key"),
|
|
"size": o.get("Size", 0),
|
|
"last_modified": o.get("LastModified"),
|
|
}
|
|
for o in resp.get("Contents", [])
|
|
]
|
|
objects.sort(key=lambda x: x["last_modified"] or _EPOCH, reverse=True)
|
|
return objects
|
|
|
|
|
|
def list_prefix(bucket: str, prefix: str, max_keys: int = 200) -> list[str]:
|
|
"""List object keys under a prefix (best-effort)."""
|
|
try:
|
|
resp = s3_client().list_objects_v2(Bucket=bucket, Prefix=prefix, MaxKeys=max_keys)
|
|
except Exception: # noqa: BLE001
|
|
return []
|
|
return [o.get("Key") for o in resp.get("Contents", [])] |