This documentation was generated with AI assistance and is still being audited. Some, or potentially a lot, of this information may be inaccurate. Learn more.
defclose_log_streams()->None:"""Close file log streams and reset to stderr."""importprovide.foundation.streams.coreascore_module# Import here to avoid circular dependencyfromprovide.foundation.testmode.detectionimportis_in_click_testingwith_get_stream_lock():ifcore_module._LOG_FILE_HANDLE:withcontextlib.suppress(Exception):core_module._LOG_FILE_HANDLE.close()core_module._LOG_FILE_HANDLE=None# Don't reset stream to stderr if we're in Click testing contextifnotis_in_click_testing():core_module._PROVIDE_LOG_STREAM=sys.stderr# Reconfigure structlog to use stderr_reconfigure_structlog_stream()
defconfigure_file_logging(log_file_path:str|None)->None:"""Configure file logging if a path is provided. Args: log_file_path: Path to log file, or None to disable file logging """# Import core module to modify the actual global variablesimportprovide.foundation.streams.coreascore_module# Import here to avoid circular dependencyfromprovide.foundation.testmode.detectionimportis_in_click_testingwith_get_stream_lock():# Don't modify streams if we're in Click testing contextifis_in_click_testing():return# Close existing file handle if it existsif(core_module._LOG_FILE_HANDLEandcore_module._LOG_FILE_HANDLEisnotcore_module._PROVIDE_LOG_STREAM):withcontextlib.suppress(Exception):core_module._LOG_FILE_HANDLE.close()core_module._LOG_FILE_HANDLE=None# Check if we're in testing modeis_test_stream=core_module._PROVIDE_LOG_STREAMisnotsys.stderrandnotisinstance(core_module._PROVIDE_LOG_STREAM,io.TextIOWrapper,)iflog_file_path:try:Path(log_file_path).parent.mkdir(parents=True,exist_ok=True)core_module._LOG_FILE_HANDLE=Path(log_file_path).open("a",encoding="utf-8",buffering=1)core_module._PROVIDE_LOG_STREAM=core_module._LOG_FILE_HANDLE# Reconfigure structlog to use the new file stream_reconfigure_structlog_stream()exceptExceptionase:# Log error to stderr and fall back_safe_error_output(f"Failed to open log file {log_file_path}: {e}")core_module._PROVIDE_LOG_STREAM=get_safe_stderr()# Reconfigure structlog to use stderr fallback_reconfigure_structlog_stream()elifnotis_test_stream:core_module._PROVIDE_LOG_STREAM=get_safe_stderr()# Reconfigure structlog to use stderr_reconfigure_structlog_stream()
defflush_log_streams()->None:"""Flush all log streams."""importprovide.foundation.streams.coreascore_modulewith_get_stream_lock():ifcore_module._LOG_FILE_HANDLE:try:core_module._LOG_FILE_HANDLE.flush()exceptExceptionase:_safe_error_output(f"Failed to flush log file handle: {e}")
defreset_streams()->None:"""Reset all stream state (for testing)."""# Import here to avoid circular dependencyfromprovide.foundation.testmode.detectionimportis_in_click_testing# Don't reset streams if we're in Click testing contextifnotis_in_click_testing():close_log_streams()