feat: implement s3 pipeline for media uploads

app/core/config.py

@@ -18,6 +18,8 @@ class Settings(BaseSettings):
     s3_port: int = Field(alias='S3_PORT')
     minio_root_user: str = Field(alias='MINIO_ROOT_USER')
     minio_root_password: str = Field(alias='MINIO_ROOT_PASSWORD')
+    minio_bucket_name: str = Field(alias='MINIO_BUCKET_NAME')
+    minio_url: str = Field(alias='MINIO_URL')
     pool_size: int = Field(alias='POOL_SIZE')
     max_overflow: int = Field(alias='MAX_OVERFLOW')
     jwt_algorithm: str = Field(default='HS256', alias='JWT_ALGORITHM')
@@ -28,6 +30,9 @@ class Settings(BaseSettings):
     rate_limit_ttl_seconds: int = Field(alias='RATE_LIMIT_TTL_SECONDS')
     idempotent_key_lifetime_sec: int = Field(alias='IDEMPOTENT_KEY_LIFETIME_SEC')
     reserve_timeout_minutes: int = Field(alias='RESERVE_TIMEOUT_MINUTES')
+    presigned_url_expire_seconds: int = Field(alias='PRESIGNED_URL_EXPIRE_SECONDS')
+    min_file_size_bytes: int = Field(alias='MIN_FILE_SIZE_BYTES')
+    max_file_size_bytes: int = Field(alias='MAX_FILE_SIZE_BYTES')
     secret_key: str = Field(alias='SECRET_KEY')
     debug_mode: bool = Field(default=False, alias='DEBUG_MODE')
 

app/core/s3.py

@@ -0,0 +1,44 @@
+import logging
+from collections.abc import AsyncGenerator
+from typing import Any
+
+import aioboto3  # type: ignore
+from botocore.exceptions import ClientError
+
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+session = aioboto3.Session()
+
+
+async def get_s3_client() -> AsyncGenerator[Any, None]:
+    async with session.client(
+        's3',
+        endpoint_url=settings.minio_url,
+        region_name='us-east-1',
+        aws_access_key_id=settings.minio_root_user,
+        aws_secret_access_key=settings.minio_root_password,
+        verify=False,
+    ) as client:
+        yield client
+
+
+async def init_s3_bucket() -> None:
+    async with session.client(
+        's3',
+        endpoint_url=settings.minio_url,
+        region_name='us-east-1',
+        aws_access_key_id=settings.minio_root_user,
+        aws_secret_access_key=settings.minio_root_password,
+        verify=False,
+    ) as client:
+        try:
+            await client.head_bucket(Bucket=settings.minio_bucket_name)
+            logger.info(f'Bucket {settings.minio_bucket_name} already exists')
+        except ClientError as e:
+            if e.response['Error']['Code'] == '404':
+                await client.make_bucket(Bucket=settings.minio_bucket_name)
+                logger.info(f'Bucket {settings.minio_bucket_name} created')
+            else:
+                logger.error(e)

app/main.py

@@ -8,8 +8,10 @@
 from app.core.config import settings
 from app.core.logging import setup_logging
 from app.core.lua_scripts import RATE_LIMIT_LUA_SCRIPT
+from app.core.s3 import init_s3_bucket
 from app.core.setup import setup_exception_handlers
 from app.services.inventory.routes import router_v1 as inventory_router_v1
+from app.services.media.routes import router_v1 as media_router_v1
 from app.services.orders.routes import router_v1 as order_router_v1
 from app.services.user.routes import router_v1 as user_router_v1
 
@@ -22,6 +24,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     client = Redis.from_url(settings.redis_url, decode_responses=True, encoding='utf-8')
     app.state.redis = client
     app.state.rate_limit_script = client.register_script(RATE_LIMIT_LUA_SCRIPT)
+    await init_s3_bucket()
     try:
         logger.info('redis connected')
         yield
@@ -42,6 +45,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 app.include_router(user_router_v1, prefix='/api/v1', tags=['Users'])
 app.include_router(order_router_v1, prefix='/api/v1', tags=['Orders'])
 app.include_router(inventory_router_v1, prefix='/api/v1', tags=['Inventory'])
