b2c_tooling_sdk.operations.logs

Log operations for B2C Commerce instances.

Mirrors src/operations/logs/index.ts. Provides functions for listing, tailing, and analyzing log files on B2C Commerce instances via WebDAV.

Example - list log files:

from b2c_tooling_sdk.operations.logs import list_log_files, ListLogsOptions

files = await list_log_files(
    instance, ListLogsOptions(prefixes=["error", "customerror"], sort_by="date", sort_order="desc")
)

for file in files:
    print(f"{file.name} ({file.size} bytes)")

Example - tail logs in real-time:

from b2c_tooling_sdk.operations.logs import (
    tail_logs, TailLogsOptions, create_path_normalizer, PathNormalizerOptions,
)

normalizer = create_path_normalizer(PathNormalizerOptions(cartridge_path="./cartridges"))

result = await tail_logs(
    instance,
    TailLogsOptions(
        prefixes=["error", "customerror"],
        path_normalizer=normalizer,
        on_entry=lambda entry: print(f"[{entry.file}] {entry.level}: {entry.message}"),
        on_error=lambda err: print(f"Error: {err}"),
    ),
)

# Stop after 30 seconds.
await asyncio.sleep(30)
await result.stop()
await result.done

Example - get recent logs (one-shot):

from b2c_tooling_sdk.operations.logs import get_recent_logs, GetRecentLogsOptions

entries = await get_recent_logs(instance, GetRecentLogsOptions(prefixes=["error"], max_entries=50))

for entry in entries:
    print(f"[{entry.timestamp}] {entry.message}")

Classes

GetRecentLogsOptions

class GetRecentLogsOptions

Options for getting recent logs (one-shot retrieval).

Fields

Name	Type	Default
prefixes	list[str]	field(default_factory=lambda: ['error', 'customerror'])
max_entries	int	100
tail_bytes	int	65536
path_normalizer	Callable[[str], str] | None	None

ListLogsOptions

class ListLogsOptions

Options for listing log files.

Fields

Name	Type	Default
prefixes	list[str] | None	None
sort_by	Literal['name', 'date', 'size']	'date'
sort_order	Literal['asc', 'desc']	'desc'

LogEntry

class LogEntry

Represents a parsed log entry.

Fields

Name	Type	Default
file	str
message	str
raw	str
level	str | None	None
timestamp	str | None	None

LogFile

class LogFile

Represents a log file on a B2C Commerce instance.

Fields

Name	Type
name	str
prefix	str
size	int
last_modified	datetime
path	str

PathNormalizerOptions

class PathNormalizerOptions

Options for creating a path normalizer.

Fields

Name	Type	Default
cartridge_path	str | None	None
cartridges	list[CartridgeMapping]	field(default_factory=list)

TailLogsCallbacks

class TailLogsCallbacks

Callback functions for tail operation events.

Fields

Name	Type	Default
on_entry	Callable[[LogEntry], None] | None	None
on_error	Callable[[Exception], None] | None	None
on_file_discovered	Callable[[LogFile], None] | None	None
on_file_rotated	Callable[[LogFile], None] | None	None

TailLogsOptions

class TailLogsOptions(TailLogsCallbacks)

Options for tailing logs.

Fields

Name	Type	Default
prefixes	list[str]	field(default_factory=lambda: ['error', 'customerror'])
poll_interval	float	3.0
last_entries	int	1
max_entries	int | None	None
path_normalizer	Callable[[str], str] | None	None

TailLogsResult

class TailLogsResult

Result of a tail operation.

Fields

Name	Type
stop	Callable[[], Awaitable[None]]
files	list[LogFile]
entries	list[LogEntry]
done	Awaitable[None]

Functions

aggregate_log_entries

def aggregate_log_entries(lines: list[str], pending_lines: list[str] | None = None) -> AggregatedLogEntries

Aggregate lines into multi-line log entries.

B2C log entries can span multiple lines. A new entry starts when a line begins with a timestamp pattern: [YYYY-MM-DD HH:MM:SS.mmm GMT].

Parameters

