Add Cloudinary storage provider

backend/README.md

@@ -142,10 +142,19 @@ REDIS_URL=redis://localhost:6379/0
 LLM_MODEL_NAME=groq/openai/gpt-oss-120b
 GROQ_API_KEY=your-groq-api-key
 
+# Storage
+STORAGE_PROVIDER=supabase
+
 # Supabase Storage
 SUPABASE_URL=https://your-project.supabase.co
 SUPABASE_KEY=your-service-role-key
 SUPABASE_BUCKET_NAME=resumes
+
+# Cloudinary Storage (set STORAGE_PROVIDER=cloudinary)
+CLOUDINARY_CLOUD_NAME=your-cloud-name
+CLOUDINARY_API_KEY=your-api-key
+CLOUDINARY_API_SECRET=your-api-secret
+CLOUDINARY_FOLDER=resumes
 ```
 
 All variables are defined and validated in `app/config.py`. Access them anywhere via the `settings` singleton:
@@ -253,13 +262,13 @@ All major integrations follow the same pattern — abstractions in `app/interfac
 
 ```
 app/interfaces/llm_provider.py      →  app/ai/lite_llm.py
-app/interfaces/storage_provider.py  →  app/utils/supabase.py
+app/interfaces/storage_provider.py  →  app/utils/supabase.py, app/utils/cloudinary_provider.py
 app/interfaces/hasher.py            →  app/utils/ (BcryptHasher)
 app/interfaces/encrypter.py         →  app/utils/ (JwtEncrypter)
 app/interfaces/base_agent.py        →  app/ai/resume_evaluator.py
 ```
 
-This makes it straightforward to swap providers (e.g., Groq → OpenAI, Supabase → S3) without touching business logic.
+This makes it straightforward to swap providers (e.g., Groq → OpenAI, Supabase → Cloudinary) without touching business logic.
 
 ### Model Structure
 

backend/app/config.py

@@ -30,6 +30,12 @@ class Settings(BaseSettings):
     SUPABASE_KEY: str = ""
     SUPABASE_BUCKET_NAME: str = "resumes"
 
+    # Cloudinary
+    CLOUDINARY_CLOUD_NAME: str = ""
+    CLOUDINARY_API_KEY: str = ""
+    CLOUDINARY_API_SECRET: str = ""
+    CLOUDINARY_FOLDER: str = "resumes"
+
     # Piston (code execution)
     PISTON_URL: str = "http://localhost:2000"
 

backend/app/models/dsa_question.py

@@ -1,11 +1,13 @@
 from typing import Any
 
-from sqlalchemy import Integer, String, Text
+from sqlalchemy import JSON, Integer, String, Text
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.models.base import BaseTable
 
+JSON_TYPE = JSON().with_variant(JSONB, "postgresql")
+
 
 class DsaQuestion(BaseTable):
     __tablename__ = "dsa_questions"
@@ -16,8 +18,8 @@ class DsaQuestion(BaseTable):
     description: Mapped[str] = mapped_column(Text, nullable=False)
 
     # Each entry: {"stdin": str, "expected_stdout": str}. Multi-line values use "\n".
-    test_cases: Mapped[list[dict[str, Any]]] = mapped_column(JSONB, nullable=False)
-    sample_test_cases: Mapped[list[dict[str, Any]] | None] = mapped_column(JSONB, nullable=True)
+    test_cases: Mapped[list[dict[str, Any]]] = mapped_column(JSON_TYPE, nullable=False)
+    sample_test_cases: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON_TYPE, nullable=True)
 
     sample_solution: Mapped[str | None] = mapped_column(Text, nullable=True)
     time_limit_ms: Mapped[int] = mapped_column(Integer, nullable=False, default=5000)

backend/app/routers/application.py

@@ -130,9 +130,7 @@ async def shortlist_application(
         org.id,
     )
 
-    app_result = await db.execute(
-        select(Application).where(Application.id == application_id)
-    )
+    app_result = await db.execute(select(Application).where(Application.id == application_id))
     application = app_result.scalar_one_or_none()
 
     if not application:

backend/app/routers/interview.py

