Skip to content

fastflowtransform.api.context

reset_for_node 

reset_for_node(node_name)

Reset the current context stats and set the node name. Equivalent to: _get_http_ctx().reset_for_node(node_name)

Source code in src/fastflowtransform/api/context.py 
 96
 97
 98
 99
100
101
def reset_for_node(node_name: str) -> None:
    """
    Reset the current context stats and set the node name.
    Equivalent to: _get_http_ctx().reset_for_node(node_name)
    """
    _get_http_ctx().reset_for_node(node_name)

record 

record(io_key, content_hash, cache_hit, byte_len, used_offline)

Record a single HTTP IO event into the current context.

Parameters

io_key : str Stable IO cache key or identifier. content_hash : str Hash of the content (used for diagnostics). cache_hit : bool Whether the response came from cache. byte_len : int Number of response bytes (>= 0). used_offline : bool Whether the response was served in offline mode.

Source code in src/fastflowtransform/api/context.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def record(
    io_key: str,
    content_hash: str,
    cache_hit: bool,
    byte_len: int,
    used_offline: bool,
) -> None:
    """
    Record a single HTTP IO event into the current context.

    Parameters
    ----------
    io_key : str
        Stable IO cache key or identifier.
    content_hash : str
        Hash of the content (used for diagnostics).
    cache_hit : bool
        Whether the response came from cache.
    byte_len : int
        Number of response bytes (>= 0).
    used_offline : bool
        Whether the response was served in offline mode.
    """
    stats = _get_http_ctx()
    stats.requests += 1
    if cache_hit:
        stats.cache_hits += 1
    stats.bytes += int(byte_len or 0)

    # Deduplicate consecutive repeats while preserving order.
    if io_key and (not stats.keys or stats.keys[-1] != io_key):
        stats.keys.append(io_key)
    if content_hash and (not stats.hashes or stats.hashes[-1] != content_hash):
        stats.hashes.append(content_hash)

    stats.used_offline = bool(used_offline)

snapshot 

snapshot()

Return a snapshot dict of the current context. If the context is not yet initialized, return an empty/default snapshot instead of raising.

Source code in src/fastflowtransform/api/context.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def snapshot() -> dict[str, Any]:
    """
    Return a snapshot dict of the current context. If the context is not yet
    initialized, return an empty/default snapshot instead of raising.
    """
    s = _ctx.get()
    if s is None:
        return {
            "node": None,
            "requests": 0,
            "cache_hits": 0,
            "bytes": 0,
            "used_offline": False,
            "content_hashes": [],
            "keys": [],
        }
    return s.snapshot()