+app.include_router(media_router_v1, prefix='/api/v1/media', tags=['Media'])
 
 
 @app.get('/health')

app/services/inventory/internal.py

@@ -66,3 +66,14 @@ async def cancel_reservation_by_order_and_return_stock(
     if product:
         product.qty_available += reservation.qty_reserved
     reservation.status = OrderStatus.CANCELLED
+
+
+async def ensure_product_exists(
+    session: AsyncSession,
+    product_id: UUID,
+) -> None:
+    prod_result = await session.execute(
+        select(Product.id).where(Product.id == product_id)
+    )
+    if prod_result.scalar_one_or_none() is None:
+        raise NotFoundError

app/services/media/models.py

@@ -0,0 +1,30 @@
+from datetime import datetime
+from enum import StrEnum
+from uuid import UUID, uuid4
+
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.sql import func
+from sqlalchemy.types import DateTime, String
+
+from app.core.database import Base
+
+
+class ImageStatus(StrEnum):
+    PENDING = 'pending'
+    ACTIVE = 'active'
+    INACTIVE = 'inactive'
+
+
+class ProductImage(Base):
+    __tablename__ = 'product_images'
+    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
+    product_id: Mapped[UUID] = mapped_column(ForeignKey('products.id'))
+    file_path: Mapped[str] = mapped_column(String(), nullable=False)
+    status: Mapped[str] = mapped_column(
+        String(), nullable=False, default=ImageStatus.PENDING
+    )
+    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, server_default=func.now(), onupdate=func.now()
+    )

app/services/media/routes.py

@@ -0,0 +1,38 @@
+from typing import Any
+from uuid import UUID
+
+from fastapi import APIRouter, Depends
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import get_session
+from app.core.s3 import get_s3_client
+from app.services.media.schemas import (
+    ImageUploadRequest,
+    ImageUploadResponse,
+    MinioWebhookEvent,
+)
+from app.services.media.service import generate_upload_url, handle_minio_webhook
+from app.services.user.models import User
+from app.shared.deps import get_current_user
+
+router_v1 = APIRouter(prefix='/api/v1/media', tags=['Media'])
+
+
+@router_v1.post('/products/{product_id}/upload_url', response_model=ImageUploadResponse)
+async def create_upload_url(
+    product_id: UUID,
+    req: ImageUploadRequest,
+    session: AsyncSession = Depends(get_session),
+    s3_client: Any = Depends(get_s3_client),
+    current_user: User = Depends(get_current_user),
+) -> ImageUploadResponse:
+    return await generate_upload_url(session, s3_client, product_id, req)
+
+
+@router_v1.post('/webhook/minio')
+async def minio_webhook(
+    event: MinioWebhookEvent,
+    session: AsyncSession = Depends(get_session),
+) -> dict[str, str]:
+    await handle_minio_webhook(session, event)
+    return {'status': 'ok'}

app/services/media/schemas.py

@@ -0,0 +1,31 @@
+from uuid import UUID
+
+from pydantic import BaseModel, Field
+
+
+class ImageUploadRequest(BaseModel):
+    filename: str
+    content_type: str
+
+
+class ImageUploadResponse(BaseModel):
+    image_id: UUID
+    url: str
+    fields: dict[str, str]
+
+
+class S3Object(BaseModel):
+    key: str
+
+
+class S3Entity(BaseModel):
+    object: S3Object
+
+
+class S3Record(BaseModel):
+    event_name: str = Field(alias='eventName')
+    s3: S3Entity
+
+
+class MinioWebhookEvent(BaseModel):
+    records: list[S3Record] = Field(alias='Records')

app/services/media/service.py

