s3

migrations/models/3_20260106175850_move_creative_media_to_s3.py

@@ -0,0 +1,90 @@
+from tortoise import BaseDBAsyncClient
+
+RUN_IN_TRANSACTION = True
+
+
+async def upgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        ALTER TABLE "creative" ADD "media_s3_key" VARCHAR(512);
+        ALTER TABLE "creative" DROP COLUMN "media_data";"""
+
+
+async def downgrade(db: BaseDBAsyncClient) -> str:
+    return """
+        ALTER TABLE "creative" ADD "media_data" BYTEA;
+        ALTER TABLE "creative" DROP COLUMN "media_s3_key";"""
+
+
+MODELS_STATE = (
+    "eJztXW1z2roS/isMn3pmcjsJkCanc+fOAKEtpwQYXtLe03Y8jlGIb4zN8UvSTKf//Uq2hS"
+    "W/YRuDbdgvnVTSCvtZSdY+2l39qq+0BVKMt91HUVWRUn9f+1VXxRXCf/irzmp1cb32KkiB"
+    "Kd4rdluJaXRvmLoombj4QVQMhIsWyJB0eW3KmopLVUtRSKEm4YayuvSKLFX+x0KCqS2R+Y"
+    "h0XPHtBy6W1QX6iQz63/WT8CAjZcE9rLwgv22XC+br2i6bz/s3H+yW5OfuBUlTrJXqtV6/"
+    "mo+aumluWfLiLZEhdUukIl000YJ5DfKU7gvTIueJcYGpW2jzqAuvYIEeREshYNT//WCpEs"
+    "GgZv8S+af1n3oKeCRNJdDKqkmw+PXbeSvvne3SOvmp7qf25E3z3R/2W2qGudTtShuR+m9b"
+    "UDRFR9TG1QNS0hF5bUE0g4De4BpTXqFwUHlJH7gLV/Qt/SMLyLTAQ9kbYRRmCl82TOv4HR"
+    "YjVXl1NRiD8ax/25vO2rdj8iYrw/hHsSFqz3qkpmGXvvpK3zgq0fD8cObNppPal/7sU438"
+    "t/b3aNjzK27TbvZ3nTyTaJmaoGovgrhgBhstpcDglp5irfUio2J5SVBsoYp1H97TK16LUT"
+    "a98pI56NV92gOqtSJqpK8dO0FNrI6lLq6EsG9ZR172VTNcjz5BnyIxVPv5oO04IZfkV/71"
+    "Z6PRbF41zpvvri9bV1eX1+fXuK39SMGqqxh1d/of+8MZrz1S8JvHWDYVFEQX73H0CGypgA"
+    "9V/ColXehW4k9BQerSfMT/bVxexmB2157YOwXcyjfuh25Vw6njQbQMpNt/p8CRlckHyr0P"
+    "0P0DKUoSMgzhUTQe0814n2CmGX/4xbqYKb82jSC4kci6rauFaOOiddW6br5rbYDclMTh52"
+    "BFTKqHJ8YWIAX3ovT0IuoLgathQFVECa2QGoZtx5X98HmCFNF+pSCirnE5pv2UczH9TYcH"
+    "LQ3bgJF9wK4w4C6qjICu/Q9JO4Pg9FJlHCxdwmsyElw6ZFdA3O4YBqaqwCB9JRsG7lowJI"
+    "2sQTsB80XTn4w1XjnmeFcx3vQ9JV2Xcp2OBImstVpDi1p9g1WrxspfIqri0n5q8tvklyhx"
+    "R/gY+RnVw0g9WncWy+qxrYDWA1oP2B+g9UCxQOsBrZec1kvLlOTLkhwF4WSinyGzYIZLow"
+    "jRn2Fjv7QQxg333tcZN9IpUG9u21//4Eb7YDT8SJszwHYHo44PzxVayKIDS4qByUtlwrYA"
+    "hoQbnc1GgsHZbESOTVIVBuWDrKBQ2n4bmoxgJQG9vEiCKG4VCaldF4ap0RSe0Gt6SD05QJ"
+    "Qiem+ZJv7RIJh/TUfDcDAZER+OcxW/4LeFLJlnNUU2zB/7WlQZ0+rekhVTVo235Af3ZF0R"
+    "LOKXWv+q6tstkA78S61hiqYVAjwZxT3VWtng9/EjiqqEAkrwpA/3KatjI9+1+XmM6+3urH"
+    "/Xe19zGnxXCTq44AaX6NIjLnJGUcoBf51guF9HDvZr/1B3ucjQ1TiaOeCl8mQQgpDnfE61"
+    "C1/gwfZCKbWUwPnlTgS6wKFJCJJBGD9oOpKX6mf0Gpj2W4jOMsMXYDVxsS6+bNi8wBjB7+"
+    "nYqDbQ7Wm3fdOrh03iHBDM9VChIPz4xSkcPTi0y3ZUk9MRTZWRgLOZw5/NDLSlrM60J6TW"
+    "Q05nmNqzuPMZhbQTzE1DOKKBIxpg8uGIBhQLRzRwRJPC85p+QRN7BVMBcGV1IUQ/1zLWSI"
+    "a5wEtWc407pslgGdk+VQasZyVSoZ6Sx2NEgMIjYOTAPc3dbsoK2lbiiRkUaVmnfRrPHv8U"
+    "Yjtz5FS06bzmmoHhDIZzVfceR2NfgeF8pIoFw/lYN5ovokq0sfmYCgT5tDqN7KSa0/aY9C"
+    "u5cWm+jbCiiRGel1TAp7oHIlG12XgzmncGvdp40uv2p33XN2ijHruSFOEC2dkWT3rtgY8W"
+    "kbQV3WQmdV5lRCriwHZo91VZfcaIC4qsPqUh7XxiVXEOPoB7IDipbYM9Zyc1GliXkqTxiZ"
+    "0IUcM5K2w2CW6IbVo/vwj5U4QSV6dFzxPZAbDtH6hy4gWOpZlgozHxKXHjxU5wuPlyCWTF"
+    "L7d1rqI4gmMzODaDY3M58Uvi2By2bc4BPjYBRmXx89kDCYaff/ubB5J5ZqcpaiBGmAUJEG"
+    "UcwHed0cl9yYv6Fm8Hkt+2JccvxwGZPm1S+eFMPSzDidvUUCbLS1Ze/DyDlcNs2pvVhvPB"
+    "IFncjGHdb55yx+iIKdNVOelGiBkpa8yIPRnDPF7cSRrj7EJb5Orn8q3Or0orZBj4ycn/fo"
+    "APzLaxAz4wOc7Ho3GVAB+YI1Us+MAcq48E89kLKDIyzTQvlE/G/gPMxhzSTUNer/05RtCV"
+    "4UHXVhuTMfsCE9INrDZFe2RlOg06ycPumKMMYD85GjkBuVTW/CIlsv3ZafosoxdDeJQNU9"
+    "PlnZOM4LF9Rzr8ZPf3WuqP3eE5EQ6aCH7ED188VyKw6nvdB3PCOt8gU3p0tvNAmwBtAtY1"
+    "0CagWKBNjt2QcT6xkmaFRUdE8iY+qVMlTpg9Q8pZwEseZnU7nEVYuVlQmBN2qdSW3ZA/rJ"
+    "9DeU34KEeHojM1uE6KYVaZ578YY4wxjXK2wPy+mwwJAmYYmGGwWwczDBQLZtixb0CPOfS3"
+    "P6QlslrqcGDPa9oJUScdCvSNM6pmW5+HVNga7+OeRSVEZeN5Z9Dvvq+trXtFlrB6xuPJ6K"
+    "49wOphhVKq5+I8gX4uziMVRKp8Adtw7Jp8nwfBd3BiXfyJdeg4zAG/UwpdTHLqT4PQdjzk"
+    "zjcgr5ioCLhfBe5XgViZUviF0HETxjwyYyqGemRb7TM/7DeW5XQNqh9AOQLlCMwUUI6nq1"
+    "igHIFyrB7lWEp6EbINQso3YP1KC12cewuki4J0UcWliwLa1A9hfrQpBEv5zzZyogpzPd84"
+    "KjKMIhPDiTHgbafG2JxduQdI8YnEwDtv7/s3oMqOnFEBquxIFQtUGVBlpaHK8L4Wf6kXIV"
+    "zZeNAeDgk15jah3l42W2Z7e5GySe+vXndGynRErD1S1h8KuOHHSW86Jb58Am5LUDa+q93R"
+    "7XjQs5tL2mptD94sjNvFRRKPsYtoj7GLgE+f85JC6iuU/IJwlRJcpWS/JlyltPMXf/9XKT"
+    "HIOHhkXLvD+ikY5j056+Z92AGeuvUsRx1FX9NSJeDiGPtC8sGXlzBNnRAeHJ0hNVel2FUu"
+    "pXsItepP+R7Nqxr+lnlzqptbLpzh5P7ePdKBVwVeFeg34FVBscCrngKvWm7aI+e9JlwgvZ"
+    "NL53w4nXem3Um/Q1hmS93smrIRzUnQjwE/iD37RFm+OEFxWJ6Kzgrm26gnZmN8cidCx3Ar"
+    "EWfTpMAuIHgi4MVxWayxvyuZlW9Ma+H3RCals7xhlQOIJOSz0vgFJlmZktXZ6IYQOBT1aO"
+    "LGoi32GScKfAzwMWC2Ax9zuooFPuZYDR4Tq2Opi6vQPXtHXkbmA/cJ5pMPfPsHbccJ6WQD"
+    "/7PRaDavGufNd9eXraury+vzTVrwYFVcfvBO/yNJEc5pL5gznGxS7L8DAEcTXqxMRTyoeL"
+    "qlcXmZgG/BrSIJF7vOl31d1g1TSIslLwVoUjQVMQOYnNBpYxkw4JP4AyjaUsaIaU9o1xvN"
+    "B6SnGemonLufRBFocMN7ABGb5LDPO/JKWNW3e6swJkzwJwATDgzZMOSZ4KxioOyTnvICqE"
+    "M4Ki66Opqo4gK6ga0CtioILpAawFaBYoGtArYqcoKmNVZ3slOLmIIlMlQhuzRklw7BwskE"
+    "tSsSeebEKgYHyLLNeXSCRQ4WeQEWuTtc4uxyb0QlsM5dfil/Iz147SEZHhD9A/Y7mHlgv4"
+    "NiwX4/Bfu9gtEqa6QuiHoCCqyPe8Ob/vDj+5rb5Lva7nZ7YzslkihJaG06WZXuRp+dpErP"
+    "2lO20JWcE3U4e5yFcP+a0lM/IHginvp+Z56UsDEiJwgY5CTPKyoEkkKfZUoKHb725QBi5a"
+    "NCAgv6dgBpxMWpQ8cs6mUKpeEZozheZHtwDU9lAScCnAhwIqWyuYATORnFAicCnEhpOJHE"
+    "GTz6w7u+zYe4G83vamcw6tp0yL2iSRnpkLiIHEqHXEXSIVeBNB5g0YNFDxZ9Ocyq9BY9GK"
+    "Q7G6RhV6bv57L0cu4Sk3n8lOo2+TKhdDA+gxlH25gNfsgl5TiENS+2N7qD3UB4Pwm0B9Ae"
+    "YB0D7QGKBdrj2GkP/kubhfrgeyj6ipb2zW1/KHyYD8ilLIuVrAoPuN13dTwZkWu1psKk1y"
+    "Y3cLn+8wKZoUztl0l/1mOqX3TZRN/VLpYilAoV3wRluPJevduB18DtYTxod3u3vaH3BJto"
+    "BvoMXgv6FF4Tt5f2sD3476zfpZ3g7ZryasoS7cOrJyNnNJ8J3UG/+3nKtnyRsYYsU5AUWX"
+    "oiN4h9IreSDTav5t5FS9+M1tIXo9X2E2WhjRqtJNE0rehgmpafOQrdz6VnQ06PTUpCiQg5"
+    "GfZ78KEvnBupwuGz36ZNbrFtbOAMZptjm+dvvIFltodlACyzo9/Ag2V2pIoFywwsM7DMwD"
+    "Irp2VW8F2k2xeg0phiHGruEE0LGy92eriV4NaQqiLnrKppceOkTg814J2Adyon7xQyt3OA"
+    "Mnm6mKIm9Pb7a7glK8FlzEzapB3RS5GBqbTw+bYYCQZfQTcolRbB9BcoFXEfeGnh2+068H"
+    "2S7m2ky9JjPYRed2vO4oh00WtTGr488s6JpFdNuArcjSffkedzbppoXLSuWtfNd63NBROb"
+    "kjgvdnqHRDQ9/oz0aIYmHD1G5MTzIbKrHJkaKUB0m1cTwIvz8wQA4lbRl6CSOh9poKlm6H"
+    "f2r+loGEEYeCI+IOcqfsFvC1kyz2qKbJg/yglrDIrkrTmml4L35rb91Y9rdzDq+E0V0kEn"
+    "XdLO/D8vv/8PAyERnQ=="
+)

pyproject.toml

@@ -6,6 +6,7 @@ readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
     "aerich>=0.9.2",
+    "aioboto3>=13.3.0",
     "aiogram>=3.16.0",
     "aiolimiter>=1.2.1",
     "asyncpg>=0.30.0",
@@ -19,6 +20,7 @@ dependencies = [
     "python-multipart>=0.0.20",
     "tortoise-orm>=0.25.1",
     "tortoise-orm-stubs>=1.0.2",
+    "types-aiobotocore-s3>=2.15.2",
     "uvicorn>=0.38.0",
 ]
 

src/adapter/s3.py

@@ -0,0 +1,88 @@
+import logging
+import uuid
+
+import aioboto3
+from pydantic import BaseModel
+from types_aiobotocore_s3.client import S3Client
+
+from src.usecase import S3Storage
+
+log = logging.getLogger(__name__)
+
+
+class S3Config(BaseModel):
+    ENDPOINT_URL: str
+    ACCESS_KEY_ID: str
+    SECRET_ACCESS_KEY: str
+    BUCKET_NAME: str
+    REGION: str = 'us-east-1'
+
+
+class S3(S3Storage):
+    def __init__(self, config: S3Config) -> None:
+        self.config = config
+        self.session = aioboto3.Session()
+        self._client: S3Client | None = None
+
+    async def connect(self) -> None:
+        # Создаем клиента для проверки/создания bucket
+        async with self.session.client(
+            's3',
+            endpoint_url=self.config.ENDPOINT_URL,
+            aws_access_key_id=self.config.ACCESS_KEY_ID,
+            aws_secret_access_key=self.config.SECRET_ACCESS_KEY,
+            region_name=self.config.REGION,
+        ) as client:
+            # Проверяем существование bucket
+            try:
+                await client.head_bucket(Bucket=self.config.BUCKET_NAME)
+                log.info(f'S3 bucket {self.config.BUCKET_NAME} exists')
+            except Exception:
+                # Bucket не существует, создаем
+                log.info(f'Creating S3 bucket {self.config.BUCKET_NAME}')
+                await client.create_bucket(Bucket=self.config.BUCKET_NAME)
+                log.info(f'S3 bucket {self.config.BUCKET_NAME} created')
+
+    async def close(self) -> None:
+        # aioboto3 использует context manager, не нужно закрывать
+        pass
+
+    async def upload(self, key: str, data: bytes, content_type: str) -> None:
+        async with self.session.client(
+            's3',
+            endpoint_url=self.config.ENDPOINT_URL,
+            aws_access_key_id=self.config.ACCESS_KEY_ID,
+            aws_secret_access_key=self.config.SECRET_ACCESS_KEY,
+            region_name=self.config.REGION,
+        ) as client:
+            await client.put_object(
+                Bucket=self.config.BUCKET_NAME,
+                Key=key,
+                Body=data,
+                ContentType=content_type,
+            )
+            log.info(f'Uploaded file to S3: {key}')
+
+    async def get(self, key: str) -> bytes:
+        async with self.session.client(
+            's3',
+            endpoint_url=self.config.ENDPOINT_URL,
+            aws_access_key_id=self.config.ACCESS_KEY_ID,
+            aws_secret_access_key=self.config.SECRET_ACCESS_KEY,
+            region_name=self.config.REGION,
+        ) as client:
+            response = await client.get_object(Bucket=self.config.BUCKET_NAME, Key=key)
+            data = await response['Body'].read()
+            log.info(f'Downloaded file from S3: {key}')
+            return data
+
+    async def delete(self, key: str) -> None:
+        async with self.session.client(
+            's3',
+            endpoint_url=self.config.ENDPOINT_URL,
+            aws_access_key_id=self.config.ACCESS_KEY_ID,
+            aws_secret_access_key=self.config.SECRET_ACCESS_KEY,
+            region_name=self.config.REGION,
+        ) as client:
+            await client.delete_object(Bucket=self.config.BUCKET_NAME, Key=key)
+            log.info(f'Deleted file from S3: {key}')

src/config.py

@@ -6,6 +6,7 @@ from shared.datebase_base import DatabaseConfig
 from shared.jwt_base import JWTConfig
 from shared.logger import LoggerConfig
 from shared.telegram_base import TelegramConfig
+from src.adapter.s3 import S3Config
 
 
 class AppConfig(BaseModel):
@@ -25,6 +26,7 @@ class Settings(BaseSettings):
     telegram: TelegramConfig
     jwt: JWTConfig
     parser: ParserConfig
+    s3: S3Config
 
 
 settings: Settings = load_settings(Settings)

src/domain/creative.py

@@ -23,7 +23,7 @@ class Creative(TimestampedModel):
     text = fields.TextField()
     media_type = fields.CharField(max_length=32, null=True)
     media_file_id = fields.CharField(max_length=512, null=True)
-    media_data = fields.BinaryField(null=True)
+    media_s3_key = fields.CharField(max_length=512, null=True)
     buttons = fields.JSONField(default=list)
     status = fields.CharEnumField(CreativeStatus, default=CreativeStatus.ACTIVE)
 

src/main.py

@@ -12,6 +12,7 @@ from src import deps
 from src.adapter.jwt import JWT
 from src.adapter.parser import ParserClient
 from src.adapter.postgres import Postgres
+from src.adapter.s3 import S3
 from src.adapter.telegram_bot import TelegramBot
 from src.config import settings
 from src.controller.http_v1 import api_router
@@ -22,11 +23,13 @@ from src.usecase import Usecase
 @asynccontextmanager
 async def lifespan(_: FastAPI) -> AsyncGenerator[None]:
     await postgres.connect()
+    await s3.connect()
     await fetch_placement_post_worker.start()
 
     yield
 
     await fetch_placement_post_worker.stop()
+    await s3.close()
     await postgres.close()
 
 
@@ -37,12 +40,14 @@ postgres = Postgres(settings.db)
 telegram = TelegramBot(settings.telegram)
 jwt_encoder = JWT(settings.jwt)
 parser_client = ParserClient(base_url=settings.parser.URL)
+s3 = S3(settings.s3)
 
 usecase = Usecase(
     database=postgres,
     telegram_bot=telegram,
     jwt_encoder=jwt_encoder,
     parser=parser_client,
+    s3=s3,
 )
 deps.set_usecase(usecase)
 

src/usecase/__init__.py

@@ -290,12 +290,25 @@ class Parser(typing.Protocol):
     async def fetch_telegram_channel(self, username: str) -> FetchChannelResponse | None: ...
 
 
+class S3Storage(typing.Protocol):
+    async def connect(self) -> None: ...
+
+    async def close(self) -> None: ...
+
+    async def upload(self, key: str, data: bytes, content_type: str) -> None: ...
+
+    async def get(self, key: str) -> bytes: ...
+
+    async def delete(self, key: str) -> None: ...
+
+
 @dataclass
 class Usecase:
     database: Database
     telegram_bot: TelegramBotWriter
     jwt_encoder: JWTEncoder
     parser: Parser
+    s3: S3Storage
 
     async def ensure_workspace_access(self, workspace_id: UUID, user_id: UUID) -> domain.Workspace:
         workspace = await self.database.get_workspace_for_user(workspace_id, user_id)

src/usecase/creative/create_creative.py

@@ -26,12 +26,20 @@ async def create_creative(
 
     creative_text = domain.replace_invite_link_with_tag(input.text)
 
+    media_s3_key: str | None = None
+    if input.media_data:
+        file_id = uuid.uuid4()
+        media_s3_key = f'creatives/{workspace_id}/{file_id}'
+        content_type = _get_content_type(input.media_type)
+        await self.s3.upload(media_s3_key, input.media_data, content_type)
+        log.info('Uploaded creative media to S3: %s', media_s3_key)
+
     creative = domain.Creative(
         name=input.name,
         text=creative_text,
         media_type=input.media_type,
         media_file_id=input.media_file_id,
-        media_data=input.media_data,
+        media_s3_key=media_s3_key,
         buttons=[button.model_dump() for button in input.buttons],
         status=domain.CreativeStatus.ACTIVE,
         project_id=project.id,
@@ -52,3 +60,16 @@ async def create_creative(
         status=creative.status,
         placements_count=0,
     )
+
+
+def _get_content_type(media_type: str | None) -> str:
+    """Map Telegram media type to MIME content type."""
+    if not media_type:
+        return 'application/octet-stream'
+
+    mapping = {
+        'photo': 'image/jpeg',
+        'video': 'video/mp4',
+        'animation': 'image/gif',
+    }
+    return mapping.get(media_type, 'application/octet-stream')

src/usecase/creative/delete_creative.py

@@ -26,4 +26,12 @@ async def delete_creative(self: 'Usecase', input: dto.DeleteCreativeInput) -> No
         log.warning('Creative %s is used in placements and cannot be deleted', input.creative_id)
         raise domain.CreativeInUse(input.creative_id)
 
+    media_key = creative.media_s3_key
+    if media_key:
+        try:
+            await self.s3.delete(media_key)
+            log.info('Deleted creative media from S3: %s', media_key)
+        except Exception as e:
+            log.warning('Failed to delete creative media from S3: %s', e)
+
     await self.database.delete_creative(input.creative_id)

src/usecase/creative/update_creative.py

@@ -32,7 +32,27 @@ async def update_creative(
         creative.text = domain.replace_invite_link_with_tag(input.text)
     if input.media_data is not None:
         domain.validate_media_size(input.media_data)
-        creative.media_data = input.media_data
+
+        # Удаляем старое медиа из S3
+        old_media_key = creative.media_s3_key
+        if old_media_key:
+            try:
+                await self.s3.delete(old_media_key)
+                log.info('Deleted old creative media from S3: %s', old_media_key)
+            except Exception as e:
+                log.warning('Failed to delete old creative media from S3: %s', e)
+
+        # Загружаем новое медиа в S3
+        new_media_data = input.media_data
+        if new_media_data:
+            file_id = uuid.uuid4()
+            new_media_key = f'creatives/{workspace_id}/{file_id}'
+            content_type = _get_content_type(input.media_type)
+            await self.s3.upload(new_media_key, new_media_data, content_type)
+            log.info('Uploaded new creative media to S3: %s', new_media_key)
+            creative.media_s3_key = new_media_key
+        else:
+            creative.media_s3_key = None
     if input.media_type is not None:
         creative.media_type = input.media_type
     if input.media_file_id is not None:
@@ -59,3 +79,16 @@ async def update_creative(
         status=creative.status,
         placements_count=placements_count,
     )
+
+
+def _get_content_type(media_type: str | None) -> str:
+    """Map Telegram media type to MIME content type."""
+    if not media_type:
+        return 'application/octet-stream'
+
+    mapping = {
+        'photo': 'image/jpeg',
+        'video': 'video/mp4',
+        'animation': 'image/gif',
+    }
+    return mapping.get(media_type, 'application/octet-stream')