Python API¶

This page is generated from Python docstrings.

`log_searcher`¶

`octoprint_logmonitor.log_searcher` ¶

Log Searcher Module

Provides efficient log file searching with pagination and severity filtering. Memory-efficient implementation that streams through large log files.

`LogSearcher` ¶

Efficient log file searcher with pagination support.

Features: - Memory-efficient line-by-line reading - Free-text search (case-insensitive) - Severity level filtering - Pagination support - Context lines (lines before/after match) - Regex search mode (optional)

`init(logger=None)` ¶

Initialize the log searcher.

Parameters:

Name	Type	Description	Default
`logger`	`Optional[Any]`	Optional logger instance for debugging	`None`

`export_to_csv(results)` ¶

Export search results to CSV format.

Parameters:

Name	Type	Description	Default
`results`	`list[dict[str, Any]]`	List of search result dictionaries	required

Returns:

Type	Description
`str`	CSV string

`export_to_txt(results)` ¶

Export search results to plain text format.

Parameters:

Name	Type	Description	Default
`results`	`list[dict[str, Any]]`	List of search result dictionaries	required

Returns:

Type	Description
`str`	Plain text string

`get_file_stats(filepath)` ¶

Get statistics about a log file.

Parameters:

Name	Type	Description	Default
`filepath`	`str`	Path to the log file	required

Returns:

Type	Description
`dict[str, Any]`	Dictionary with file statistics

`search(filepath, query='', levels=None, offset=0, limit=50, case_sensitive=False, use_regex=False, context_lines=0)` ¶

Search log file for matching entries.

Parameters:

Name	Type	Description	Default
`filepath`	`str`	Path to the log file	required
`query`	`str`	Search query (free text or regex)	`''`
`levels`	`Optional[list[str]]`	List of severity levels to filter by (None = all levels)	`None`
`offset`	`int`	Number of matches to skip (for pagination)	`0`
`limit`	`int`	Maximum number of results to return	`50`
`case_sensitive`	`bool`	Whether search should be case-sensitive	`False`
`use_regex`	`bool`	Whether to treat query as regex pattern	`False`
`context_lines`	`int`	Number of lines to include before/after each match	`0`

Returns:

Type	Description
`dict[str, Any]`	Dictionary with: - results: List of matching log entries - total: Total number of matches found - offset: Current offset - limit: Current limit

`log_tailer`¶

`octoprint_logmonitor.log_tailer` ¶

Log Tailer Module

Provides background thread-based log file tailing functionality. Similar to 'tail -f' behavior for live log streaming.

`LogTailer` ¶

Background thread that continuously monitors a log file for new lines.

Features: - Thread-safe start/stop - Handles file rotation - Parses OctoPrint log format - Calls callback for each new line

`init(filepath, callback, poll_interval=0.5, logger=None)` ¶

Initialize the log tailer.

Parameters:

Name	Type	Description	Default
`filepath`	`str`	Path to the log file to tail	required
`callback`	`Callable[[dict[str, Any]], None]`	Function to call for each new log line (receives parsed dict)	required
`poll_interval`	`float`	Polling interval in seconds (default: 0.5)	`0.5`
`logger`	`Optional[Any]`	Optional logger instance for debugging	`None`

`get_last_n_lines(n=100)` ¶

Read the last N lines from the log file.

Parameters:

Name	Type	Description	Default
`n`	`int`	Number of lines to read	`100`

Returns:

Type	Description
`list`	List of parsed log line dictionaries

`is_running()` ¶

Check if the tailer is currently running.

`start()` ¶

Start the log tailer in a background thread.

Returns:

Type	Description
`bool`	True if started successfully, False otherwise

`stop(timeout=5.0)` ¶

Stop the log tailer gracefully.

Parameters:

Name	Type	Description	Default
`timeout`	`float`	Maximum time to wait for thread to stop	`5.0`

Returns:

Type	Description
`bool`	True if stopped successfully, False if timeout

`security`¶

`octoprint_logmonitor.security` ¶

Security Utilities Module

Centralized security helpers for the OctoPrint Log Monitor plugin. Provides path validation, input sanitization, rate limiting, and sensitive data masking to protect against common web security threats.

`RateLimiter` ¶

Simple thread-safe sliding-window rate limiter.

Tracks call timestamps per client key string (e.g. an IP address or user identifier). Calls that exceed the configured quota within the sliding time window are rejected.

Example::

limiter = RateLimiter(max_calls=10, period=60.0)  # 10 req / min

if not limiter.is_allowed(client_ip):
    return flask.jsonify({"error": "Rate limit exceeded"}), 429

`init(max_calls, period)` ¶

Initialise the rate limiter.

Parameters:

Name	Type	Description	Default
`max_calls`	`int`	Maximum number of calls allowed within period.	required
`period`	`float`	Sliding window duration in seconds.	required

`cleanup()` ¶

Remove all fully-expired client entries to free memory.

`is_allowed(client_key)` ¶

Check whether client_key is within the rate limit.

