app/services/s3.py - master - tsps/core

import boto3
from botocore.config import Config
from fastapi import UploadFile
from app.core.config import config


class S3Service:
    def __init__(self):
        self.client = boto3.client(
            "s3",
            endpoint_url=config.S3_ENDPOINT,
            aws_access_key_id=config.S3_ACCESS_KEY,
            aws_secret_access_key=config.S3_SECRET_KEY,
            region_name=config.S3_REGION,
            config=Config(signature_version="s3v4"),
        )
        self.bucket = config.S3_BUCKET

    def upload_file(self, file: UploadFile, key: str) -> str:
        """Загружает файл в S3, возвращает ключ объекта"""
        content = file.file.read()
        self.client.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=content,
            ContentType=file.content_type,
        )
        return key

    def get_presigned_url(self, key: str, expires_in: int = 3600) -> str:
        """Генерирует временную ссылку на скачивание"""
        return self.client.generate_presigned_url(
            "get_object",
            Params={"Bucket": self.bucket, "Key": key},
            ExpiresIn=expires_in,
        )

    def delete_file(self, key: str) -> None:
        """Удаляет файл из S3"""
        self.client.delete_object(Bucket=self.bucket, Key=key)


s3_service = S3Service()