Skip to content

API Reference

Auto-generated from docstrings.

deep_merge(d1, d2, *, deepcopy_first=True)

Recursively merge two mappings and return the merged result.

This function returns a new dict containing the keys from d1 updated by the keys from d2. When a key exists in both mappings and both values are mapping types (collections.abc.Mapping), the merge is performed recursively so nested mappings are merged instead of replaced.

The merge is implemented iteratively using a stack to avoid recursion limits. By default a deep copy of d1 is made first to guarantee that the returned mapping shares no nested mutable structures with the original d1. To optimize for performance and avoid copying, set deepcopy_first=False which will start from a shallow copy of d1 instead (note: nested mutable objects from d1 may be shared in that case).

Parameters:

Name Type Description Default
d1 dict[str, Any]

The base mapping whose values will be updated.

required
d2 dict[str, Any]

The mapping to merge into d1. Values in d2 take precedence over values in d1 when keys collide.

required
deepcopy_first bool

If True (default) produce a deep copy of d1 before merging. If False, use a shallow copy of d1 which is faster but may share nested mutable objects with the original.

True

Returns:

Type Description
dict[str, Any]

A new dict representing the merged mapping.

Examples:

>>> deep_merge({"a": 1, "b": {"x": 1}}, {"b": {"y": 2}})
{'a': 1, 'b': {'x': 1, 'y': 2}}
Notes

Non-mapping values are replaced by values from d2. Sequences and other non-mapping iterables are not merged element-wise.

b64decode(data)

Decode a URL-safe Base64-encoded string into a UTF-8 string.

The helper will add the required padding characters (=) if they are missing from the input before decoding.

Parameters:

Name Type Description Default
data str

A base64 (URL-safe) encoded string. Padding may be omitted.

required

Returns:

Type Description
str

The decoded Unicode string.

Raises:

Type Description
Error

If the input is not valid base64.

detect_encoding(filepath, num_bytes=8192)

Detect the character encoding of a file.

This helper tries UTF-8 first (most common), then uses :mod:chardet to guess an encoding. For low-confidence results, it will try reading more data to improve accuracy.

Parameters:

Name Type Description Default
filepath Path

Path to the file to probe.

required
num_bytes int

Initial number of bytes to read from the start of the file for detection (defaults to 8192).

8192

Returns:

Type Description
str | None

The name of the detected encoding, or None if unknown.

iter_filepath_lines(filepath)

Yield lines from filepath using a detected encoding.

This generator will open the file with the encoding guessed by :func:detect_encoding and yield each line as a string. Useful for processing files with unknown encodings safely.

Parameters:

Name Type Description Default
filepath Path

Path to the file to read.

required

Yields:

Type Description
str

Lines from the file as unicode strings (with trailing newline if

str

present).

read_json(filepath)

Read and parse JSON from filepath using detected encoding.

Parameters:

Name Type Description Default
filepath Path

Path to a JSON file.

required

Returns:

Type Description
dict | list | None

The parsed JSON object (typically a dict or list). If the

dict | list | None

file is empty or cannot be parsed a JSONDecodeError will propagate.

save_json(filepath, data, sort_keys=True)

Serialize data to a JSON file using UTF-8 encoding.

The output file is written with indentation and UTF-8 encoding. By default keys are sorted to produce stable output.

Parameters:

Name Type Description Default
filepath Path

Path where the JSON will be written.

required
data dict

A JSON-serializable Python object (commonly a dict).

required
sort_keys bool

If True, sort dictionary keys in the output for deterministic result.

True

glob_respect_gitignore(directory, glob='*')

Yield entries matching glob at top level of directory while skipping those ignored by any applicable .gitignore files.

Parameters:

Name Type Description Default
directory Path

Directory to glob in.

required
glob str

Glob pattern (defaults to "*").

'*'

Yields:

Type Description
Path

Paths that match the glob and are not ignored.

iter_dirs_with_respect_gitignore(directory, respect_gitignore=False)

Yield directories under directory recursively, optionally respecting .gitignore rules.

Parameters:

Name Type Description Default
directory Path

Root directory to iterate.

required
respect_gitignore bool

If True, discovered .gitignore files are respected and ignored directories are skipped.

False

Yields:

Type Description
Path

Path objects for every directory discovered.

iter_files_with_respect_gitignore(directory, respect_gitignore=False)

Yield files under directory recursively, optionally respecting .gitignore rules.

Parameters:

Name Type Description Default
directory Path

Root directory to iterate.

required
respect_gitignore bool

If True, discovered .gitignore files are respected and ignored files are skipped.

False

Yields:

Type Description
Path

Path objects for every file discovered.

load_directory_gitignore_specs(directory)

Load all .gitignore files found under directory.

The returned mapping maps the directory that contains each .gitignore file to a GitIgnoreSpec object (from the pathspec package).

Parameters:

Name Type Description Default
directory Path

Root directory to search recursively for .gitignore files.

required

Returns:

Type Description
dict[Path, GitIgnoreSpec]

A dict mapping the parent directory of each discovered .gitignore

