API reference

This reference is generated from source docstrings via Sphinx autodoc.

mirror

mirror.conf: Config
mirror.packages: Packages
mirror.confPath: Path
mirror.publishPath: Path
mirror.status: dict

mirror.command

mirror.config

mirror.config.CONFIG_PATH: Path
mirror.config.STAT_DATA_PATH: Path
mirror.config.STATUS_PATH: Path
mirror.config.SOCKET_PATH: str
mirror.config.load(conf_path)[source]

Loads the main config file, derives other paths from it, synchronizes with the persistent stat file, and loads the state into the application.

Parameters:

conf_path (Path)

mirror.config.reload()[source]

Reloads all configurations.

This is now a thin wrapper around _perform_reload; see that function for the orchestration details.

Return type:

dict

mirror.config.generate_and_save_web_status()[source]

Generates the web status dictionary from the current package states and saves it to the status.json file.

mirror.config.save_stat_data()[source]

Saves the current package states to the persistent stat file.

mirror.event

mirror.event.on(event_name, listener=None, priority=50)[source]

Register a listener for an event, or return a decorator if listener is omitted.

Parameters:

  • event_name (str) – Event name to listen for.

  • listener (Callable, optional) – Callback to register. If None, returns a decorator.

  • priority (int, optional) – Execution order. Defaults to 50.

mirror.event.once(event_name, listener, priority=50)[source]

Register a one-shot listener via the global manager.

Parameters:

  • event_name (str) – Event name to listen for.

  • listener (Callable) – Callback to invoke once.

  • priority (int, optional) – Execution order. Defaults to 50.

Return type:

None

mirror.event.off(event_name, listener)[source]

Unregister a listener via the global manager.

Parameters:

  • event_name (str) – Event name the listener is registered under.

  • listener (Callable) – Listener to remove.

Return type:

None

mirror.event.post_event(event_name, *args, wait=False, **kwargs)[source]

Fire an event via the global manager.

Parameters:

  • event_name (str) – Name of the event to fire.

  • *args – Positional payload forwarded to listeners.

  • wait (bool, keyword-only) – If True, block until all listeners complete.

  • **kwargs – Keyword payload forwarded to listeners.

Return type:

None

mirror.event.listener(event_name, priority=50)[source]

Decorator to register a function as an event listener.

Parameters:

  • event_name (str) – Event name to listen for.

  • priority (int, optional) – Execution order. Defaults to 50.

class mirror.event.EventManager(max_workers=20)[source]

Bases: object

Central event management system using Pub/Sub pattern. Supports synchronous and asynchronous (threaded) listeners.

Parameters:

max_workers (int)

on(event_name, listener, priority=50)[source]

Register a listener for a specific event.

Lower priority number means higher precedence (executes earlier).

Parameters:

  • event_name (str) – Name of the event to listen for.

  • listener (Callable) – Callback to invoke when the event fires.

  • priority (int, optional) – Execution order (lower = earlier). Defaults to 50.

Return type:

None

once(event_name, listener, priority=50)[source]

Register a one-shot listener that auto-removes itself after first invocation.

Parameters:

  • event_name (str) – Name of the event to listen for.

  • listener (Callable) – Callback to invoke once.

  • priority (int, optional) – Execution order. Defaults to 50.

Return type:

None

off(event_name, listener)[source]

Unregister a previously registered listener.

Parameters:

  • event_name (str) – Event name the listener is registered under.

  • listener (Callable) – Listener to remove.

Return type:

None

post_event(event_name, *args, wait=False, **kwargs)[source]

Fire an event, executing all registered listeners.

Parameters:

  • event_name (str) – Name of the event to fire.

  • *args – Positional payload forwarded to listeners.

  • wait (bool, keyword-only) – If True, block until all listeners complete.

  • **kwargs – Keyword payload forwarded to listeners.

Return type:

None

shutdown(wait=True)[source]

Shut down the event manager and its thread pool.

Parameters:

wait (bool, optional) – If True, block until all running listeners complete. Defaults to True.

Return type:

None

mirror.logger

class mirror.logger.PromptHandler(stream=None)[source]

Bases: StreamHandler

Log handler that prints via prompt_toolkit ANSI when the terminal supports it, and falls back to plain text otherwise.

emit(record)[source]

Emit a record.

If a formatter is specified, it is used to format the record. The record is then written to the stream with a trailing newline. If exception information is present, it is formatted using traceback.print_exception and appended to the stream. If the stream has an ‘encoding’ attribute, it is used to determine how to do the output to the stream.

Parameters:

record (LogRecord)

Return type:

None

class mirror.logger.DynamicGzipRotatingFileHandler(base_path, folder_template, filename_template, gzip_enabled=True, encoding='utf-8', delay=False)[source]

Bases: FileHandler

FileHandler that rotates when the formatted path changes. Supports dynamic folders and filenames based on time templates.

Parameters:

  • base_path (str | Path)

  • folder_template (str)

  • filename_template (str)

  • gzip_enabled (bool)

  • encoding (str | None)

  • delay (bool)