Name	Type	Description
lines	list[str]	Individual lines.
pending_lines	list[str] | None	Lines carried over from the previous chunk (incomplete entry).

Returns: Complete entries and any pending lines for the next chunk.

create_path_normalizer

def create_path_normalizer(options: PathNormalizerOptions) -> Callable[[str], str] | None

Create a path normalizer function for converting remote cartridge paths to local paths in log messages.

Supports two modes:

1. Cartridge mappings (precise): Uses discovered cartridges to map each cartridge name to its actual local path. Best for projects with cartridges in different locations.
2. Cartridge path (simple): Prefixes all paths with a base directory. Best when all cartridges are in a single directory.

Parameters

Name	Type	Description
options	PathNormalizerOptions	Normalizer options.

Returns: Function that normalizes paths in a message string, or None if no options were provided. >>> cartridges = find_cartridges('./my-project') >>> normalize = create_path_normalizer(PathNormalizerOptions(cartridges=cartridges)) >>> # Or, using a simple cartridge path: >>> normalize = create_path_normalizer(PathNormalizerOptions(cartridge_path='./cartridges')) >>> # Input: "(app_storefront/cartridge/controllers/Home.js:45)" >>> # Output: "(./cartridges/app_storefront/cartridge/controllers/Home.js:45)"

discover_and_create_normalizer

def discover_and_create_normalizer(directory: str | None = None) -> Callable[[str], str] | None

Discover cartridges and create a path normalizer automatically.

Convenience function that combines find_cartridges with create_path_normalizer for easy setup. Cartridge paths are converted to relative paths from the current project directory.

Parameters

Name	Type	Description
directory	str | None	Directory to search for cartridges (defaults to cwd).

Returns: Path normalizer function, or None if no cartridges were found. >>> normalize = discover_and_create_normalizer() >>> normalize = discover_and_create_normalizer('./my-project')

extract_paths

def extract_paths(message: str) -> list[str]

Extract all cartridge paths from a message.

Useful for testing or analysis of log messages.

Parameters

Name	Type	Description
message	str	Log message to extract paths from.

Returns: List of extracted paths. >>> extract_paths("Error at (app_storefront/cartridge/controllers/Home.js:45)") ['app_storefront/cartridge/controllers/Home.js:45']

extract_prefix

def extract_prefix(filename: str) -> str

Extract the log prefix from a filename.

Parameters

Name	Type	Description
filename	str	Log file name (e.g., "error-blade1-20250125.log").

Returns: The prefix (e.g., "error") or "unknown". >>> extract_prefix("error-blade1-20250125.log") 'error' >>> extract_prefix("customerror-blade1-20250125.log") 'customerror' >>> extract_prefix("custom-mylog-blade1-20250125.log") 'custom-mylog'

filter_by_level

def filter_by_level(entries: list[LogEntry], levels: list[str]) -> list[LogEntry]

Filter entries by log level.

filter_by_search

def filter_by_search(entries: list[LogEntry], search: str) -> list[LogEntry]

Filter entries by text search (case-insensitive substring match).

filter_by_since

def filter_by_since(entries: list[LogEntry], since: datetime) -> list[LogEntry]

Filter entries by timestamp.

get_recent_logs

async def get_recent_logs(instance: B2CInstance, options: GetRecentLogsOptions | None = None) -> list[LogEntry]

Get recent log entries (one-shot retrieval).

Useful for MCP server integration or programmatic access without continuous tailing. Reads the tail end of log files and returns parsed entries.

Parameters

Name	Type	Description
instance	B2CInstance	B2C instance to get logs from.
options	GetRecentLogsOptions | None	Retrieval options.

Returns: Recent log entries.

list_log_files

async def list_log_files(instance: B2CInstance, options: ListLogsOptions | None = None) -> list[LogFile]

List log files on a B2C Commerce instance.

Filters in ListLogsOptions.prefixes are matched in one of two ways:

• Prefix filters (no /, e.g. "error") match the extracted log-category prefix of files in the top-level Logs/ directory.
• Path filters (contain a /, e.g. "internal/server") recurse into the named subdirectory of Logs/ and match against each file's path relative to Logs/. This is the only case that lists subdirectories — by default only the top-level Logs/ directory is scanned.

