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 |
required |
deepcopy_first
|
bool
|
If True (default) produce a deep copy of |
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 |
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 | 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 |
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 |
False
|
Yields:
| Type | Description |
|---|---|
Path
|
|
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 |
False
|
Yields:
| Type | Description |
|---|---|
Path
|
|
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 |
required |
Returns:
| Type | Description |
|---|---|
dict[Path, GitIgnoreSpec]
|
A dict mapping the parent directory of each discovered |
dict[Path, GitIgnoreSpec]
|
file to a |
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 |
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 |
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 |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the path should be ignored according to any applicable |
bool
|
|
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: |
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: |
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 |
required |
level
|
int
|
The default logging level to set on the root logger if not in
debug mode. Default is |
INFO
|
file_logging
|
bool | Path
|
Controls rotating file logging.
|
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: |
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 |
required |
level
|
int
|
The default logging level to set on the root logger if not in
debug mode. Default is |
INFO
|
file_logging
|
bool | Path
|
Controls rotating file logging.
|
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: |
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. fileobjmay be a file-like object orNone. When a file-like object is provided it will be wrapped by a :class:pyzstd.ZstdFileinstance.
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'
|
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: |
None
|
zstd_dict
|
ZstdDict | None
|
Optional dictionary for zstd compression/decompression. |
None
|
**kwargs
|
Any
|
Additional arguments forwarded to :meth: |
{}
|
Returns:
| Name | Type | Description |
|---|---|---|
A |
TarFile
|
class: |
TarFile
|
decompressed tarstream. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
ReadError
|
If attempting to read a non-zstd file (wrapped from
:class: |
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: |
{}
|
Yields:
| Type | Description |
|---|---|
TarFile
|
An open :class: |
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 |
''
|
TelegramBot
¶
Bases: BaseNotifier
Telegram Bot client for one-way notifications.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
Bot token like |
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 |
required |