emit(record)[source]

Check if path needs rotation before emitting.

do_rotation(new_path)[source]

Close current file, optionally compress it, and open the new one.

Parameters:

new_path (str)

mirror.logger.input(message=None, *, editing_mode=None, refresh_interval=None, vi_mode=None, lexer=None, completer=None, complete_in_thread=None, is_password=None, key_bindings=None, bottom_toolbar=None, style=None, color_depth=None, cursor=None, include_default_pygments_style=None, style_transformation=None, swap_light_and_dark_colors=None, rprompt=None, multiline=None, prompt_continuation=None, wrap_lines=None, enable_history_search=None, search_ignore_case=None, complete_while_typing=None, validate_while_typing=None, complete_style=None, auto_suggest=None, validator=None, clipboard=None, mouse_support=None, input_processors=None, placeholder=None, reserve_space_for_menu=None, enable_system_prompt=None, enable_suspend=None, enable_open_in_editor=None, tempfile_suffix=None, tempfile=None, show_frame=None, default='', accept_default=False, pre_run=None, set_exception_handler=True, handle_sigint=True, in_thread=False, inputhook=None)

Display the prompt.

The first set of arguments is a subset of the PromptSession class itself. For these, passing in None will keep the current values that are active in the session. Passing in a value will set the attribute for the session, which means that it applies to the current, but also to the next prompts.

Note that in order to erase a Completer, Validator or AutoSuggest, you can’t use None. Instead pass in a DummyCompleter, DummyValidator or DummyAutoSuggest instance respectively. For a Lexer you can pass in an empty SimpleLexer.

Additional arguments, specific for this prompt:

Parameters:

  • default (str | Document) – The default input text to be shown. (This can be edited by the user).

  • accept_default (bool) – When True, automatically accept the default value without allowing the user to edit the input.

  • pre_run (Callable[[], None] | None) – Callable, called at the start of Application.run.

  • in_thread (bool) – Run the prompt in a background thread; block the current thread. This avoids interference with an event loop in the current thread. Like Application.run(in_thread=True).

  • message (AnyFormattedText | None)

  • editing_mode (EditingMode | None)

  • refresh_interval (float | None)

  • vi_mode (bool | None)

  • lexer (Lexer | None)

  • completer (Completer | None)

  • complete_in_thread (bool | None)

  • is_password (bool | None)

  • key_bindings (KeyBindingsBase | None)

  • bottom_toolbar (AnyFormattedText | None)

  • style (BaseStyle | None)

  • color_depth (ColorDepth | None)

  • cursor (AnyCursorShapeConfig | None)

  • include_default_pygments_style (FilterOrBool | None)

  • style_transformation (StyleTransformation | None)

  • swap_light_and_dark_colors (FilterOrBool | None)

  • rprompt (AnyFormattedText | None)

  • multiline (FilterOrBool | None)

  • prompt_continuation (PromptContinuationText | None)

  • wrap_lines (FilterOrBool | None)

  • enable_history_search (FilterOrBool | None)

  • search_ignore_case (FilterOrBool | None)

  • complete_while_typing (FilterOrBool | None)

  • validate_while_typing (FilterOrBool | None)

  • complete_style (CompleteStyle | None)

  • auto_suggest (AutoSuggest | None)

  • validator (Validator | None)

  • clipboard (Clipboard | None)

  • mouse_support (FilterOrBool | None)

  • input_processors (list[Processor] | None)

  • placeholder (AnyFormattedText | None)

  • reserve_space_for_menu (int | None)

  • enable_system_prompt (FilterOrBool | None)

  • enable_suspend (FilterOrBool | None)

  • enable_open_in_editor (FilterOrBool | None)

  • tempfile_suffix (str | Callable[[], str] | None)

  • tempfile (str | Callable[[], str] | None)

  • show_frame (FilterOrBool | None)

  • set_exception_handler (bool)

  • handle_sigint (bool)

  • inputhook (InputHook | None)

Return type:

_T

This method will raise KeyboardInterrupt when control-c has been pressed (for abort) and EOFError when control-d has been pressed (for exit).

mirror.logger.compress_file(filepath)[source]

Compress a file with gzip and remove the original.

Parameters:

filepath (str | Path) – Path to the file to compress.

Returns:

Path to the .gz file, or None if compression failed.

Return type:

gz_path(Path | None)

mirror.logger.create_logger(name, start_time)[source]

Create a per-package logger for a sync session.

Parameters:

  • name (str) – Package name used to identify the logger and format paths.

  • start_time (float) – Unix timestamp of when the sync started.

Returns:

Configured logger with file and prompt handlers.

Return type:

pkg_logger(logging.Logger)

mirror.logger.close_logger(pkg_logger, compress=None)[source]

Close a package logger and optionally compress the log file.

Parameters:

  • pkg_logger (logging.Logger) – The logger to close.

  • compress (bool, optional) – Override gzip setting. Uses config value if None.

Returns:

Path to the (compressed) log file, or None if no file handler.

Return type:

log_path(Path | None)

