oss_store.py

"""oss_store -- OPTIONAL Alibaba Cloud OSS persistence for Flashbulb memory snapshots.

A session's learned neural-memory weights + explicit store can be snapshotted to Alibaba
Cloud Object Storage Service (OSS), so memory is durable off-box and survives a restart.
All credentials + the bucket/endpoint are read from the environment; nothing is hardcoded,
and no secret ever lives in source.

OPTIONAL SINCE D-022. The original hackathon deployment ran on Alibaba ECS and used this
module as the "backend runs on Alibaba Cloud services" proof. That deployment is retired.
The `oss2` SDK is no longer a required dependency: when it is absent, or when the env vars
below are unset, `oss_configured()` returns False and app.py transparently falls back to
local-file snapshots. The cross-session boundary demo works either way -- only the
reported backend label changes ("alibaba-oss" vs "local-only").

Env expected when you DO want OSS (never committed):
  ALIBABA_CLOUD_ACCESS_KEY_ID, ALIBABA_CLOUD_ACCESS_KEY_SECRET  (RAM user with OSS access)
  OSS_ENDPOINT   e.g. https://oss-ap-southeast-1.aliyuncs.com  (region of your bucket)
  OSS_BUCKET     the bucket name

Kept OUT of memory/sidecar.py on purpose: the memory core stays dependency-clean (D-008);
cloud persistence is a thin wrapper the app calls, never the core.
"""
from __future__ import annotations

import os

try:  # optional dependency -- absent on the default (no-Alibaba) install
    import oss2  # Alibaba Cloud OSS Python SDK
except ImportError:  # pragma: no cover - exercised only on installs without oss2
    oss2 = None


def _bucket() -> "oss2.Bucket":
    """Build an OSS bucket handle from environment credentials. Raises a clear error if
    the SDK or any credential is missing, rather than failing deep inside the SDK."""
    if oss2 is None:
        raise RuntimeError(
            "the 'oss2' SDK is not installed; run `pip install oss2` to enable Alibaba OSS "
            "snapshots, or leave it out to use local-file snapshots (see this module's docstring)."
        )
    try:
        ak = os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"]
        sk = os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"]
        bucket_name = os.environ["OSS_BUCKET"]
    except KeyError as e:
        raise RuntimeError(f"OSS credential {e} not set; see cloud/oss_store.py docstring.") from e
    endpoint = os.environ.get("OSS_ENDPOINT", "https://oss-ap-southeast-1.aliyuncs.com")
    return oss2.Bucket(oss2.Auth(ak, sk), endpoint, bucket_name)


def oss_configured() -> bool:
    """True only when the SDK is installed AND all OSS env vars are present -- lets the app
    treat cloud snapshots as an optional enhancement without branching everywhere."""
    return oss2 is not None and all(os.environ.get(k) for k in
               ("ALIBABA_CLOUD_ACCESS_KEY_ID", "ALIBABA_CLOUD_ACCESS_KEY_SECRET", "OSS_BUCKET"))


def upload_snapshot(local_path: str, key: str) -> str:
    """Push a local memory-snapshot file to OSS under `key`. Returns the object key."""
    _bucket().put_object_from_file(key, local_path)
    return key


def download_snapshot(key: str, local_path: str) -> str:
    """Pull an OSS memory-snapshot object down to `local_path`. Returns local_path."""
    _bucket().get_object_to_file(key, local_path)
    return local_path


def snapshot_exists(key: str) -> bool:
    """Whether an object with `key` exists in the bucket (for resume-if-present logic)."""
    return _bucket().object_exists(key)