pdf-quiz-generator/backend/app/services/storage_service.py

"""Where uploaded files live.

Two backends behind one interface:

  * ``local``  — the container volume, which is where every existing upload
                 already sits;
  * ``s3``     — MinIO (or any S3-compatible service).

Object storage is the right home for media: a container volume can only be
mounted by one host, has no presigned URLs, and no lifecycle rules. But the
existing 800-odd MB of PDFs live on the volume, so reads fall back to local when
an object is missing. That keeps the switch reversible and lets files migrate
gradually instead of in one risky pass.

A stored path is always the key, never a URL: URLs embed the backend, and a row
that hardcodes ``http://minio:9000/...`` breaks the moment the backend changes.
"""
import logging
import mimetypes
import os
from pathlib import Path

from app.config import settings

logger = logging.getLogger(__name__)

_client = None


def using_s3() -> bool:
    return settings.STORAGE_BACKEND == "s3"


def _s3():
    """A cached boto3 client pointed at MinIO."""
    global _client
    if _client is None:
        import boto3
        from botocore.config import Config

        _client = boto3.client(
            "s3",
            endpoint_url=settings.S3_ENDPOINT_URL,
            aws_access_key_id=settings.S3_ACCESS_KEY,
            aws_secret_access_key=settings.S3_SECRET_KEY,
            region_name=settings.S3_REGION,
            # MinIO needs path style; virtual-host style expects DNS per bucket.
            config=Config(signature_version="s3v4", s3={"addressing_style": "path"}),
        )
    return _client


def ensure_bucket() -> None:
    """Create the bucket if it is not there. Safe to call repeatedly."""
    if not using_s3():
        return
    client = _s3()
    try:
        client.head_bucket(Bucket=settings.S3_BUCKET)
    except Exception:
        try:
            client.create_bucket(Bucket=settings.S3_BUCKET)
            logger.info("Created bucket %s", settings.S3_BUCKET)
        except Exception:
            logger.warning("Could not create bucket %s", settings.S3_BUCKET, exc_info=True)


def local_path(key: str) -> Path:
    return Path(settings.UPLOAD_DIR) / key


def save(key: str, data: bytes, content_type: str | None = None) -> str:
    """Store bytes under `key` and return the key."""
    if using_s3():
        ensure_bucket()
        _s3().put_object(
            Bucket=settings.S3_BUCKET, Key=key, Body=data,
            ContentType=content_type or mimetypes.guess_type(key)[0] or "application/octet-stream",
        )
        return key
    target = local_path(key)
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_bytes(data)
    return key


def load(key: str) -> bytes | None:
    """Read an object, falling back to the local volume for files not yet migrated."""
    if using_s3():
        try:
            return _s3().get_object(Bucket=settings.S3_BUCKET, Key=key)["Body"].read()
        except Exception:
            logger.debug("Object %s not in S3; trying the local volume", key)
    target = local_path(key)
    return target.read_bytes() if target.is_file() else None


def exists(key: str) -> bool:
    if using_s3():
        try:
            _s3().head_object(Bucket=settings.S3_BUCKET, Key=key)
            return True
        except Exception:
            pass
    return local_path(key).is_file()


def delete(key: str) -> None:
    if using_s3():
        try:
            _s3().delete_object(Bucket=settings.S3_BUCKET, Key=key)
        except Exception:
            logger.warning("Could not delete %s from S3", key, exc_info=True)
    target = local_path(key)
    if target.is_file():
        try:
            target.unlink()
        except OSError:
            logger.warning("Could not delete %s from disk", key, exc_info=True)


def presigned_url(key: str, expires: int = 3600) -> str | None:
    """A time-limited direct URL, so large media need not stream through the API.

    None when the object is not in S3, in which case the caller serves it from
    the volume as before.
    """
    if not using_s3():
        return None
    try:
        _s3().head_object(Bucket=settings.S3_BUCKET, Key=key)
        return _s3().generate_presigned_url(
            "get_object",
            Params={"Bucket": settings.S3_BUCKET, "Key": key},
            ExpiresIn=expires,
        )
    except Exception:
        return None


def s3_object(key: str) -> bytes | None:
    """Read strictly from S3, with no fallback.

    `load` falls back to the volume, which is right for serving but useless for
    verifying a migration: it would compare a local file against itself.
    """
    if not using_s3():
        return None
    try:
        return _s3().get_object(Bucket=settings.S3_BUCKET, Key=key)["Body"].read()
    except Exception:
        return None


def size_of(key: str) -> int | None:
    if using_s3():
        try:
            return _s3().head_object(Bucket=settings.S3_BUCKET, Key=key)["ContentLength"]
        except Exception:
            pass
    target = local_path(key)
    return target.stat().st_size if target.is_file() else None