mirror.logger.setup_logger()[source]

Configure the main application logger with file and console handlers.

Return type:

None

mirror.logger.get_log_path(pkg_logger)[source]

Return the file path used by the logger’s FileHandler, or None.

Parameters:

pkg_logger (logging.Logger) – Package logger to inspect.

Returns:

Log file path, or None if no FileHandler is attached.

Return type:

path(Path | None)

mirror.logger.get(pkgid)[source]

Return the logger for the given package ID.

Parameters:

pkgid (str) – Package identifier.

Returns:

Logger scoped to this package.

Return type:

logger(logging.Logger)

mirror.logger.exists(pkgid)[source]

Return True if the package logger has at least one FileHandler attached.

Parameters:

pkgid (str) – Package identifier.

Returns:

True if a FileHandler is present.

Return type:

attached(bool)

mirror.logger.reattach_logger(pkg_logger, log_file_path, pkgid)[source]

Reattach a FileHandler to an existing in-base log file.

Used by on_sync_done after master restart. Performs strict validation: - path resolves inside the configured package log base - opens with O_NOFOLLOW (refuses symlinks atomically) - fstat on the SAME fd confirms regular file with st_nlink == 1 - that same fd is then adopted by SafeAppendFileHandler — no reopen Returns True if a handler was attached, False otherwise.

Parameters:

  • pkg_logger (logging.Logger) – Logger to attach a FileHandler to.

  • log_file_path (Path) – Path to the log file (from stat.json runninglog).

  • pkgid (str) – Package identifier (used in warning messages).

Returns:

True if a handler was successfully attached.

Return type:

attached(bool)

mirror.plugin

Entry-points based plug-in framework for mirror.py.

Loading splits into two phases:
Phase A (load_builtin_plugins): imports and registers the five built-in sync

modules at package-import time so mirror.sync.methods is populated before package validation runs.

Phase B (load_external_plugins): called from mirror.config.load() after the

config dict has been parsed; disables config-disabled built-ins and discovers + registers third-party plug-ins via importlib.metadata.

Per-plugin configuration is read from a JSON file in the same directory as the main config.json. The default filename is <plugin-name>.json. Operators can override the filename per-plugin by setting config_filename on the PluginRecord. The config file is read lazily by get_config() and is never cached, so changes take effect on the next call.

The plugins block in config.json uses an enable-only shape:

{
    "<name>": {"enabled": true}
}

The config sub-key previously accepted in that block is no longer supported; move any per-plugin settings to <config_dir>/<name>.json.

Versioning

PLUGIN_API_VERSION is a (major, minor) tuple that describes the plug-in contract implemented by this core release.

  • major increments on breaking changes (calling convention, factory surface, entry-point groups, plug-in types). A plug-in whose declared major differs from the core major is skipped at load time.

  • minor increments on additive, backward-compatible changes (new optional hook, new optional field). A plug-in whose declared minor is greater than the core minor is also skipped (it was built against features the core does not yet provide). An older plug-in (declared minor <= core minor) continues to load.

Plug-ins declare their target version via the api_version parameter of the factory helpers (sync_plugin, event_plugin, status_plugin). The gate is applied only to external plug-ins loaded in Phase B; built-ins are loaded ungated in Phase A because they ship in lockstep with the core. A plug-in that omits api_version (None) still loads but emits a deprecation warning.

class mirror.plugin.StatusOutput(name, default_path, build, config_path_key=None)[source]

Bases: object

Declarative description of an additional status output file written by a status plug-in.

Parameters:

  • name (str) – Globally unique output name (across all plug-ins).

  • default_path (str) – Filesystem path to write the output to. Operator can override via plug-in config if config_path_key is set.

  • build (Callable) – Callable producing the JSON payload from an iterable of packages.

  • config_path_key (Optional[str]) – If set, the plug-in’s config dict key whose value (when present) overrides default_path.

name: str
default_path: str
build: Callable
config_path_key: str | None = None
class mirror.plugin.ConfigCreateResult(path, created)[source]

Bases: object

Outcome of a plug-in’s create_config() call.

Parameters:

  • path (str) – Filesystem path of the plug-in’s config file.

  • created (bool) – True if the file was written; False if the file already existed and was left untouched (skipped because –force was not given).

path: str
created: bool
class mirror.plugin.PluginRecord(name, type, execute=None, on_sync_done=None, setup=None, extend_stat_fields=None, extend_web_status_fields=None, transform_stat_payload=None, transform_web_status_payload=None, outputs=None, create_config=None, config_filename=None, api_version=None)[source]

Bases: object

Typed descriptor for a registered plug-in.