Parameters

Name	Type	Description
instance	B2CInstance	B2C instance to list logs from.
options	ListLogsOptions | None	Listing options (filters, sorting).

Returns: List of log files.

matches_level

def matches_level(entry: LogEntry, levels: list[str]) -> bool

Check if a single entry matches the specified log levels.

Used for streaming/tail scenarios where entries are filtered one at a time.

matches_search

def matches_search(entry: LogEntry, search: str) -> bool

Check if a single entry matches the search text (case-insensitive).

Used for streaming/tail scenarios where entries are filtered one at a time.

parse_log_entry

def parse_log_entry(first_line: str, file: str, full_message: str, path_normalizer: Callable[[str], str] | None = None) -> LogEntry

Parse the first line of a log entry to extract timestamp, level, and message.

Expected format: [timestamp GMT] LEVEL context - message. Example: [2025-01-25 10:30:45.123 GMT] ERROR PipelineCallServlet|... - Error message.

The message field will contain:

• The content portion from the first line (after LEVEL)
• Plus any continuation lines (stack traces, etc.)

If the standard B2C log format is not matched, returns an unparsed entry with only the file, message, and raw fields. The timestamp and level fields will be None in this case, but the raw log line is preserved for debugging or recovery purposes.

Parameters

Name	Type	Description
first_line	str	First line of the log entry.
file	str	File name the entry came from.
full_message	str	Complete raw message including all lines.
path_normalizer	Callable[[str], str] | None	Optional function to normalize paths in the message.

parse_log_timestamp

def parse_log_timestamp(timestamp: str) -> datetime | None

Parse a B2C log timestamp into a datetime.

Expected format: "2025-01-25 10:30:45.123 GMT".

parse_relative_time

def parse_relative_time(time_str: str) -> int | None

Parse a relative time string (e.g., "5m", "1h", "2d") into milliseconds.

Returns: The duration in milliseconds, or None if time_str is not a valid relative time format.

parse_since_time

def parse_since_time(since_str: str, now: datetime | None = None) -> datetime

Parse a since value into a datetime.

Supports:

• Relative times: "5m", "1h", "2d"
• ISO 8601: "2026-01-25T10:00:00"

Parameters

Name	Type	Description
since_str	str	The value to parse.
now	datetime | None	Reference time for relative values (defaults to the current, timezone-aware UTC time). Injectable so callers that resolve several bounds together, or that need deterministic behavior in tests, can pin a single "now". If a naive (tzinfo-less) value is passed, it is used as-is for relative-time arithmetic.

Raises

• ValueError — if since_str is neither a valid relative time nor a valid ISO 8601 timestamp.

split_lines

def split_lines(content: bytes, decoder: codecs.IncrementalDecoder, is_complete: bool = True) -> list[str]

Split content into lines, handling incomplete lines at boundaries.

Uses a UTF-8 incremental decoder (reused across calls for streaming) for proper multi-byte character handling.

Parameters

Name	Type	Description
content	bytes	Raw bytes read from a log file.
decoder	codecs.IncrementalDecoder	Incremental decoder instance (should be reused for streaming).
is_complete	bool	Whether this is the final chunk (flush the decoder).

Returns: Complete lines (without a trailing incomplete line).

tail_logs

async def tail_logs(instance: B2CInstance, options: TailLogsOptions | None = None, *, sleep: Callable[[float], Awaitable[None]] = asyncio.sleep) -> TailLogsResult

Tail log files on a B2C Commerce instance.

Continuously polls for new log content using HTTP Range requests for efficiency. Calls the on_entry callback for each new log line.

Parameters

Name	Type	Description
instance	B2CInstance	B2C instance to tail logs from.
options	TailLogsOptions | None	Tailing options (filters, callbacks, polling interval).
sleep	Callable[[float], Awaitable[None]]	Injectable sleep function (Python-only, see module docstring).

Returns: Tail result with stop() control and a done awaitable.