Skip to content

graflo.db.tigergraph.token_cache

Process-scoped TigerGraph API token cache.

Attributes

TokenCacheKey = tuple[str, str, str] module-attribute

Functions:

make_token_cache_key(gsql_url, graphname, secret)

Source code in graflo/db/tigergraph/token_cache.py 
94
95
96
def make_token_cache_key(gsql_url: str, graphname: str, secret: str) -> TokenCacheKey:
    secret_fingerprint = hashlib.sha256(secret.encode()).hexdigest()[:16]
    return (gsql_url, graphname, secret_fingerprint)

parse_tg_expiration(raw)

Convert TigerGraph expiration field to a Unix timestamp.

Source code in graflo/db/tigergraph/token_cache.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def parse_tg_expiration(raw: str | None) -> float | None:
    """Convert TigerGraph expiration field to a Unix timestamp."""
    if raw is None:
        return None
    try:
        return float(raw)
    except (ValueError, TypeError):
        pass
    try:
        dt = datetime.fromisoformat(str(raw))
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=UTC)
        return dt.timestamp()
    except (ValueError, TypeError):
        return None

reset_tigergraph_token_cache()

Clear the process token cache (for tests and long-lived REPL sessions).

Source code in graflo/db/tigergraph/token_cache.py 
 99
100
101
102
103
104
def reset_tigergraph_token_cache() -> None:
    """Clear the process token cache (for tests and long-lived REPL sessions)."""
    with _TigerGraphTokenCache._instance_lock:
        if _TigerGraphTokenCache._instance is not None:
            _TigerGraphTokenCache._instance.clear()
        _TigerGraphTokenCache._instance = None