Parameters:

  • name (str) – Globally unique plug-in name.

  • type (str) – One of “sync”, “event”, or “status”.

  • execute (Callable, optional) – Sync execute callable. Sync plug-ins only.

  • on_sync_done (Callable, optional) – Post-sync hook. Sync plug-ins only.

  • setup (Callable, optional) – Setup callable (required for event plug-ins).

  • extend_stat_fields (Callable, optional) – Returns extra stat.json fields for a package.

  • extend_web_status_fields (Callable, optional) – Returns extra web status fields for a package.

  • transform_stat_payload (Callable, optional) – Transforms the full stat.json payload dict. Single-owner: only one plug-in may register this per daemon instance.

  • transform_web_status_payload (Callable, optional) – Transforms the full web status payload dict. Single-owner: only one plug-in may register this per daemon instance.

  • outputs (list, optional) – List of StatusOutput instances describing additional output files this plug-in writes on every status update.

  • create_config (Callable, optional) – On-demand config-file creation callable. Signature create_config(force: bool) -> ConfigCreateResult. The plug-in writes its own config file at a path/name it owns, reading any global settings from mirror.conf and its own per-plug-in config via mirror.plugin.get_config(<name>); when the file already exists and force is False it skips and returns created=False.

  • config_filename (str, optional) – Optional override for the per-plugin config filename. Defaults to <name>.json when absent. The file is resolved relative to the directory that contains the main config.json.

  • api_version (tuple, optional) – (major, minor) API version this plug-in was built against. Compared against PLUGIN_API_VERSION at load time for external plug-ins. None means undeclared (loads with a deprecation warning).

name: str
type: Literal['sync', 'event', 'status']
execute: Callable | None = None
on_sync_done: Callable | None = None
setup: Callable | None = None
extend_stat_fields: Callable | None = None
extend_web_status_fields: Callable | None = None
transform_stat_payload: Callable | None = None
transform_web_status_payload: Callable | None = None
outputs: list | None = None
create_config: Callable | None = None
config_filename: str | None = None
api_version: tuple[int, int] | None = None
mirror.plugin.sync_plugin(name, execute, on_sync_done=None, setup=None, create_config=None, config_filename=None, api_version=None)[source]

Build a PluginRecord for a sync plug-in with contract validation.

Parameters:

  • name (str) – Unique plug-in name.

  • execute (Callable) – Sync execute callable — must be provided and callable.

  • on_sync_done (Callable, optional) – Post-sync hook callable.

  • setup (Callable, optional) – Optional setup callable.

  • create_config (Callable, optional) – On-demand config-file creation callable.

  • config_filename (str, optional) – Override for the per-plugin config filename. Defaults to <name>.json when absent.

  • api_version (tuple, optional) – (major, minor) API version this plug-in targets. Validated and stored on the returned PluginRecord.

Returns:

Validated sync PluginRecord.

Return type:

record(PluginRecord)

Raises:

  • TypeError – If execute is missing or not callable, or api_version has wrong type/shape.

  • ValueError – If api_version has out-of-range major or minor.

mirror.plugin.event_plugin(name, setup, create_config=None, config_filename=None, api_version=None)[source]

Build a PluginRecord for an event plug-in with contract validation.

Parameters:

  • name (str) – Unique plug-in name.

  • setup (Callable) – Required setup callable that registers event listeners.

  • create_config (Callable, optional) – On-demand config-file creation callable.

  • config_filename (str, optional) – Override for the per-plugin config filename. Defaults to <name>.json when absent.

  • api_version (tuple, optional) – (major, minor) API version this plug-in targets. Validated and stored on the returned PluginRecord.

Returns:

Validated event PluginRecord.

Return type:

record(PluginRecord)

Raises:

  • TypeError – If setup is missing or not callable, or api_version has wrong type/shape.

  • ValueError – If api_version has out-of-range major or minor.

mirror.plugin.status_plugin(name, extend_stat_fields=None, extend_web_status_fields=None, transform_stat_payload=None, transform_web_status_payload=None, outputs=None, setup=None, create_config=None, config_filename=None, api_version=None)[source]

Build a PluginRecord for a status plug-in with contract validation.

Parameters:

  • name (str) – Unique plug-in name.

  • extend_stat_fields (Callable, optional) – Returns extra stat.json fields for a package.

  • extend_web_status_fields (Callable, optional) – Returns extra web status fields for a package.

  • transform_stat_payload (Callable, optional) – Transforms the full stat.json payload dict.

  • transform_web_status_payload (Callable, optional) – Transforms the full web status payload dict.

  • outputs (list, optional) – List of StatusOutput instances for additional output files.

  • setup (Callable, optional) – Optional setup callable.

  • create_config (Callable, optional) – On-demand config-file creation callable.

  • config_filename (str, optional) – Override for the per-plugin config filename. Defaults to <name>.json when absent.

  • api_version (tuple, optional) – (major, minor) API version this plug-in targets. Validated and stored on the returned PluginRecord.

Returns:

Validated status PluginRecord.

Return type:

record(PluginRecord)

Raises:

  • TypeError – If none of extend_*, transform_*, or outputs is provided, or if any callable argument is not actually callable, or if outputs items are not StatusOutput instances, or if api_version has wrong type/shape.

  • ValueError – If api_version has out-of-range major or minor.

mirror.plugin.load_builtin_plugins()[source]

Phase A: import and register all five built-in sync plug-ins.