dict[Path, GitIgnoreSpec]

file to a GitIgnoreSpec that can be used to test whether paths

dict[Path, GitIgnoreSpec]

relative to that directory match ignore patterns.

Example

specs = load_directory_gitignore_specs(Path("/my/repo")) repo_root_spec = specs.get(Path('/my/repo')) bool(repo_root_spec.match_file('build/'))

path_walk(directory)

Lightweight wrapper around os.walk that yields Path objects.

Yields tuples (root_path, dirs, files) where root_path is a Path instance and dirs/files are lists of Path objects representing the directory entries in that root.

Parameters:

Name Type Description Default
directory Path

The root directory to walk.

required

Yields:

Type Description
tuple[Path, list[Path], list[Path]]

Tuples of (root_path, dirs, files).

path_walk_respect_gitignore(directory)

Walk directory tree while applying discovered .gitignore rules.

This function behaves like :func:path_walk but filters out any files or directories that would be ignored according to any .gitignore files found under directory.

Parameters:

Name Type Description Default
directory Path

Root directory to traverse.

required

Yields:

Type Description
Path

Tuples of (root_path, dirs, files) where dirs and files have

list[Path]

been filtered to exclude ignored entries.

rglob_respect_gitignore(directory, glob='*')

Recursively glob for files under directory while skipping ignored paths according to discovered .gitignore files.

Parameters:

Name Type Description Default
directory Path

Root directory to search.

required
glob str

Glob pattern for recursive search (defaults to "*").

'*'

Yields:

Type Description
Path

Matching Path objects that are not ignored.

should_path_ignore(path, specs)

Return True if path should be ignored by git rules.

This function checks for two things: - If any path component equals .git the path is considered ignored. - For each discovered .gitignore specification, if the specification applies to an ancestor directory of path then the path is tested relative to that ancestor and matched via the spec.

Parameters:

Name Type Description Default
path Path

A file or directory path to test.

required
specs dict[Path, GitIgnoreSpec]

Mapping from directories (parents of .gitignore files) to their compiled GitIgnoreSpec objects.

required

Returns:

Type Description
bool

True if the path should be ignored according to any applicable

bool

.gitignore spec or because it is inside a .git directory.

JsonFormatter

Bases: Formatter

format(record)

Format a LogRecord as a JSON string.

The returned value is a JSON-encoded object containing the timestamp, level, message and common record metadata. If exception information is present it will be included as a string under exc_info. Any extra attributes supplied to the logging call (commonly passed via the extra parameter) are merged into the produced JSON object.

Parameters:

Name Type Description Default
record LogRecord

The :class:logging.LogRecord to format.

required

Returns:

Type Description
str

A JSON string representing the log record.

TextFormatter

Bases: Formatter

A plain-text formatter that appends any extra= fields to the log line.

The base :class:logging.Formatter is applied first (using the fmt and datefmt arguments passed to the constructor), then any extra attributes supplied via extra= are appended as key=value pairs separated by spaces.

format(record)

Format record as plain text, appending any extra= fields.

Parameters:

Name Type Description Default
record LogRecord

The :class:logging.LogRecord to format.

required

Returns:

Type Description
str

A formatted string with extra fields appended when present.

setup_json_logger(name, level=logging.INFO, file_logging=False, max_bytes=10 * 1024 * 1024, backup_count=10)

Configure and return the root logger that emits JSON-formatted logs.

This is identical to :func:setup_logger except that any file handler created will use :class:JsonFormatter so persisted logs are JSON lines (.jsonl). A console handler is still attached which emits JSON for structured log consumers.

Parameters:

Name Type Description Default
name str

The name of the logger to return (usually __name__ of the caller module).

required
level int

The default logging level to set on the root logger if not in debug mode. Default is logging.INFO.

INFO
file_logging bool | Path

Controls rotating file logging.

  • False (default) — no file logging.
  • True — write to $XDG_STATE_HOME/<app>/log/<name>.jsonl where app is the top-level package component of name.
  • :class:~pathlib.Path — write to the given file path directly (the parent directory is created if necessary).
False
max_bytes int

Maximum size in bytes of each log file before rotation. Default is 10 MiB.

10 * 1024 * 1024
backup_count int

Number of rotated backup files to keep. Default is 10.

10

Returns:

Type Description
Logger

A configured :class:logging.Logger instance.

setup_logger(name, level=logging.INFO, file_logging=False, max_bytes=10 * 1024 * 1024, backup_count=10)

Configure and return the root logger for the process.

Parameters:

Name Type Description Default
name str

The name of the logger to return (usually __name__ of the caller module).

required
level int

The default logging level to set on the root logger if not in debug mode. Default is logging.INFO.

INFO
file_logging bool | Path

Controls rotating file logging.

  • False (default) — no file logging.
  • True — write to $XDG_STATE_HOME/<app>/log/<name>.log where app is the top-level package component of name.
  • :class:~pathlib.Path — write to the given file path directly (the parent directory is created if necessary).
False
max_bytes int

