herfy-gcs

GCS storage SDK for Herfy applications.
Files are uploaded to and downloaded from Google Cloud Storage through the Herfy Control Center — apps never handle raw GCS credentials directly.

How it works

App ──(client_id + secret)──► Control Center ──(service account)──► GCS bucket

1. Your app registered with the Control Center and received a client_id + client_secret.
2. The SDK exchanges those credentials for a short-lived CC access token (OAuth2 client credentials flow).
3. Upload / download requests go to the CC storage API; the CC proxies them to GCS under its own service account.
4. The CC enforces per-app bucket/prefix scoping — your app can only touch its own folder.

Installation

# From PyPI (once published)
pip install herfy-gcs==0.1.0

# Local editable install (monorepo / during development)
pip install -e /path/to/herfy-shared/herfy-gcs

Quick start

Environment variables

CONTROL_CENTER_URL=https://controlcenter-xxx.run.app
AUTH_CLIENT_ID=app_your_client_id        # from CC registration
AUTH_CLIENT_SECRET=your_client_secret    # from CC registration

Optional overrides:

Sync client (default — works in FastAPI sync endpoints)

from herfy_gcs import HerfyGCSClient

client = HerfyGCSClient.from_env()

# Upload a file
with open("deposit.pdf", "rb") as f:
    result = client.upload(
        content=f,
        filename="deposit.pdf",
        content_type="application/pdf",
        folder="daily-cash/2025",
    )

print(result.file_ref)   # store this in your database

# Get a signed download URL (valid for 1 hour)
url = client.get_download_url(result.file_ref, expires_in=3600)

# Download raw bytes
data = client.download(result.file_ref)

# Resolve any existing reference (GCS URL or CC ref) to a usable URL
url = client.resolve_url(existing_db_ref)

# Delete a file
client.delete(result.file_ref)

Async client (for async FastAPI endpoints)

from herfy_gcs import AsyncHerfyGCSClient

client = AsyncHerfyGCSClient.from_env()

result = await client.upload(content=pdf_bytes, filename="doc.pdf", folder="daily-cash")
url = await client.get_download_url(result.file_ref)
data = await client.download(result.file_ref)

Explicit credentials (no env vars)

client = HerfyGCSClient.from_credentials(
    client_id="app_7c144431f50c",
    client_secret="your-secret",
    control_center_url="https://controlcenter-xxx.run.app",
)

FastAPI dependency example

from functools import lru_cache
from fastapi import Depends
from herfy_gcs import HerfyGCSClient

@lru_cache(maxsize=1)
def get_gcs_client() -> HerfyGCSClient:
    return HerfyGCSClient.from_env()

@app.post("/upload")
async def upload_file(
    file: UploadFile,
    gcs: HerfyGCSClient = Depends(get_gcs_client),
):
    result = gcs.upload(
        content=await file.read(),
        filename=file.filename,
        content_type=file.content_type or "application/octet-stream",
        folder="daily-cash",
    )
    return {"file_ref": result.file_ref, "url": result.url}

Control Center storage API

The CC must expose these endpoints (authenticated with a CC bearer token):

Upload request

POST /api/storage/upload
Authorization: Bearer <cc_token>
Content-Type: multipart/form-data

file=<bytes>
filename=deposit.pdf
folder=daily-cash/2025

Response 200 OK:

{
  "file_ref": "cc:bucket_herfy_datacloud/daily-cash/2025/deposit.pdf",
  "url": "https://storage.googleapis.com/bucket_herfy_datacloud/...",
  "filename": "deposit.pdf",
  "content_type": "application/pdf",
  "size": 204800,
  "folder": "daily-cash/2025"
}

Signed URL request

POST /api/storage/signed-url
Authorization: Bearer <cc_token>
Content-Type: application/json

{"file_ref": "cc:...", "expires_in": 3600}

Response 200 OK:

{"url": "https://storage.googleapis.com/...?X-Goog-Signature=...", "file_ref": "cc:...", "expires_in": 3600}

Error handling

from herfy_gcs import HerfyGCSClient, UploadError, AuthError, FileNotFoundError, GCSError

client = HerfyGCSClient.from_env()

try:
    result = client.upload(content=data, filename="doc.pdf")
except AuthError as e:
    # Invalid client_id/secret or CC unreachable
    print(f"Auth failed: {e.message}")
except UploadError as e:
    print(f"Upload failed (HTTP {e.status_code}): {e.message}")
except GCSError as e:
    print(f"Storage error: {e.message}")

License

MIT