Hard-codes the five canonical sync module references so that mirror.sync.methods is fully populated before package validation runs. ImportError for any individual module is logged as a warning and skipped; a successful import that yields a malformed PluginRecord raises immediately (that is a programmer error, not a deployment error).

Return type:

None

mirror.plugin.load_external_plugins(plugin_settings)[source]

Phase B: disable built-ins per config and load third-party plug-ins.

Must be called after mirror.conf is populated (i.e., from mirror.config.load()) and before mirror.packages is constructed.

Parameters:

plugin_settings (dict) – Mapping of plug-in name to PluginSettings (or equivalent with an .enabled bool). Each entry uses the shape {"<name>": {"enabled": true}}. If this is a plain list (legacy format) a deprecation warning is logged and the function returns without doing anything.

Return type:

None

mirror.plugin.get_record(name)[source]

Return the registered PluginRecord for the given plug-in name, or None.

Parameters:

name (str) – Plug-in name to look up.

Returns:

The registered record, or None if absent.

Return type:

record(PluginRecord | None)

mirror.plugin.get_config(name)[source]

Return the per-plug-in config dict for a registered plug-in.

Config is read from a JSON file in the same directory as the main config.json. The filename defaults to <name>.json and can be overridden per-plugin via PluginRecord.config_filename. The file is read on every call (no caching).

Parameters:

name (str) – Registered plug-in name.

Returns:

The parsed JSON object from the plug-in config file,

or an empty dict if the file is absent, unreadable, or not a JSON object.

Return type:

config(dict)

Raises:

KeyError – If name is not in the registry (plug-in not loaded).

mirror.socket

Mirror.py Socket Communication Module

Provides Unix socket IPC between master daemon and worker processes with automatic handshake protocol for version and role exchange.

class mirror.socket.BaseServer(socket_path, role, socket_uid=None, socket_gid=None, socket_mode=None)[source]

Bases: object

Base server for Unix socket IPC with handshake and command dispatch

Parameters:

  • socket_path (Path | str) – Path to the Unix domain socket file

  • role (str) – Server role identifier for handshake

  • socket_uid (int, optional) – UID for chown on the socket file

  • socket_gid (int, optional) – GID for chown on the socket file

  • socket_mode (int, optional) – Permission mode for the socket file

set_version(version)[source]

Set application version for handshake

Parameters:

version (str) – Version string

Return type:

None

register_handler(command, handler)[source]

Register a command handler

Parameters:

  • command (str) – Command name

  • handler (Callable) – Handler function

Return type:

None

broadcast(data)[source]

Send a message to all connected clients

Parameters:

data (dict) – Payload to broadcast

Return type:

None

property client_count: int

Number of currently connected clients

start()[source]

Bind the socket and begin accepting connections

Return type:

None

stop()[source]

Close the server socket and remove the socket file

Return type:

None

class mirror.socket.BaseClient(socket_path, role)[source]

Bases: object

Base client for Unix socket IPC with handshake and async listener

Parameters:

  • socket_path (Path | str) – Path to the server’s Unix domain socket

  • role (str) – Client role identifier for handshake

set_version(version)[source]

Set application version for handshake

Parameters:

version (str) – Version string

Return type:

None

handle_notification(data)[source]

Handle server notification. Override in subclasses.

Parameters:

data (dict) – Notification payload

Return type:

None

connect()[source]

Connect to server and perform handshake

Returns:

Server’s handshake information

Return type:

server_info(HandshakeInfo)

disconnect()[source]

Disconnect from server

Return type:

None

send_command(command, recv_timeout=None, **kwargs)[source]

Send a command to server and wait for the response.

Parameters:

  • command (str) – Command name.

  • recv_timeout (float, optional) – How many seconds to wait for the server’s response before raising TimeoutError. Defaults to 30s. NOTE: this parameter is consumed client-side and is NOT forwarded to the server. Any server-side kwarg named recv_timeout would be silently shadowed — callers must rename such kwargs before calling this method.

  • **kwargs – Command arguments forwarded to the server.

Returns:

Response payload from the server.

Return type:

data(Any)

property server_info: HandshakeInfo | None

Server’s handshake info (available after connect)

property is_connected: bool

Whether the client is currently connected

class mirror.socket.HandshakeInfo(app_name, app_version, protocol_version, is_server, role)[source]

Bases: object

Information exchanged during connection handshake

Parameters:

  • app_name (str) – Application identifier

  • app_version (str) – Application version string

  • protocol_version (int) – Wire protocol version

  • is_server (bool) – Whether sender is a server

  • role (str) – Role identifier (master, worker, cli, etc.)

app_name: str
app_version: str
protocol_version: int
is_server: bool
role: str
to_dict()[source]

Serialize to dictionary

Returns:

Dataclass fields as dictionary

Return type:

data(dict)

static from_dict(data)[source]

Deserialize from dictionary

Parameters:

data (dict) – Dictionary with HandshakeInfo fields

Returns:

Deserialized instance

Return type:

info(HandshakeInfo)

mirror.socket.expose(cmd_name=None)[source]

Mark a method as an exposed socket command handler