Maximum size in bytes of each log file before rotation. Default is 10 MiB.

10 * 1024 * 1024
backup_count int

Number of rotated backup files to keep. Default is 10.

10

Returns:

Type Description
Logger

A configured :class:logging.Logger instance for name.

TarFileZstd

Bases: TarFile

zstopen(name=None, mode='r', fileobj=None, level_or_option=None, zstd_dict=None, **kwargs) classmethod

Open a zstd-compressed tar archive for reading or writing.

This classmethod extends :class:tarfile.TarFile with support for zstd-compressed tar archives using :mod:pyzstd. The returned object behaves like a standard TarFile and can be used to list and extract members.

Notes
  • Mode must be one of 'r', 'w' or 'x'; append mode is intentionally not supported.
  • fileobj may be a file-like object or None. When a file-like object is provided it will be wrapped by a :class:pyzstd.ZstdFile instance.

Parameters:

Name Type Description Default
name str | PathLike[str] | None

Path-like or file-like object identifying the archive.

None
mode Literal['r', 'w', 'x']

One of 'r', 'w', or 'x'.

'r'
fileobj Any

Optional file-like object to read from or write to.

None
level_or_option int | Mapping[int, int] | None

Optional compression level (or options) passed through to :class:pyzstd.ZstdFile.

None
zstd_dict ZstdDict | None

Optional dictionary for zstd compression/decompression.

None
**kwargs Any

Additional arguments forwarded to :meth:taropen.

{}

Returns:

Name Type Description
A TarFile

class:tarfile.TarFile instance that can operate on the

TarFile

decompressed tarstream.

Raises:

Type Description
ValueError

If mode is not one of the supported values.

ReadError

If attempting to read a non-zstd file (wrapped from :class:pyzstd.ZstdError or :class:EOFError).

ZstdTarReader(name, *, zstd_dict=None, level_or_option=None, **kwargs)

Context manager that yields a :class:tarfile.TarFile for a zstd compressed archive.

This helper decompresses the zstd archive into a temporary file-like object and opens it via :class:tarfile.TarFile. It is useful when the archive consumer expects a tarfile object rather than a streaming wrapper. The temporary file is cleaned up automatically on exit.

Parameters:

Name Type Description Default
name str | PathLike[str]

Path-like or file-like object pointing to the zstd-compressed tar archive.

required
zstd_dict ZstdDict | None

Optional dictionary for zstd decompression.

None
level_or_option int | Mapping[int, int] | None

Optional level or options passed to pyzstd.

None
**kwargs Any

Forwarded to :class:tarfile.TarFile when opening the decompressed tar stream.

{}

Yields:

Type Description
TarFile

An open :class:tarfile.TarFile instance for the decompressed archive.

Example

with ZstdTarReader('archive.tar.zst') as tar: ... tar.extractall('outdir')

Notification channels: Telegram, DingTalk, Feishu, and WeChat Work.

BaseNotifier

Bases: ABC

Abstract base for notification channel senders.

Implementations must provide a :meth:send method that returns True on success. Concrete classes may add richer methods (e.g. send_markdown) that raise on failure.

send(text) abstractmethod

Send a simple text notification.

Returns True on success. Should not raise — implementation is expected to catch and log errors internally.

ConfigFile

Bases: BaseModel

Root model for a notify.toml file.

.. code-block:: toml

[notify.alerts]
type = "wechat"
key = "..."

[notify.logbot]
type = "telegram"
token = "..."
default_chat_id = "-100..."

DingTalkBot

Bases: BaseNotifier

DingTalk custom bot for sending messages via webhook.

Parameters:

Name Type Description Default
access_token str

The access token from the DingTalk bot webhook URL.

required
secret str

Optional signing secret. When provided every request is signed with HMAC-SHA256.

''

FeishuBot

Bases: BaseNotifier

Send notifications via Feishu / Lark custom bot webhook.

Parameters:

Name Type Description Default
webhook_url str

Full webhook URL from the Feishu bot settings.

required
secret str

Optional signing secret for message verification. When set, every request includes a timestamp and sign field computed with HMAC-SHA256 over an empty body using timestamp + "\n" + secret as the key.

''

TelegramBot

Bases: BaseNotifier

Telegram Bot client for one-way notifications.

Parameters:

Name Type Description Default
token str

Bot token like "123456:ABC-DEF...".

required
chat_id int | str

Target chat identifier (user, group, or channel).

required
timeout float

HTTP request timeout in seconds.

5.0

send_rich_message(html=None, markdown=None, **kwargs)

Send a rich formatted message via sendRichMessage.

Exactly one of html or markdown must be provided. See the Rich Markdown and Rich HTML docs for syntax.

.. _Rich Markdown: https://core.telegram.org/bots/api#rich-markdown-style .. _Rich HTML: https://core.telegram.org/bots/api#rich-html-style

WechatWorkBot

Bases: BaseNotifier

WeChat Work bot for sending messages via webhook.

Parameters:

Name Type Description Default
key str

The webhook key (the key query parameter in the webhook URL).

required