Safety Module EN

繁體中文版

Safety Module — beam_pipeline_toolkit/safety/

Two small, independent guardrails used throughout the rest of the toolkit.

leak_guard.py

A recursive forbidden-key scanner for sanctioned outputs (metric rows, telemetry events, run manifests — anything about to be persisted or transmitted), not for raw records flowing internally through a pipeline (which legitimately carry things like a customer id).

• LeakGuard(forbidden_substrings=..., allowed_key_exceptions=...) — construct your own per app/domain.
• assert_safe(obj) — recursively raises SafetyViolation if any dict key anywhere in the object graph (dict/list/tuple) contains a forbidden substring (case-insensitive). Rejects outright rather than silently stripping the key — silent stripping could hide a real integration bug.
• is_safe(obj) — non-raising boolean wrapper.
• default_guard / module-level assert_safe/is_safe — a ready-to-use instance with DEFAULT_FORBIDDEN_KEY_SUBSTRINGS (email, phone, customer_id, token, secret, credential, traceback, raw, ...) and no app-specific exceptions. This is what pipeline/aggregation.py, pipeline/telemetry.py, pipeline/sinks.py, etc. all call. An app with its own legitimate field names should build its own LeakGuard instead of editing this default.

cloud_gate.py

assert_gates_open(*, flags, confirmation=None, expected_confirmation=None, confirmation_flag_name="--confirm-execution") — the single choke point every cloud-resource-building code path (a Pub/Sub client, a BigQuery client, a Dataflow submission) must call immediately before constructing a client, not earlier. Raises CloudGateError listing every missing/invalid gate unless every flag in flags is True and (if given) confirmation == expected_confirmation exactly. Supports any number of flags — a two-flag or three-gate convention both fit the same function.