Parameters:

cmd_name (str, optional) – Command name. Defaults to method name.

Returns:

Decorator that tags the method

Return type:

decorator(Callable)

mirror.socket.init(role, **kwargs)[source]

Initialize a socket server or client by role

Parameters:

  • role (str) – “master” or “worker” for servers, “client”/”master_client” for MasterClient, “worker_client” for WorkerClient

  • **kwargs – Arguments passed to the constructor

Returns:

Initialized server or connected client

Return type:

instance(Any)

mirror.socket.stop()[source]

Stop all running servers and disconnect clients

Return type:

None

mirror.structure

class mirror.structure.Options[source]

Bases: object

get(key, default=None)[source]

Return attribute value by name, or default if not present.

Parameters:

key (str)

to_dict()[source]

Serialize dataclass fields to a dictionary.

Return type:

dict

to_json()[source]

Serialize dataclass fields to a JSON string.

Return type:

str

class mirror.structure.PackageSettings(hidden: bool, src: str, dst: str, options: dict = <factory>)[source]

Bases: Options

Parameters:
hidden: bool
src: str
dst: str
options: dict
classmethod from_dict(data)[source]

Build PackageSettings from a config dict, ignoring unknown keys.

Parameters:

data (dict) – Raw package settings dictionary.

Returns:

Populated instance.

Return type:

settings(PackageSettings)

class mirror.structure.Package(pkgid: str, name: str, status: str, href: str, synctype: str, syncrate: int, link: list[mirror.structure.Package.Link], settings: mirror.structure.PackageSettings, lastsync: float = 0.0, disabled: bool = False, timestamp: float = 0.0, statusinfo: mirror.structure.Package.StatusInfo = <factory>)[source]

Bases: object

Parameters:

Bases: Options

Parameters:
rel: str
href: str
class StatusInfo(lasterrorlog: str | None = None, lastsuccesslog: str | None = None, runninglog: str | None = None, errorcount: int = 0, lastsuccesstime: float = 0.0, lasterrortime: float = 0.0)[source]

Bases: Options

Parameters:

  • lasterrorlog (str | None)

  • lastsuccesslog (str | None)

  • runninglog (str | None)

  • errorcount (int)

  • lastsuccesstime (float)

  • lasterrortime (float)

lasterrorlog: str | None = None
lastsuccesslog: str | None = None
runninglog: str | None = None
errorcount: int = 0
lastsuccesstime: float = 0.0
lasterrortime: float = 0.0
classmethod from_dict(data)[source]
Parameters:

data (dict)

Return type:

StatusInfo

pkgid: str
name: str
status: str
href: str
synctype: str
syncrate: int
settings: PackageSettings
lastsync: float = 0.0
disabled: bool = False
timestamp: float = 0.0
statusinfo: StatusInfo
static from_dict(config)[source]
Parameters:

config (dict)

Return type:

Package

set_status(status, logfile=None)[source]
Parameters:

  • status (Literal['ACTIVE', 'SYNC', 'ERROR', 'UNKNOWN'])

  • logfile (Path | None)

Return type:

None

to_dict()[source]

Serialize the package to a stat-format dictionary.

Returns:

Package fields with “id” key and ISO 8601 syncrate.

Return type:

data(dict)

to_json()[source]

Serialize the package to a JSON string.

Return type:

str

is_syncing()[source]

Return True if the package status is SYNC.

Return type:

bool

is_disabled()[source]

Return True if the package is disabled.

Return type:

bool

class mirror.structure.Packages(pkgs)[source]

Bases: Options

Parameters:

pkgs (dict)

get(key)[source]

Return attribute value by name, or default if not present.

Parameters:

key (str)

Return type:

Package | None

items()[source]
Return type:

dict[str, Package]

keys()[source]
Return type:

list[str]

values()[source]
Return type:

list[Package]

to_dict()[source]

Serialize dataclass fields to a dictionary.

Return type:

dict

class mirror.structure.PluginSettings(enabled: bool = True)[source]

Bases: Options

Parameters:

enabled (bool)

enabled: bool = True
classmethod from_dict(data)[source]

Build PluginSettings from a config dict, ignoring unknown keys.

Parameters:

data (dict) – Raw plugin settings dictionary.

Returns:

Populated instance.

Return type:

settings(PluginSettings)

class mirror.structure.Config(name: str, hostname: str, lastsettingmodified: int, errorcontinuetime: int, logfolder: pathlib._local.Path, webroot: pathlib._local.Path, statusfile: pathlib._local.Path, ftpsync: mirror.structure.Config.FTPSync, uid: int, gid: int, maintainer: dict, localtimezone: str, logger: dict, max_runtime_seconds: int = 0, plugins: dict[str, mirror.structure.PluginSettings]=<factory>, socket: 'Config.SocketSettings' = <factory>)[source]

Bases: object

Parameters:
class FTPSync(maintainer: str = '', sponsor: str = '', country: str = '', location: str = '', throughput: str = '', include: str = '', exclude: str = '')[source]

Bases: Options

