Storage

@dataclass(frozen=True)
class FileInfo:
    """Metadata describing a stored file. This (as a dict, via
    [to_dict][starlette_admin.storage.FileInfo.to_dict]) is what file fields
    persist in the database column.

    Attributes:
        filename: Original (sanitised) filename, for display.
        content_type: MIME type reported at upload time.
        size: File size in bytes.
        storage: Name of the storage backend holding the file.
        key: Storage-relative path / object key.
        url: Public-facing URL. Refreshed on every render by
            `BaseStorage.url`; an empty string is stored for backends
            (like `LocalStorage`) whose URLs are request-dependent.
        uploaded_at: UTC timestamp of the upload.
        width: Image width in pixels (`ImageField` only).
        height: Image height in pixels (`ImageField` only).
        thumbnail: Thumbnail metadata (`ImageField` with `thumbnail_size` only),
            holding `{key, width, height, size}`. The thumbnail lives in the
            same storage as the parent file, so no separate storage name is
            recorded.
        extra: Free-form backend-specific metadata.
    """

    filename: str
    content_type: str
    size: int
    storage: str
    key: str
    url: str = ""
    uploaded_at: datetime | None = None
    width: int | None = None
    height: int | None = None
    thumbnail: dict | None = None
    extra: dict = dc_field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        """JSON-serializable dict form, suitable for a database JSON column."""
        data = asdict(self)
        if self.uploaded_at is not None:
            data["uploaded_at"] = self.uploaded_at.isoformat()
        return {k: v for k, v in data.items() if v is not None}

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "FileInfo":
        """Rebuild a `FileInfo` from its stored dict form."""
        kwargs = {k: v for k, v in data.items() if k in cls.__dataclass_fields__}
        uploaded_at = kwargs.get("uploaded_at")
        if isinstance(uploaded_at, str):
            kwargs["uploaded_at"] = datetime.fromisoformat(uploaded_at)
        return cls(**kwargs)

class BaseStorage(ABC):
    """Base class for file storage backends.

    Each instance registers itself under its `name` so that stored values
    (which carry only the storage *name* in their metadata) can be resolved
    back to the backend that holds them. See
    [get_storage][starlette_admin.storage.get_storage].
    """

    name: str = ""

    def __init__(self, name: str | None = None) -> None:
        if name is not None:
            self.name = name
        assert self.name, "storage must have a non-empty name"
        register_storage(self)

    @abstractmethod
    async def save(self, upload: UploadFile, dest: str) -> FileInfo:
        """Persist `upload` under the (sanitised, storage-relative) path
        `dest` and return the resulting [FileInfo][starlette_admin.storage.FileInfo].
        Implementations must never overwrite an existing file: `dest` is
        uniquified when taken.
        """
        raise NotImplementedError()

    @abstractmethod
    async def url(
        self, request: Request, key: str, *, signed: bool = False, expires: int = 3600
    ) -> str:
        """Return a public-facing URL for `key`. Called on every page render;
        signed URLs are never stored.
        """
        raise NotImplementedError()

    @abstractmethod
    async def delete(self, key: str) -> None:
        """Remove the file identified by `key`. Must not fail if the file is
        already gone.
        """
        raise NotImplementedError()

    async def read(self, key: str) -> bytes:
        """Return the full contents of the file identified by `key` as bytes.

        Used by the export system to embed files into ZIP archives. Backends
        whose files live on an external host (e.g. S3) must override this;
        the default raises `NotImplementedError` so callers can skip the file
        gracefully rather than crashing the export.

        Raises:
            NotImplementedError: for backends that have not implemented this.
            FileNotFoundError: if the file does not exist at `key`.
        """
        raise NotImplementedError(
            f"{type(self).__name__} does not implement read(); "
            "override it to support file embedding in ZIP exports."
        )

    async def serve(self, request: Request, key: str) -> Response:
        """Serve the file identified by `key` over HTTP. Backs the admin's
        ``/files/{storage}/{path}`` route; backends whose URLs point at an
        external host don't need to implement it.
        """
        raise NotImplementedError()

