DataOps-platform/app/core/system/tenant_resources.py

"""Tenant namespace and hash-only manifest contracts for non-PostgreSQL stores."""

from __future__ import annotations

import re
from typing import Any

from app.core.system.tenant_context import TenantScope


class TenantResourceError(ValueError):
    pass


_RESOURCE_TYPES = frozenset(
    {
        "object",
        "graph",
        "cache",
        "index",
        "keys",
        "connectors",
        "models",
        "plugins",
        "backup",
        "audit_export",
    }
)
_RESOURCE_NAME = re.compile(r"^[a-z0-9][a-z0-9._-]{0,120}$")
_DIGEST = re.compile(r"^[0-9a-f]{64}$")


def _resource_token(value: str, *, allowed: frozenset[str] | None = None) -> str:
    if not isinstance(value, str) or not _RESOURCE_NAME.fullmatch(value):
        raise TenantResourceError("tenant_resource_invalid")
    if allowed is not None and value not in allowed:
        raise TenantResourceError("tenant_resource_invalid")
    return value


def build_tenant_namespace(scope: TenantScope, resource: str, name: str) -> str:
    """Return an opaque prefix; callers must not concatenate their own paths."""
    resource = _resource_token(resource, allowed=_RESOURCE_TYPES)
    name = _resource_token(name)
    return f"{scope.tenant_id}/{resource}/v{scope.version}/{name}"


def build_tenant_resource_manifest(
    scope: TenantScope,
    *,
    resource: str,
    name: str,
    digest: str,
) -> dict[str, Any]:
    if not isinstance(digest, str) or not _DIGEST.fullmatch(digest):
        raise TenantResourceError("tenant_resource_invalid")
    namespace = build_tenant_namespace(scope, resource, name)
    return {
        "schema_version": 1,
        "tenant_id": scope.tenant_id,
        "delivery_mode": scope.mode.value,
        "resource": resource,
        "namespace": namespace,
        "digest": f"sha256:{digest}",
        "provider_enabled": False,
    }