Parameters:

  • maintainer (str)

  • sponsor (str)

  • country (str)

  • location (str)

  • throughput (str)

  • include (str)

  • exclude (str)

maintainer: str = ''
sponsor: str = ''
country: str = ''
location: str = ''
throughput: str = ''
include: str = ''
exclude: str = ''
class SocketSettings(uid: int | None = None, gid: int | None = None, mode: int = 384)[source]

Bases: Options

Parameters:

  • uid (int | None)

  • gid (int | None)

  • mode (int)

uid: int | None = None
gid: int | None = None
mode: int = 384
classmethod from_dict(data)[source]

Build SocketSettings from the settings.socket config dict.

Parameters:

data (dict) – Raw socket settings ({“uid”, “gid”, “mode”}).

Returns:

Parsed instance. uid/gid stay None

when absent (no chown). mode defaults to 0o600 and is parsed from an octal string (e.g. “0770”) via parse_file_mode.

Return type:

settings(SocketSettings)

to_config_dict()[source]

Serialize back to the settings.socket config shape.

mode is emitted as an octal string; uid/gid are omitted when None so the result round-trips through from_dict.

Returns:

{“mode”: <octal string>} plus uid/gid when set.

Return type:

data(dict)

name: str
hostname: str
lastsettingmodified: int
errorcontinuetime: int
logfolder: Path
webroot: Path
statusfile: Path
ftpsync: FTPSync
uid: int
gid: int
maintainer: dict
localtimezone: str
logger: dict
max_runtime_seconds: int = 0
plugins: dict[str, PluginSettings]
socket: SocketSettings
static load_from_dict(config)[source]

Build a Config instance from the parsed JSON config dict.

Parameters:

config (dict) – Top-level config dictionary.

Returns:

Populated Config instance.

Return type:

conf(Config)

to_dict()[source]

Serialize Config to a dictionary matching the config.json schema.

Returns:

Config as a serializable dict.

Return type:

data(dict)

to_json()[source]

Serialize Config to a JSON string.

Return type:

str

mirror.sync

mirror.sync.set_standalone_mode(enabled)[source]

Enable or disable standalone execution mode.

In standalone mode, on_sync_done records the result but does NOT write stat.json, status.json, or mutate package status. execute_command runs the subprocess in the foreground without a worker socket.

Enabling standalone mode clears any previously recorded results so a reused pkgid cannot read a stale result from an earlier in-process run (e.g. a run that raises before on_sync_done would otherwise see the prior success).

Parameters:

enabled (bool) – True to activate standalone mode, False to deactivate.

Return type:

None

mirror.sync.get_standalone_result(pkgid)[source]

Return the recorded standalone sync result for a package, if any.

Parameters:

pkgid (str) – Package identifier.

Returns:

(success, returncode) tuple, or None if no

result has been recorded for this pkgid.

Return type:

result(tuple, optional)

mirror.sync.get_module(method)[source]

Return the loaded sync module for the given method name.

Parameters:

method (str) – Sync method name (e.g. “rsync”, “ftpsync”).

Returns:

The loaded sync module object.

Return type:

module(Callable)

mirror.sync.mark_watchdog_fired(pkgid)[source]

Atomically claim the watchdog kill for this pkgid.

Parameters:

pkgid (str) – Package identifier.

Returns:

True if this is the first time the watchdog fired for

pkgid since its last sync start; False if already fired.

Return type:

first(bool)

mirror.sync.release_watchdog_fired(pkgid)[source]

Release a previously-claimed watchdog marker (e.g., on stop_command failure).

Parameters:

pkgid (str) – Package identifier.

Return type:

None

mirror.sync.should_kill_for_max_runtime(uptime, max_runtime_seconds)[source]

Decide whether the watchdog should kill a sync for exceeding max_runtime.

Parameters:

  • uptime (float, optional) – Seconds since the sync started, as reported by the worker, or None if the worker did not return uptime info.

  • max_runtime_seconds (int) – The package’s configured cap; 0 disables the watchdog.

Returns:

True if uptime is known and exceeds the cap.

Return type:

kill(bool)

mirror.sync.start(package, trigger='auto', extra_args=None)[source]

Start sync for a package.

Rejects if a sync for the same pkgid is already in progress.

Parameters:

  • package (mirror.structure.Package) – Package to sync.

  • trigger (str) – Source of the trigger (“auto”, “manual”, etc.).

  • extra_args (dict[str, str], optional) – Extra key-value pairs to associate with this sync (str->str). Validated before the lock is acquired. Cleared from the registry on completion or on scoped launch failure. Raises ValueError on bad input (non-str keys/values, empty key, ‘=’ or NUL in key, NUL in value). Raises RuntimeError if a sync for this pkgid is already in progress (existing behavior unchanged).

Raises:

  • ValueError – If sync method is unknown or extra_args is invalid.

  • RuntimeError – If a sync for this pkgid is already in progress.

Return type:

None

mirror.sync.get_extra_args(pkgid)[source]

Return a shallow copy of the extra_args registered for an in-flight sync, or empty.

