ros_control/app/s3.py

"""Yandex Object Storage (S3-совместимый): ListObjectsV2, GetObject, PutObject, DeleteObject."""
import asyncio
from functools import lru_cache

import boto3
from botocore.config import Config

from app.config import get_settings


@lru_cache
def _client():
    s = get_settings()
    return boto3.client(
        "s3",
        endpoint_url=s.s3_endpoint,
        region_name=s.s3_region,
        aws_access_key_id=s.s3_access_key,
        aws_secret_access_key=s.s3_secret_key,
        config=Config(signature_version="s3v4"),
    )


def device_prefix(device_name: str) -> str:
    return f"{get_settings().s3_prefix}/{device_name}/"


def key_allowed(key: str) -> bool:
    """Ключ должен лежать под префиксом бэкапов (защита от произвольных ключей бакета)."""
    return key.startswith(get_settings().s3_prefix + "/") and ".." not in key


def presign_get(key: str, expires: int = 300) -> str:
    return _client().generate_presigned_url(
        "get_object", Params={"Bucket": get_settings().s3_bucket, "Key": key}, ExpiresIn=expires
    )


def _list(prefix: str) -> list[dict]:
    paginator = _client().get_paginator("list_objects_v2")
    items = []
    for page in paginator.paginate(Bucket=get_settings().s3_bucket, Prefix=prefix):
        for o in page.get("Contents", []):
            items.append({"key": o["Key"], "size": o["Size"], "last_modified": o["LastModified"]})
    return sorted(items, key=lambda i: i["last_modified"], reverse=True)


async def list_backups(device_name: str | None = None) -> list[dict]:
    prefix = device_prefix(device_name) if device_name else get_settings().s3_prefix + "/"
    return await asyncio.to_thread(_list, prefix)


async def upload_file(path, key: str, metadata: dict[str, str] | None = None) -> None:
    """PutObject: загрузка файла в бакет от имени Control Server. metadata — пользовательские метаданные объекта
    (x-amz-meta-*): сюда кладутся ID резервной копии и устройства."""
    extra = {"ExtraArgs": {"Metadata": metadata}} if metadata else {}
    await asyncio.to_thread(_client().upload_file, str(path), get_settings().s3_bucket, key, **extra)


async def delete_object(key: str) -> None:
    await asyncio.to_thread(
        _client().delete_object, Bucket=get_settings().s3_bucket, Key=key
    )