Skip to main content

easyfabric.fabric.logging_utils

ExcludeLoggersFilter Objects​

class ExcludeLoggersFilter(logging.Filter)

Drops records from EXCLUDED_LOGGERS so progress/metrics noise stays out of the OneLake log file (and therefore the logging table).

is_top_level_notebook​

def is_top_level_notebook() -> bool

Identifies if the current execution is the top-most notebook.

A notebook is top-level when it was not invoked via notebookutils.notebook.run from another notebook. Fabric sets isReferenceRun=True on every child invocation and False (or absent) on the entry-point notebook.

save_log_file_to_table​

def save_log_file_to_table(end_log: bool = False) -> None

Reads a log file from OneLake, parses structured logs (including multiline), and bulk inserts into Meta.dbo.logging using Spark DataFrame.

If it's a top-level notebook or end_log is True, logs an END entry with duration, persists log entries to the table and only then clears the logging handlers and resets state. Otherwise returns without emitting anything.

A failed persist stays visible: the failure is logged while the handlers are still attached and printed to the cell output, which survives both the handler cleanup and a log table that could not be written.

flush_log_handlers​

def flush_log_handlers() -> None

Write out every entry the log handlers are still holding. Never raises.

save_historical_log_file_to_table​

def save_historical_log_file_to_table(abfs_path: str) -> None

Parses a specific log file by ABFS path and inserts missing logs into Meta.dbo.logging.

FabricLoggerAdapter Objects​

class FabricLoggerAdapter(logging.LoggerAdapter)

Adapter that automatically includes log_type and log_category in all log records.

OneLakeFileHandler Objects​

class OneLakeFileHandler(logging.Handler)

Appends log entries to the OneLake file a whole batch shares.

notebookutils.fs.append is an offset write, so concurrent writers collide on InvalidFlushPosition. A write runs against a deadline instead of a try count, and an entry that does not land is held and written with the next one. No single write has to be won, so the deadline only has to outlast ordinary contention; waiting longer would hold the handler lock for nothing. Waiting is bounded per write and per run.

flush​

def flush()

Write what is still held. Takes the lock emit gets from handle.

SafeFormatter Objects​

class SafeFormatter(logging.Formatter)

Formatter that ensures custom fields exist to prevent KeyErrors from third-party libraries.

formatTime​

def formatTime(record, datefmt=None)

Include milliseconds in the timestamp for better sorting.

to_snake_case​

def to_snake_case(string: str) -> str

Convert a string from camel case to snake case.

set_verbose_mode​

def set_verbose_mode(enabled=True)

Enable or disable verbose logging mode globally.

SegmentHandle Objects​

class SegmentHandle()

Context manager returned by log_segment(). Accumulates a structured payload during the segment and emits it as |CTX:{...}| JSON on the END line. Four outcome states: Success (default), Skipped, Unchanged, Failed (exception escaped).

log_segment​

def log_segment(type: str, name: str) -> SegmentHandle

Context manager to log the start and end of a logic segment. Usage: with log_segment("Data Load", "Bronze Loading") as seg: seg.record(files=nr_of_files) if skipped_by_config: seg.skip(reason="loadskip_configured") return if nothing_to_do: seg.unchanged(reason="files_unchanged") return ... logic ...

current_segment​

def current_segment() -> Optional[SegmentHandle]

Return the innermost active segment, or None if outside a segment.

segment_record​

def segment_record(**kwargs) -> None

Record kwargs onto the current segment's payload. No-op if no segment is active.

segment_accumulate​

def segment_accumulate(**kwargs) -> None

Add kwargs to the current segment's payload, summing with any value already recorded under the same key. Use it for measures that a batch produces in parts (rows written per source file) so the END record carries the batch total. No-op if no segment is active.

segment_skip​

def segment_skip(reason: str) -> None

Mark the current segment as Skipped. No-op if no segment is active.

segment_unchanged​

def segment_unchanged(reason: str) -> None

Mark the current segment as Unchanged. No-op if no segment is active.

init_logging​

def init_logging(log_source: str = "Sys",
                 log_object: str = None,
                 base_batch_id: str = None) -> str

Call once at the very top of the entry-point notebook / wheel. Returns the absolute OneLake path of the log file.

base_batch_id keeps a whole logical run under one batch_id when a parent offloads children as separate Fabric jobs (each gets its own activityId). Pass the parent's id explicitly through every notebook boundary; absent it, the notebook's own activityId is used. The log file name always uses the own activityId so concurrent jobs write distinct files.

parent_run_id resolves to: the native Fabric parentRunId when present; else, for a reference run (notebook.run child), the own activityId — which equals the calling notebook's run id; else base_batch_id so an offloaded entry notebook links to the root run.

get_log_file_path​

def get_log_file_path() -> Optional[str]

Returns the path of the current log file. Checks singleton state, global config, and active handlers to ensure reliability even in nested notebooks or after module reloads.

extract_real_error​

def extract_real_error(log_text: str) -> str

Extracts the most relevant error message from a Spark stack trace.

A NotebookExecutionException (child notebook failure via notebookutils.notebook.run) carries the child's real error on the lines after the exception name, so that block is returned as a whole, up to the JVM stack frames.

Otherwise the most specific named Spark error wins; when a Caused by: chain is present its last link is appended to the primary line instead of replacing it, so the informative Python-level exception line survives next to the JVM root cause.