Parameters:

pkgid (str) – Package ID.

Returns:

Copy of the stored mapping (empty if none).

Return type:

extra_args(dict[str, str])

mirror.sync.on_sync_done(pkgid, success, returncode)[source]

Handle sync completion: log result, call per-module hook, update package status.

Parameters:

  • pkgid (str) – Package identifier.

  • success (bool) – Whether the sync succeeded.

  • returncode (int, optional) – Process return code, or None if unavailable.

Return type:

None

mirror.sync.execute(package, logger, trigger='auto')[source]

Module-level execute placeholder; sync modules override this.

Parameters:
Return type:

None

mirror.toolbox

mirror.toolbox.parse_iso_duration(iso8601)[source]

Parse an ISO 8601 duration string into total seconds.

Only supports days, hours, minutes, and seconds.

Parameters:

iso8601 (str) – ISO 8601 duration string (e.g. “P1DT2H3M4S”) or “PUSH” or “”.

Returns:

Total duration in seconds. Returns -1 for “PUSH”, 0 for “”.

Return type:

seconds(int)

mirror.toolbox.format_iso_duration(duration)[source]

Format total seconds into an ISO 8601 duration string.

Only supports days, hours, minutes, and seconds.

Parameters:

duration (int) – Duration in seconds. Use -1 for “PUSH”.

Returns:

ISO 8601 duration string, “PUSH” for -1, or “” for 0.

Return type:

iso8601(str)

mirror.toolbox.parse_file_mode(value)[source]

Parse an octal file-mode string into an integer.

Accepts forms like “0770”, “0o770”, or “770”; all interpreted as base-8.

Parameters:

value (str) – Octal mode string from configuration.

Returns:

Parsed file mode as an integer (e.g. 0o770).

Return type:

mode(int)

mirror.toolbox.set_rsync_user(url, user)[source]

Embed a username into an rsync URL.

Parameters:

  • url (str) – Rsync source URL (rsync:// or :: form).

  • user (str) – Username to embed.

Returns:

URL with the username inserted.

Return type:

url_with_user(str)

mirror.toolbox.has_root_or_sudo()[source]

Check that user has root or passwordless sudo permission.

Returns:

True if EUID is 0 or sudo -n true succeeds.

Return type:

ok(bool)

mirror.toolbox.command_exists(command)[source]

Check whether the given command is available on PATH.

Parameters:

command (str) – Command name to look up.

Returns:

True if the command is found on PATH.

Return type:

exists(bool)

mirror.worker

class mirror.worker.Job(job_id, commandline, env, uid, gid, nice, log_path=None, log_helper_command=None)[source]

Bases: object

Represents a worker process.

Parameters:
start()[source]

Spawn the subprocess with the configured command, uid, gid, and niceness.

Return type:

None

get_pipe(stream)[source]

Return the file descriptor for the specified stream.

Parameters:

stream (str) – One of ‘stdin’, ‘stdout’, ‘stderr’

Returns:

File descriptor, or None if unavailable

Return type:

fd(int | None)

property pid: int | None
property is_running: bool
property returncode: int | None
stop(timeout=5)[source]

Terminate the worker process, killing it if it does not stop in time.

Parameters:

timeout (int, optional) – Seconds to wait before sending SIGKILL. Defaults to 5.

Return type:

None

reap()[source]

Advance helper lifecycle after the main process exits.

Return type:

None

info()[source]

Return a snapshot dict of the job’s current state.

Returns:

Job metadata including id, pid, running status, and uptime.

Return type:

data(dict)

mirror.worker.create(job_id, commandline, env, uid, gid, nice, log_path=None, log_helper_command=None)[source]

Create and start a new worker.

Parameters:

  • job_id (str) – Unique identifier for the job

  • commandline (list[str]) – Command to execute

  • env (dict[str, str]) – Extra environment variables

  • uid (int | None) – User ID for the subprocess

  • gid (int | None) – Group ID for the subprocess

  • nice (int) – Niceness value

  • log_path (Path, optional) – File to redirect stdout/stderr into

  • log_helper_command (list[str] | None)

Returns:

The started Job instance

Return type:

job(Job)

Raises:

ValueError – If a job with the given ID already exists

mirror.worker.get(job_id)[source]

Retrieve a worker by ID.

Parameters:

job_id (str) – Job identifier

Returns:

The Job, or None if not found

Return type:

job(Job | None)

mirror.worker.get_all()[source]

Return a snapshot list of all registered jobs.

Returns:

All current jobs

Return type:

jobs(list[Job])

mirror.worker.prune_finished()[source]

Remove finished jobs from the registry after notifying clients.

Notification is attempted via mirror.socket.worker.send_finished_notification. If notification fails, the attempt counter is incremented. After NOTIFY_ATTEMPT_BUDGET consecutive failures the job is force-pruned.

mirror.worker.manage(interval=1)[source]

Run the worker manager loop, pruning finished jobs at each interval.

Parameters:

interval (int, optional) – Sleep duration in seconds between prune cycles. Defaults to 1.

Return type:

None