Calling this method counts as one request attempt regardless of the return value.

Parameters:

Name	Type	Description	Default
`client_key`	`str`	An opaque string identifying the caller (e.g. IP).	required

Returns:

Type	Description
`bool`	`True` if the request is within quota, `False` if it should
`bool`	be rejected.

`check_file_size(filepath, max_bytes=MAX_FILE_SIZE_BYTES)` ¶

Return True if filepath exists and is within max_bytes.

Files that cannot be stat-ed (missing, permissions, etc.) return False so the caller can handle the error appropriately.

Parameters:

Name	Type	Description	Default
`filepath`	`str`	Absolute path to the file.	required
`max_bytes`	`int`	Size limit in bytes. Defaults to :data:`MAX_FILE_SIZE_BYTES`.	`MAX_FILE_SIZE_BYTES`

Returns:

Type	Description
`bool`	`True` if the file is within the size limit.

`is_safe_path(base_dir, filename)` ¶

Determine whether filename resolves to a path inside base_dir.

This prevents path-traversal attacks (../, absolute paths, symlink escapes, etc.). Both base_dir and the resolved candidate path are canonicalised with :func:os.path.realpath before comparison so that symbolic links cannot be used to escape the allowed directory.

Parameters:

Name	Type	Description	Default
`base_dir`	`str`	The directory that all file access must be confined to.	required
`filename`	`str`	A bare filename (no directory components expected) provided by the user / API caller.	required

Returns:

Type	Description
`bool`	`True` if and only if the resolved path is strictly inside
`bool`	(or equal to) base_dir.

`mask_sensitive_data(text)` ¶

Apply all sensitive-data masking patterns to text and return the result.

This can be used for optional log-output sanitisation where administrators want to prevent passwords, API keys, or e-mail addresses from appearing in streamed log content sent to the browser.

Parameters:

Name	Type	Description	Default
`text`	`str`	The raw log text to sanitise.	required

Returns:

Type	Description
`str`	The text with sensitive values replaced.

`validate_filename(filename)` ¶

Check that filename is a plain filename with no path components.

Rejects: * Empty strings * Names containing / or \ * Names starting with . (hidden / relative traversal start) * Names that differ from os.path.basename(filename)

Parameters:

Name	Type	Description	Default
`filename`	`str`	The filename string to validate.	required

Returns:

Type	Description
`bool`	`True` if the filename is considered safe.

`validate_pagination(offset, limit, max_limit=MAX_SEARCH_LIMIT)` ¶

Validate offset and limit pagination parameters.

Parameters:

Name	Type	Description	Default
`offset`	`int`	The requested result offset.	required
`limit`	`int`	The requested result count.	required
`max_limit`	`int`	Maximum permitted limit value.	`MAX_SEARCH_LIMIT`

Returns:

Type	Description
`bool`	A `(valid, error_message)` tuple. error_message is empty when
`str`	valid is `True`.

`validate_severity_levels(levels)` ¶

Partition levels into valid and invalid severity level strings.

Parameters:

Name	Type	Description	Default
`levels`	`Optional[list[str]]`	User-supplied list of level strings (may be `None`).	required

Returns:

Type	Description
`tuple[list[str], list[str]]`	`(valid_levels, invalid_levels)` where each is a list of strings.

`plugin core`¶

`octoprint_logmonitor` ¶

OctoPrint Log Monitor Plugin

Provides live log streaming and searching capabilities with severity-based alerting.

`LogmonitorPlugin` ¶

Bases: StartupPlugin, TemplatePlugin, SettingsPlugin, AssetPlugin, BlueprintPlugin

Main plugin class implementing OctoPrint Log Monitor functionality.

Features: - Live log streaming via WebSocket - Full-text log search with severity filtering - Navbar and Sidebar indicators for severity alerts - Configurable severity triggers

`clear_alert_history()` ¶

Clear alert history.

`download_log_file(filename)` ¶

Download a log file directly.

`export_results()` ¶

Export search results to CSV or TXT format.

`frontend_debug_log()` ¶

Write frontend debug events into OctoPrint server logs.

`get_active_streams()` ¶

Get list of active streaming files (multi-file support).

`get_alert_history()` ¶

Get alert history.

`get_alert_monitor_status()` ¶

Return current alert-monitor configuration and active files.

`get_assets()` ¶

Define asset files to include in the UI.

`get_log_files()` ¶

Get list of available log files.

`get_settings_defaults()` ¶

Define default plugin settings.

`get_template_configs()` ¶

Define template configurations for tab, navbar, and sidebar.

`get_template_vars()` ¶

Provide additional template variables for rendering settings UI.

`get_update_information()` ¶

Configure Software Update Plugin integration.

`is_blueprint_csrf_protected()` ¶

Explicitly require CSRF protection for blueprint routes.

`is_template_autoescaped()` ¶

Enable Jinja autoescaping for all plugin templates.

`on_after_startup()` ¶

Initialize plugin after OctoPrint startup.

`on_settings_save(data)` ¶