class LocalStorage(BaseStorage):
    """Filesystem storage rooted at `base_dir`.

    Files are served exclusively through the admin's ``/_files/{storage}/{path}``
    route; no external `StaticFiles` mount is needed or supported.

    Parameters:
        base_dir: Directory under which all files are stored (created if absent).
        name: Registry name for this instance. Override it when using several
            `LocalStorage` instances with different directories.
    """

    name = "local"

    def __init__(
        self,
        base_dir: str | Path,
        name: str | None = None,
    ) -> None:
        self.base_dir = Path(base_dir).resolve()
        self.base_dir.mkdir(parents=True, exist_ok=True)
        super().__init__(name)

    def _full_path(self, key: str) -> Path:
        """Resolve ``key`` inside ``base_dir``.

        Rejects keys that contain ``..`` path components (even when the final
        resolved path would still be inside ``base_dir``) and keys whose
        resolved path escapes ``base_dir``.
        """
        parts = Path(key).parts
        if ".." in parts:
            logger.warning("Rejected key %r: contains '..' path component", key)
            raise HTTPException(HTTP_404_NOT_FOUND)
        path = (self.base_dir / key).resolve()
        if not path.is_relative_to(self.base_dir):
            logger.warning("Rejected key %r: resolves outside %s", key, self.base_dir)
            raise HTTPException(HTTP_404_NOT_FOUND)
        return path

    async def save(self, upload: UploadFile, dest: str) -> FileInfo:
        path = self._full_path(dest)
        path.parent.mkdir(parents=True, exist_ok=True)
        while path.exists():
            path = path.with_name(f"{uuid.uuid4().hex[:8]}_{path.name}")
        size = 0
        with path.open("wb") as out:
            while chunk := await upload.read(_CHUNK_SIZE):
                size += len(chunk)
                out.write(chunk)
        logger.debug("Saved %r (%d bytes) to %s", dest, size, path)
        return FileInfo(
            filename=path.name,
            content_type=upload.content_type or "application/octet-stream",
            size=size,
            storage=self.name,
            key=path.relative_to(self.base_dir).as_posix(),
            uploaded_at=datetime.now(UTC),
        )

    async def url(
        self, request: Request, key: str, *, signed: bool = False, expires: int = 3600
    ) -> str:
        route_name = request.app.state.ROUTE_NAME
        return str(request.url_for(f"{route_name}:file", storage=self.name, path=key))

    async def delete(self, key: str) -> None:
        path = self._full_path(key)
        if path.is_file():
            path.unlink()
            logger.debug("Deleted %r from %s", key, self.base_dir)
        else:
            logger.debug("Delete no-op: %r not found in %s", key, self.base_dir)

    async def read(self, key: str) -> bytes:
        path = self._full_path(key)
        if not path.is_file():
            logger.warning("Read failed: %r not found in %s", key, self.base_dir)
            raise FileNotFoundError(f"No file at key {key!r} in {self.base_dir}")
        return path.read_bytes()

    async def serve(self, request: Request, key: str) -> Response:
        path = self._full_path(key)
        if not path.is_file():
            logger.warning("Serve failed: %r not found in %s", key, self.base_dir)
            raise HTTPException(HTTP_404_NOT_FOUND)
        return FileResponse(path)

