dodil/erp-brick-expenses

DODIL ERP brick: expenses

src/brick_expenses/infrastructure/objects.py

"""The receipt store: where receipt images are kept as DataK3 objects (raw/expenses/<report_id>/<line_no>/<sha256>.<ext>).

GAP (recorded, not papered over): erp-platform 0.9.13 has no object helper - its capabilities (capabilities.yaml) are an outside-world seam, not
a store, and brick-gl wrote its own SigV4 client for the same endpoint (gl infrastructure/objects.py, which this one follows). So this brick
declares the PORT it needs - `ReceiptStore.put / get` - and ships one implementation of it, DataK3's S3 endpoint
(OBJECT_ENDPOINT, BUCKET, OBJECT_ACCESS_KEY, OBJECT_SECRET_KEY, OBJECT_REGION; object.uk-lon-1.dodil.io by default). When the platform grows the
helper, `S3Store` is the only thing to delete. No endpoint configured is a 503 for the use case that needs it (add_receipt, get_receipt), never a
silent drop: a receipt that was not stored is not a receipt.

An object is kept ONCE by key: a key that exists is never overwritten (evidence is not replaced), and since the key carries the image's sha256,
storing the same image again is the same object.
"""
import hashlib
import hmac
import os
from typing import Optional, Protocol


class StoreDown(Exception):
    """The store could not answer. `http` is what the use case answers: 503 not configured / silent, 502 failed."""

    def __init__(self, http: int, reason: str):
        super().__init__(reason)
        self.http, self.reason = http, reason


class ReceiptStore(Protocol):
    def put(self, key: str, body: bytes, content_type: str) -> str: ...
    def get(self, key: str) -> Optional[bytes]: ...


class S3Store:
    """DataK3's S3 endpoint, SigV4, over httpx. Configured from the environment."""

    def __init__(self):
        self.endpoint = os.environ.get("OBJECT_ENDPOINT", "https://object.uk-lon-1.dodil.io").rstrip("/")
        self.bucket = os.environ.get("BUCKET", "")
        self.access, self.secret = os.environ.get("OBJECT_ACCESS_KEY", ""), os.environ.get("OBJECT_SECRET_KEY", "")
        self.region = os.environ.get("OBJECT_REGION", "us-east-1")

    @property
    def configured(self) -> bool:
        return bool(self.bucket and self.access and self.secret)

    def _signed(self, method: str, key: str, body: bytes, content_type: str = "") -> tuple:
        import datetime as dt
        from urllib.parse import quote, urlparse
        host = urlparse(self.endpoint).netloc
        path = "/" + quote(f"{self.bucket}/{key}")
        now = dt.datetime.now(dt.timezone.utc)
        amz, day = now.strftime("%Y%m%dT%H%M%SZ"), now.strftime("%Y%m%d")
        payload = hashlib.sha256(body).hexdigest()
        headers = {"host": host, "x-amz-content-sha256": payload, "x-amz-date": amz}
        if content_type:
            headers["content-type"] = content_type
        signed = ";".join(sorted(headers))
        canon = "\n".join([method, path, "", "".join(f"{k}:{headers[k]}\n" for k in sorted(headers)), signed, payload])
        scope = f"{day}/{self.region}/s3/aws4_request"
        sts = "\n".join(["AWS4-HMAC-SHA256", amz, scope, hashlib.sha256(canon.encode()).hexdigest()])

        def h(k, m):
            return hmac.new(k, m.encode(), hashlib.sha256).digest()
        k = h(h(h(h(("AWS4" + self.secret).encode(), day), self.region), "s3"), "aws4_request")
        headers["authorization"] = (f"AWS4-HMAC-SHA256 Credential={self.access}/{scope}, SignedHeaders={signed}, "
                                    f"Signature={hmac.new(k, sts.encode(), hashlib.sha256).hexdigest()}")
        return f"{self.endpoint}{path}", headers

    def get(self, key: str) -> Optional[bytes]:
        import httpx
        url, headers = self._signed("GET", key, b"")
        try:
            r = httpx.get(url, headers=headers, timeout=30)
        except httpx.HTTPError as e:
            raise StoreDown(503, f"the object store did not answer ({e.__class__.__name__})") from None
        if r.status_code == 404:
            return None
        if r.status_code != 200:
            raise StoreDown(502, f"the object store answered {r.status_code}")
        return r.content

    def put(self, key: str, body: bytes, content_type: str) -> str:
        import httpx
        if self.get(key) is not None:
            return key
        url, headers = self._signed("PUT", key, body, content_type)
        try:
            r = httpx.put(url, headers=headers, content=body, timeout=30)
        except httpx.HTTPError as e:
            raise StoreDown(503, f"the object store did not answer ({e.__class__.__name__})") from None
        if r.status_code >= 300:
            raise StoreDown(502, f"the object store answered {r.status_code}")
        return key


class MemoryStore:
    """For tests and local runs: a dict."""

    def __init__(self):
        self.objects: dict = {}
        self.puts = 0
        self.down: Optional[StoreDown] = None

    def put(self, key: str, body: bytes, content_type: str) -> str:
        if self.down:
            raise self.down
        self.puts += 1
        self.objects.setdefault(key, (body, content_type))
        return key

    def get(self, key: str) -> Optional[bytes]:
        if self.down:
            raise self.down
        got = self.objects.get(key)
        return got[0] if got else None


class Ports:
    """`store` is replaceable (tests do); None means S3 from the environment, and an unconfigured environment is a 503 naming what to set."""

    def __init__(self):
        self.store: Optional[ReceiptStore] = None

    def receipts(self) -> ReceiptStore:
        if self.store is not None:
            return self.store
        s3 = S3Store()
        if not s3.configured:
            raise StoreDown(503, "no receipt store is configured: set OBJECT_ENDPOINT, BUCKET, OBJECT_ACCESS_KEY and OBJECT_SECRET_KEY")
        return s3


ports = Ports()