@@ -0,0 +1,78 @@
+from typing import Any
+from urllib.parse import unquote
+from uuid import UUID, uuid4
+
+import structlog
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.config import settings
+from app.services.inventory.internal import ensure_product_exists
+from app.services.media.models import ImageStatus, ProductImage
+from app.services.media.schemas import (
+    ImageUploadRequest,
+    ImageUploadResponse,
+    MinioWebhookEvent,
+)
+
+logger = structlog.get_logger(__name__)
+
+
+async def generate_upload_url(
+    session: AsyncSession,
+    s3_client: Any,
+    product_id: UUID,
+    req: ImageUploadRequest,
+) -> ImageUploadResponse:
+    await ensure_product_exists(session, product_id)
+    image_id = uuid4()
+    file_path = f'products/{product_id}/{image_id}-{req.filename}'
+    db_image = ProductImage(
+        id=image_id,
+        product_id=product_id,
+        file_path=file_path,
+    )
+    session.add(db_image)
+    await session.commit()
+    presigned_url = await s3_client.generate_presigned_post(
+        Bucket=settings.minio_bucket_name,
+        Key=file_path,
+        Fields={'Content-Type': req.content_type},
+        Conditions=[
+            {'Content-Type': req.content_type},
+            [
+                'content-length-range',
+                settings.min_file_size_bytes,
+                settings.max_file_size_bytes,
+            ],
+        ],
+        ExpiresIn=settings.presigned_url_expire_seconds,
+    )
+    return ImageUploadResponse(
+        image_id=image_id,
+        url=presigned_url['url'],
+        fields=presigned_url['fields'],
+    )
+
+
+async def handle_minio_webhook(
+    session: AsyncSession,
+    event: MinioWebhookEvent,
+) -> None:
+    if not event.records:
+        return
+    for record in event.records:
+        if not record.event_name.startswith('s3:ObjectCreated:'):
+            continue
+        object_key = unquote(record.s3.object.key)
+        result = await session.execute(
+            select(ProductImage)
+            .with_for_update()
+            .where(ProductImage.file_path == object_key)
+        )
+        image = result.scalar_one_or_none()
+        if image is not None and image.status == ImageStatus.PENDING:
+            image.status = ImageStatus.ACTIVE
+            await session.commit()
+        else:
+            logger.warning('image not found or not in pending status')

migrations/env.py

@@ -9,6 +9,7 @@
 from app.core.config import settings
 from app.core.database import Base
 from app.services.inventory.models import Product, Reservation
+from app.services.media.models import ProductImage
 from app.services.orders.models import Order, OrderItem
 from app.services.user.models import User
 

migrations/versions/9ae77f428203_add_productimage_model.py

@@ -0,0 +1,49 @@
+"""Add ProductImage model
+
+Revision ID: 9ae77f428203
+Revises: 8ca7fefa94b6
+Create Date: 2026-02-22 17:46:22.021381
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = '9ae77f428203'
+down_revision: str | Sequence[str] | None = '8ca7fefa94b6'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        'product_images',
+        sa.Column('id', sa.Uuid(), nullable=False),
+        sa.Column('product_id', sa.Uuid(), nullable=False),
+        sa.Column('file_path', sa.String(), nullable=False),
+        sa.Column('status', sa.String(), nullable=False),
+        sa.Column(
+            'created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False
+        ),
+        sa.Column(
+            'updated_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False
+        ),
+        sa.ForeignKeyConstraint(
+            ['product_id'],
+            ['products.id'],
+        ),
+        sa.PrimaryKeyConstraint('id'),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('product_images')
+    # ### end Alembic commands ###

pyproject.toml

@@ -5,13 +5,15 @@ description = 'Add your description here'
 readme = 'README.md'
 requires-python = '>=3.10'
 dependencies = [
+    'aioboto3>=15.5.0',
     'alembic>=1.18.4',
     'arq>=0.27.0',
     'asyncpg>=0.31.0',
-    "bcrypt==3.2.2",
+    'bcrypt==3.2.2',
+    'boto3-stubs[essential,s3]>=1.42.54',
     'email-validator>=2.3.0',
     'fastapi>=0.129.0',
-    "passlib[bcrypt]==1.7.4",
+    'passlib[bcrypt]==1.7.4',
     'prometheus-fastapi-instrumentator>=7.1.0',
     'pydantic-settings>=2.12.0',
     'python-jose[cryptography]>=3.5.0',