oc_diagnostics package

oc_diagnostics package#

oc_diagnostics#

Purpose#

Package entry point for oc_diagnostics. Exports the public runtime API so downstream projects can use a single import:

from oc_diagnostics import _dbg, init, apply_instrumentation

Responsibilities#

  • Re-export _dbg, DEBUG_LEVELS, and init from unified_debug.

  • Re-export apply_instrumentation from instrumentation_middleware when Dash is installed; raise an actionable ImportError at call time when it is not.

  • Re-export DiagnosticsMiddleware from fastapi_middleware when FastAPI is installed; raise an actionable ImportError at call time when it is not.

  • Declare __all__ so the public surface is explicit and IDE-discoverable.

  • Remain completely side-effect-free at import time.

Diagnostics#

Domain: GENERAL Levels:

L2 — lifecycle L3 — semantic details L4 — deep tracing

Contracts#

  • Must not activate stream interception or exception hooks at import time. Callers must call oc_diagnostics.init() explicitly.

  • Optional dependencies (Dash, FastAPI) must not be required for import. Missing dependencies must only raise ImportError at call time.

  • __all__ must list every symbol exported by this module.

class oc_diagnostics.DiagnosticsMiddleware(app, domain: str = 'HTTP')[source]#

Bases: object

Pure ASGI middleware that logs HTTP requests and responses via _dbg().

Parameters:

  • app (ASGI callable) – The next ASGI application in the middleware stack.

  • domain (str, optional) – Debug domain used for all log output. Defaults to "HTTP". Must be configured in diagnostics/debug_config.json with a non-zero level; unknown domains are silently filtered by _dbg().

  • Usage

  • -----

  • registered:: (Add to a FastAPI application after all routes are) –

    from fastapi import FastAPI from oc_diagnostics import DiagnosticsMiddleware

    app = FastAPI() app.add_middleware(DiagnosticsMiddleware)

  • domain:: (With a custom) – app.add_middleware(DiagnosticsMiddleware, domain=”API”)

  • diagnostics/debug_config.json:: (Enable output by adding the HTTP domain to) – “HTTP”: { “level”: 2, “color”: “blue” }

  • middleware (Level guide for this) – L2 — request/response lifecycle lines (always recommended) L3 — User-Agent, 4xx/5xx status class commentary L4 — full request and response header dumps

class oc_diagnostics.OcDiagnosticsHandler(domain: str = 'GENERAL', level_map: dict[int, int] | None = None)[source]#

Bases: Handler

A logging.Handler that routes records through oc_diagnostics._dbg().

Parameters:

  • domain (str) – The oc_diagnostics domain used for all records routed through this handler. Default "GENERAL".

  • level_map (dict[int, int] | None) – Mapping from Python logging level integers to oc_diagnostics diagnostic level integers. When None, the built-in default is used: DEBUG→4, INFO→2, WARNING/ERROR/CRITICAL→1.

Example

import logging
from oc_diagnostics import OcDiagnosticsHandler

handler = OcDiagnosticsHandler(domain="DATA")
logging.getLogger("sqlalchemy").addHandler(handler)
emit(record: LogRecord) → None[source]#

Do whatever it takes to actually log the specified logging record.

This version is intended to be implemented by subclasses and so raises a NotImplementedError.

oc_diagnostics.apply_instrumentation(component)[source]#

Apply ID instrumentation to a component tree if enabled.

  • No manual recursion (instrument_ids handles that)

  • No structural ID rewrites

  • Logs only when instrumentation debug level is high enough

oc_diagnostics.dbg_metrics(domain: str, level: int = 3)[source]#

Decorator that logs structured performance metrics (time + memory).

Emits a JSON-style string with duration_ms and memory_mb fields suitable for observability tooling.

Parameters:

  • domain (str) – Debug domain for the metrics message.

  • level (int) – Debug level (default 3).

oc_diagnostics.dbg_profile(domain: str = 'GENERAL', func: str = '', label: str = '', level: int = 3)[source]#

Decorator that wraps a function with dbg_timed().

Example:

@dbg_profile("CONTROLLER", label="fetch_data")
def fetch_data():
    ...
oc_diagnostics.dbg_scope(domain: str, func: str, label: str = '', level: int = 2)[source]#

Context manager that logs ENTER and EXIT at the specified level without timing.

Use for pure lifecycle tracing where elapsed time is not needed. For timed tracing, use dbg_timed() instead.

Example:

with dbg_scope("DATA", "load_records"):
    result = expensive_operation()
oc_diagnostics.dbg_timed(domain: str, func: str, label: str = '', level: int = 3)[source]#

Context manager that logs ENTER and EXIT with elapsed time in milliseconds.

Example:

with dbg_timed("CONTROLLER", "my_func", label="heavy_op"):
    do_work()
oc_diagnostics.dbg_trace(domain: str, level: int = 2)[source]#

Decorator that emits START/END lifecycle messages and catches exceptions.

Injects a func local variable set to the decorated function’s __name__ so that inner _dbg calls can reference it without hardcoding a string.

Parameters:

  • domain (str) – Debug domain for lifecycle messages.

  • level (int) – Debug level for START/END messages (default 2).

Example

@dbg_trace("DATA")
def process_data(data):
    _dbg("DATA", func, "working...", 3)
    return transform(data)
oc_diagnostics.init(intercept_streams: bool = True, intercept_exceptions: bool = True, watch_config: bool = False) → None[source]#

Initialize oc_diagnostics side effects.

Must be called explicitly after importing oc_diagnostics. Importing the package does NOT activate stream interception or the custom exception hook automatically.

Parameters:

  • intercept_streams (bool) – If True, replace sys.stdout and sys.stderr with _StreamInterceptor instances that route all print() calls through _dbg(). Can also be disabled by setting the environment variable UNIFIED_DEBUG_DISABLE_STDIO_INTERCEPT=1 before calling init(). Forced to False in production mode (PROD_MODE=True). Default True.

  • intercept_exceptions (bool) – If True, install a custom sys.excepthook that logs uncaught exceptions through _dbg() and then delegates to the previously active hook. Can also be disabled by setting the environment variable UNIFIED_DEBUG_DISABLE_EXCEPTION_HOOK=1 before calling init(). Forced to False in production mode (PROD_MODE=True). Default True.

  • watch_config (bool) – If True, start a daemon thread that polls the loaded config file every 2 seconds and reloads it when the mtime changes. Useful during development to adjust log levels without restarting the application. Default False. (FS-09)

Notes

Calling init() more than once is a no-op.

oc_diagnostics.set_debug_level(domain: str, level: int) → None[source]#

Set the debug level for a domain at runtime (0–9).

Updates DEBUG_LEVELS immediately; takes effect on the next _dbg() call. Emits a UserWarning and returns without modifying anything if level is not an integer in 0–9.

oc_diagnostics.set_global_setting(key: str, value) → None[source]#

Update a global diagnostic setting at runtime.

Valid keys: silent_mode, enable_timestamps, enable_colors, include_filename, output_mode. Emits a UserWarning for unknown keys and returns without modifying anything.

Submodules#