class S3Storage(BaseStorage):
    """AWS S3 (and compatible) storage backend.

    Parameters:
        bucket: S3 bucket name.
        prefix: Key prefix for every stored object (acts like a folder).
        region: AWS region (used for URL construction and signing).
        access_key: AWS access key ID. Falls back to the standard boto3
            credential chain (env vars, shared credentials file, IAM role).
        secret_key: AWS secret access key. Falls back to the credential chain.
        public: When ``True`` (default), `url()` returns a plain
            ``https://{bucket}.s3.{region}.amazonaws.com/{key}`` URL.
            When ``False``, a pre-signed URL valid for `expires` seconds is
            generated on every render.
        expires: Lifetime of pre-signed URLs in seconds (default 3600 = 1 h).
            Only used when ``public=False`` or when ``url()`` is called with
            ``signed=True``.
        endpoint_url: Override the S3 endpoint URL for MinIO / R2 / Backblaze.
        name: Registry name for this instance. Override it when using several
            `S3Storage` instances.
    """

    name = "s3"

    def __init__(
        self,
        bucket: str,
        prefix: str = "uploads/",
        region: str = "us-east-1",
        access_key: str | None = None,
        secret_key: str | None = None,
        public: bool = True,
        expires: int = 3600,
        endpoint_url: str | None = None,
        name: str | None = None,
    ) -> None:
        try:
            import aiobotocore.session  # noqa: F401
        except ImportError as err:
            raise ImportError(
                "'aiobotocore' is required to use S3Storage. "
                "Install it with: pip install starlette-admin[s3]"
            ) from err
        self.bucket = bucket
        self.prefix = prefix.lstrip("/")
        self.region = region
        self.access_key = access_key
        self.secret_key = secret_key
        self.public = public
        self.expires = expires
        self.endpoint_url = endpoint_url
        super().__init__(name)

    def _client_kwargs(self) -> dict:

        kwargs: dict = {"region_name": self.region}
        if self.access_key and self.secret_key:
            kwargs["aws_access_key_id"] = self.access_key
            kwargs["aws_secret_access_key"] = self.secret_key
        if self.endpoint_url:
            kwargs["endpoint_url"] = self.endpoint_url
        return kwargs

    def _key(self, dest: str) -> str:
        return f"{self.prefix}{dest}" if self.prefix else dest

    def _public_url(self, key: str) -> str:
        if self.endpoint_url:
            base = self.endpoint_url.rstrip("/")
            return f"{base}/{self.bucket}/{quote(key, safe='/')}"
        return f"https://{self.bucket}.s3.{self.region}.amazonaws.com/{quote(key, safe='/')}"

    async def save(self, upload: UploadFile, dest: str) -> FileInfo:
        import aiobotocore.session

        session = aiobotocore.session.get_session()
        # Always prefix with a random token so each upload gets a unique key
        # without a head_object check. A check-then-write pair is inherently
        # racy: two concurrent uploads to the same dest can both find "not
        # found" and then overwrite each other on put_object.
        key = self._key(f"{uuid.uuid4().hex[:8]}_{dest}")
        content_type = upload.content_type or "application/octet-stream"
        async with session.create_client("s3", **self._client_kwargs()) as client:
            await client.put_object(
                Bucket=self.bucket,
                Key=key,
                Body=upload.file,
                ContentType=content_type,
                ContentLength=upload.size or 0,
            )
        logger.debug(
            "Saved %r (%d bytes) to bucket %r", key, upload.size or 0, self.bucket
        )
        url = self._public_url(key) if self.public else ""
        return FileInfo(
            filename=dest.rsplit("/", maxsplit=1)[-1],
            content_type=content_type,
            size=upload.size or 0,
            storage=self.name,
            key=key,
            url=url,
            uploaded_at=datetime.now(UTC),
        )

    async def url(
        self,
        request: Request,
        key: str,
        *,
        signed: bool = False,
        expires: int | None = None,
    ) -> str:
        if self.public and not signed:
            return self._public_url(key)
        import aiobotocore.session

        session = aiobotocore.session.get_session()
        async with session.create_client("s3", **self._client_kwargs()) as client:
            return await client.generate_presigned_url(
                "get_object",
                Params={"Bucket": self.bucket, "Key": key},
                ExpiresIn=expires if expires is not None else self.expires,
            )

    async def delete(self, key: str) -> None:
        import aiobotocore.session

        session = aiobotocore.session.get_session()
        async with session.create_client("s3", **self._client_kwargs()) as client:
            try:
                await client.delete_object(Bucket=self.bucket, Key=key)
            except Exception:  # pragma: no cover
                logger.warning(
                    "Failed to delete %r from bucket %r",
                    key,
                    self.bucket,
                    exc_info=True,
                )
            else:
                logger.debug("Deleted %r from bucket %r", key, self.bucket)

    async def read(self, key: str) -> bytes:
        """Return the object's bytes from S3.

        Raises:
            FileNotFoundError: if the object does not exist at ``key``.
        """
        import aiobotocore.session
        from botocore.exceptions import ClientError

        session = aiobotocore.session.get_session()
        async with session.create_client("s3", **self._client_kwargs()) as client:
            try:
                response = await client.get_object(Bucket=self.bucket, Key=key)
            except ClientError as exc:
                error_code = exc.response.get("Error", {}).get("Code")
                if error_code in ("NoSuchKey", "404"):
                    logger.warning(
                        "Read failed: %r not found in bucket %r", key, self.bucket
                    )
                    raise FileNotFoundError(
                        f"No file at key {key!r} in {self.name}"
                    ) from exc
                raise
            async with response["Body"] as stream:
                return await stream.read()

    async def serve(self, request: Request, key: str) -> Response:
        """Redirect to the bucket URL; S3 files are served directly from there."""
        from starlette.responses import RedirectResponse

        return RedirectResponse(await self.url(request, key, signed=not self.public))

class UnknownStorageError(StarletteAdminException):
    """Raised when a `FileInfo.storage` name doesn't match any registered storage."""