@@ -135,7 +135,6 @@ async def get_applied_interviews(
     return applied_interviews
 
 
-
 @router.post(
     "/seed-test",
     response_model=CustomInterviewResponse,
@@ -202,7 +201,6 @@ async def seed_test_interview(
 
 
 @router.get("/{interview_id}", response_model=CustomInterviewResponse)
-
 async def get_interview(
     interview_id: int,
     db: AsyncSession = Depends(get_db),

backend/app/utils/cloudinary_provider.py

@@ -0,0 +1,92 @@
+from asyncio import to_thread
+from io import BytesIO
+
+import cloudinary
+import cloudinary.uploader
+import cloudinary.utils
+import httpx
+
+from app.config import settings
+from app.exceptions.storage import (
+    StorageDeleteError,
+    StorageDownloadError,
+    StorageUploadError,
+)
+from app.interfaces.storage_proivder import StorageProviderInterface
+from app.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+class CloudinaryStorageProvider(StorageProviderInterface):
+    def __init__(
+        self,
+        cloud_name: str = settings.CLOUDINARY_CLOUD_NAME,
+        api_key: str = settings.CLOUDINARY_API_KEY,
+        api_secret: str = settings.CLOUDINARY_API_SECRET,
+        folder: str = settings.CLOUDINARY_FOLDER,
+    ):
+        if not cloud_name or not api_key or not api_secret:
+            logger.warning(
+                "Cloudinary credentials are not configured. Storage operations may fail."
+            )
+
+        self.folder = folder.strip("/")
+        cloudinary.config(
+            cloud_name=cloud_name,
+            api_key=api_key,
+            api_secret=api_secret,
+            secure=True,
+        )
+
+    def _public_id(self, file_name: str) -> str:
+        return f"{self.folder}/{file_name}" if self.folder else file_name
+
+    async def upload(self, file: bytes, file_name: str) -> str:
+        try:
+            result = await to_thread(
+                cloudinary.uploader.upload,
+                BytesIO(file),
+                public_id=self._public_id(file_name),
+                resource_type="raw",
+                overwrite=True,
+            )
+            url = result.get("secure_url") or result.get("url")
+            if not isinstance(url, str) or not url:
+                raise StorageUploadError("Cloudinary upload response did not include a URL")
+            return url
+        except StorageUploadError:
+            raise
+        except Exception as e:
+            logger.error("Cloudinary upload failed: %s", str(e), exc_info=True)
+            raise StorageUploadError(f"Failed to upload file to storage: {str(e)}") from e
+
+    async def delete(self, file_name: str) -> None:
+        try:
+            result = await to_thread(
+                cloudinary.uploader.destroy,
+                self._public_id(file_name),
+                resource_type="raw",
+            )
+            if result.get("result") not in {"ok", "not found"}:
+                raise StorageDeleteError(f"Cloudinary delete failed: {result}")
+        except StorageDeleteError:
+            raise
+        except Exception as e:
+            logger.error("Cloudinary delete failed: %s", str(e), exc_info=True)
+            raise StorageDeleteError(f"Failed to delete file from storage: {str(e)}") from e
+
+    async def download(self, file_name: str) -> bytes:
+        try:
+            url, _ = cloudinary.utils.cloudinary_url(
+                self._public_id(file_name),
+                resource_type="raw",
+                secure=True,
+            )
+            async with httpx.AsyncClient() as client:
+                response = await client.get(url)
+                response.raise_for_status()
+                return response.content
+        except Exception as e:
+            logger.error("Cloudinary download failed: %s", str(e), exc_info=True)
+            raise StorageDownloadError(f"Failed to download file from storage: {str(e)}") from e

backend/app/utils/default_providers.py

@@ -10,6 +10,11 @@ def default_storage_provider() -> StorageProviderInterface:
 
         return SupabaseStorageProvider()
 
+    if settings.STORAGE_PROVIDER == "cloudinary":
+        from app.utils.cloudinary_provider import CloudinaryStorageProvider
+
+        return CloudinaryStorageProvider()
+
     raise ValueError(f"Unknown storage provider: '{settings.STORAGE_PROVIDER}'")
 
 

backend/pyproject.toml

@@ -27,6 +27,7 @@ dependencies = [
     "taskiq-redis>=0.5.0",
     "python-multipart>=0.0.27",
     "supabase>=2.30.0",
+    "cloudinary>=1.44.0",
     "authlib>=1.7.2",
     "httpx>=0.28.1",
     "itsdangerous>=2.2.0",

backend/tests/conftest.py

@@ -48,6 +48,7 @@ async def create_tables() -> AsyncGenerator[None, None]:
     """Create all ORM tables once per test session, drop them afterwards."""
     # Import all models so Base.metadata knows about them.
     import app.models.application  # noqa: F401
+    import app.models.dsa_question  # noqa: F401
     import app.models.interaction  # noqa: F401
     import app.models.interview  # noqa: F401
     import app.models.organization  # noqa: F401

backend/tests/test_utils/test_cloudinary_provider.py

@@ -0,0 +1,144 @@
+from typing import Any
+
+import pytest
+
+from app.exceptions.storage import StorageDeleteError, StorageDownloadError, StorageUploadError
+from app.utils.cloudinary_provider import CloudinaryStorageProvider
+
+
+class _FakeResponse:
+    def __init__(self, content: bytes = b"pdf-bytes", should_raise: bool = False):
+        self.content = content
+        self.should_raise = should_raise
+
+    def raise_for_status(self) -> None:
+        if self.should_raise:
+            raise RuntimeError("download failed")
+
+
+class _FakeAsyncClient:
+    def __init__(self, response: _FakeResponse):
+        self.response = response
+        self.requested_url = ""
+
+    async def __aenter__(self) -> "_FakeAsyncClient":
+        return self
+
+    async def __aexit__(self, *args: Any) -> None:
+        return None
+
+    async def get(self, url: str) -> _FakeResponse:
+        self.requested_url = url
+        return self.response
+
+
+@pytest.mark.asyncio
+async def test_upload_stores_pdf_as_raw_asset(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls: dict[str, Any] = {}
+
+    def fake_upload(file: Any, **kwargs: Any) -> dict[str, str]:
+        calls["payload"] = file.read()
+        calls["kwargs"] = kwargs
+        return {"secure_url": "https://res.cloudinary.com/demo/raw/upload/resumes/resume.pdf"}
+
+    monkeypatch.setattr("app.utils.cloudinary_provider.cloudinary.uploader.upload", fake_upload)
+
+    provider = CloudinaryStorageProvider(
+        cloud_name="demo",
+        api_key="key",
+        api_secret="secret",
+        folder="resumes",
+    )
+
+    result = await provider.upload(b"%PDF", "resume.pdf")
+
+    assert result == "https://res.cloudinary.com/demo/raw/upload/resumes/resume.pdf"
+    assert calls["payload"] == b"%PDF"
+    assert calls["kwargs"] == {
+        "public_id": "resumes/resume.pdf",
+        "resource_type": "raw",
+        "overwrite": True,
+    }
+
+
+@pytest.mark.asyncio
+async def test_upload_wraps_cloudinary_errors(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_upload(*_args: Any, **_kwargs: Any) -> dict[str, str]:
+        raise RuntimeError("bad credentials")
+
+    monkeypatch.setattr("app.utils.cloudinary_provider.cloudinary.uploader.upload", fake_upload)
+    provider = CloudinaryStorageProvider("demo", "key", "secret")
+
+    with pytest.raises(StorageUploadError):
+        await provider.upload(b"%PDF", "resume.pdf")
+
+
+@pytest.mark.asyncio
+async def test_delete_removes_raw_asset(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls: dict[str, Any] = {}
+
+    def fake_destroy(public_id: str, **kwargs: Any) -> dict[str, str]:
+        calls["public_id"] = public_id
+        calls["kwargs"] = kwargs
+        return {"result": "ok"}
+
+    monkeypatch.setattr("app.utils.cloudinary_provider.cloudinary.uploader.destroy", fake_destroy)
+    provider = CloudinaryStorageProvider("demo", "key", "secret", folder="resumes")
+
+    await provider.delete("resume.pdf")
+
+    assert calls == {
+        "public_id": "resumes/resume.pdf",
+        "kwargs": {"resource_type": "raw"},
+    }
+
+
+@pytest.mark.asyncio
+async def test_delete_wraps_failed_result(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_destroy(*_args: Any, **_kwargs: Any) -> dict[str, str]:
+        return {"result": "error"}
+
+    monkeypatch.setattr("app.utils.cloudinary_provider.cloudinary.uploader.destroy", fake_destroy)
+    provider = CloudinaryStorageProvider("demo", "key", "secret")
+
+    with pytest.raises(StorageDeleteError):
+        await provider.delete("resume.pdf")
+
+
+@pytest.mark.asyncio
+async def test_download_fetches_cloudinary_raw_url(monkeypatch: pytest.MonkeyPatch) -> None:
+    fake_client = _FakeAsyncClient(_FakeResponse(content=b"%PDF"))
+
+    def fake_cloudinary_url(public_id: str, **kwargs: Any) -> tuple[str, dict[str, Any]]:
+        assert public_id == "resumes/resume.pdf"
+        assert kwargs == {"resource_type": "raw", "secure": True}
+        return "https://res.cloudinary.com/demo/raw/upload/resumes/resume.pdf", {}
+
+    monkeypatch.setattr(
+        "app.utils.cloudinary_provider.cloudinary.utils.cloudinary_url", fake_cloudinary_url
+    )
+    monkeypatch.setattr("app.utils.cloudinary_provider.httpx.AsyncClient", lambda: fake_client)
+    provider = CloudinaryStorageProvider("demo", "key", "secret", folder="resumes")
+
+    result = await provider.download("resume.pdf")
+
+    assert result == b"%PDF"
+    assert (
+        fake_client.requested_url == "https://res.cloudinary.com/demo/raw/upload/resumes/resume.pdf"
+    )
+
+
+@pytest.mark.asyncio
+async def test_download_wraps_http_errors(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "app.utils.cloudinary_provider.cloudinary.utils.cloudinary_url",
+        lambda *_args, **_kwargs: ("https://example.test/resume.pdf", {}),
+    )
+    monkeypatch.setattr(
+        "app.utils.cloudinary_provider.httpx.AsyncClient",
+        lambda: _FakeAsyncClient(_FakeResponse(should_raise=True)),
+    )
+    provider = CloudinaryStorageProvider("demo", "key", "secret")
+
+    with pytest.raises(StorageDownloadError):
+        await provider.download("resume.pdf")