---
editLink: false
lastUpdated: false
outline: [2, 3]
---

<!-- Generated by python/b2c-tooling-sdk/scripts/generate_api_docs.py. Do not edit. -->

# 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:

```python
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:

```python
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):

```python
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 {#getrecentlogsoptions}

```python
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 {#listlogsoptions}

```python
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 {#logentry}

```python
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 {#logfile}

```python
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 {#pathnormalizeroptions}

```python
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 {#taillogscallbacks}

```python
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 {#taillogsoptions}

```python
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 {#taillogsresult}

```python
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 {#aggregate-log-entries}

```python
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 {#create-path-normalizer}

```python
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 {#discover-and-create-normalizer}

```python
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`](/python/api/operations/code#find-cartridges) with
[`create_path_normalizer`](/python/api/operations/logs#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 {#extract-paths}

```python
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 {#extract-prefix}

```python
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 {#filter-by-level}

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

Filter entries by log level.

### filter_by_search {#filter-by-search}

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

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

### filter_by_since {#filter-by-since}

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

Filter entries by timestamp.

### get_recent_logs {#get-recent-logs}

```python
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 {#list-log-files}

```python
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 {#matches-level}

```python
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 {#matches-search}

```python
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 {#parse-log-entry}

```python
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 {#parse-log-timestamp}

```python
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 {#parse-relative-time}

```python
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 {#parse-since-time}

```python
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 {#split-lines}

```python
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 {#tail-logs}

```python
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.