Persist settings and refresh background alert monitoring.

`on_shutdown()` ¶

Clean shutdown of background threads.

`reset_alerts()` ¶

Reset severity alert counters.

`search_logs()` ¶

Search logs with pagination and severity filtering.

`start_multi_stream()` ¶

Start streaming multiple log files simultaneously (NEW).

`start_stream()` ¶

Start or switch log streaming.

`stop_multi_stream()` ¶

Stop streaming specific log files (NEW).

`stop_stream()` ¶

Stop log streaming.

`write_debug_test_entries()` ¶

Write one test entry per severity category into OctoPrint logs.

`__plugin_load__()` ¶

Load the plugin.

Python API¶

log_searcher¶

octoprint_logmonitor.log_searcher ¶

LogSearcher ¶

__init__(logger=None) ¶

export_to_csv(results) ¶

export_to_txt(results) ¶

get_file_stats(filepath) ¶

search(filepath, query='', levels=None, offset=0, limit=50, case_sensitive=False, use_regex=False, context_lines=0) ¶

log_tailer¶

octoprint_logmonitor.log_tailer ¶

LogTailer ¶

__init__(filepath, callback, poll_interval=0.5, logger=None) ¶

get_last_n_lines(n=100) ¶

is_running() ¶

start() ¶

stop(timeout=5.0) ¶

security¶

octoprint_logmonitor.security ¶

RateLimiter ¶

__init__(max_calls, period) ¶

cleanup() ¶

is_allowed(client_key) ¶

check_file_size(filepath, max_bytes=MAX_FILE_SIZE_BYTES) ¶

is_safe_path(base_dir, filename) ¶

mask_sensitive_data(text) ¶

validate_filename(filename) ¶

validate_pagination(offset, limit, max_limit=MAX_SEARCH_LIMIT) ¶

validate_severity_levels(levels) ¶

plugin core¶

octoprint_logmonitor ¶

LogmonitorPlugin ¶

clear_alert_history() ¶

download_log_file(filename) ¶

export_results() ¶

frontend_debug_log() ¶

get_active_streams() ¶

get_alert_history() ¶

get_alert_monitor_status() ¶

get_assets() ¶

get_log_files() ¶

get_settings_defaults() ¶

get_template_configs() ¶

get_template_vars() ¶

get_update_information() ¶

is_blueprint_csrf_protected() ¶

is_template_autoescaped() ¶

on_after_startup() ¶

on_settings_save(data) ¶

on_shutdown() ¶

reset_alerts() ¶

search_logs() ¶

start_multi_stream() ¶

start_stream() ¶

stop_multi_stream() ¶

stop_stream() ¶

write_debug_test_entries() ¶

__plugin_load__() ¶

`log_searcher`¶

`octoprint_logmonitor.log_searcher` ¶

`LogSearcher` ¶

`init(logger=None)` ¶

`export_to_csv(results)` ¶

`export_to_txt(results)` ¶

`get_file_stats(filepath)` ¶

`search(filepath, query='', levels=None, offset=0, limit=50, case_sensitive=False, use_regex=False, context_lines=0)` ¶

`log_tailer`¶

`octoprint_logmonitor.log_tailer` ¶

`LogTailer` ¶

`init(filepath, callback, poll_interval=0.5, logger=None)` ¶

`get_last_n_lines(n=100)` ¶

`is_running()` ¶

`start()` ¶

`stop(timeout=5.0)` ¶

`security`¶

`octoprint_logmonitor.security` ¶

`RateLimiter` ¶

`init(max_calls, period)` ¶

`cleanup()` ¶

`is_allowed(client_key)` ¶

`check_file_size(filepath, max_bytes=MAX_FILE_SIZE_BYTES)` ¶

`is_safe_path(base_dir, filename)` ¶

`mask_sensitive_data(text)` ¶

`validate_filename(filename)` ¶

`validate_pagination(offset, limit, max_limit=MAX_SEARCH_LIMIT)` ¶

`validate_severity_levels(levels)` ¶

`plugin core`¶

`octoprint_logmonitor` ¶

`LogmonitorPlugin` ¶

`clear_alert_history()` ¶

`download_log_file(filename)` ¶

`export_results()` ¶

`frontend_debug_log()` ¶

`get_active_streams()` ¶

`get_alert_history()` ¶

`get_alert_monitor_status()` ¶

`get_assets()` ¶

`get_log_files()` ¶

`get_settings_defaults()` ¶

`get_template_configs()` ¶

`get_template_vars()` ¶

`get_update_information()` ¶

`is_blueprint_csrf_protected()` ¶

`is_template_autoescaped()` ¶

`on_after_startup()` ¶

`on_settings_save(data)` ¶

`on_shutdown()` ¶

`reset_alerts()` ¶

`search_logs()` ¶

`start_multi_stream()` ¶

`start_stream()` ¶

`stop_multi_stream()` ¶

`stop_stream()` ¶

`write_debug_test_entries()` ¶

`__plugin_load__()` ¶