DataOps-platform/deployment/app/core/common/identifiers.py

from __future__ import annotations

import secrets
import time
import uuid
from typing import Any, MutableMapping


def new_governance_uid() -> str:
    """Return an RFC 9562 UUIDv7-compatible identifier."""
    timestamp_ms = int(time.time_ns() // 1_000_000) & ((1 << 48) - 1)
    random_a = secrets.randbits(12)
    random_b = secrets.randbits(62)
    value = (
        (timestamp_ms << 80)
        | (0x7 << 76)
        | (random_a << 64)
        | (0b10 << 62)
        | random_b
    )
    return str(uuid.UUID(int=value))


def ensure_governance_uid(properties: MutableMapping[str, Any]) -> str:
    """Attach a UUIDv7 to mutable node properties and preserve it on retries."""
    existing = properties.get("uid")
    if existing:
        try:
            parsed = uuid.UUID(str(existing))
        except (ValueError, AttributeError, TypeError) as exc:
            raise ValueError("governance uid must be a valid UUIDv7") from exc
        if parsed.version != 7 or parsed.variant != uuid.RFC_4122:
            raise ValueError("governance uid must be a valid UUIDv7")
        value = str(parsed)
    else:
        value = new_governance_uid()
    properties["uid"] = value
    return value