Skip to content

logging_utils

Logging utilities for structured JSON logging to Seq.

The logging utils are made available to easily log events and make them available on Seq.

Installation

get_logger is the main entry point for logging. Import it from the physical_operations_utils package:

from physical_operations_utils.logging_utils import get_logger

Basic usage

Call get_logger() to obtain a configured loguru <https://loguru.readthedocs.io/>_ logger instance. The logger writes JSON-serialized records to stderr, ready for ingestion by Seq.

from physical_operations_utils.logging_utils import get_logger

logger = get_logger()

logger.info("Processing started")
logger.warning("Something looks odd")
logger.error("Something went wrong")
logger.debug("Detailed diagnostics")
logger.success("Task completed successfully")
logger.critical("System is in a bad state")

Each call emits a single JSON line to stderr containing the message, log level, timestamp, source location, and metadata fields.

Setting the team name

Use the team parameter to tag every log record with your team name. This makes it easy to filter logs in Seq by team. The default is "physical_operations".

logger = get_logger(team="trading")
logger.info("Order placed")
# => {"team": "trading", "@m": "Order placed", ...}

Adding application name and job ID

For extra context you can pass application_name and/or job_id. These fields are only included in the JSON output when explicitly provided:

logger = get_logger(
    team="analytics",
    application_name="price_forecast",
    job_id="run-2025-03-06",
)
logger.info("Forecast complete")
# => {..., "application_name": "price_forecast", "job_id": "run-2025-03-06", ...}

If you omit them, the keys will not appear in the log output at all, keeping records compact.

Environment detection

The logger reads the ENVIRONMENT environment variable and writes its value into every log record under the environment key. If the variable is not set, it defaults to "nonprod":

import os

os.environ["ENVIRONMENT"] = "production"

logger = get_logger()
logger.info("Starting job")
# => {..., "environment": "production", ...}

This lets you distinguish production logs from development or test logs without changing any code.

Logging exceptions

Inside an except block, use logger.exception() to log the message and attach the full traceback. The traceback is stored in the @x field of the JSON record:

logger = get_logger()

try:
    result = 1 / 0
except Exception:
    logger.exception("Calculation failed")
# => {..., "@m": "Calculation failed", "@l": "ERROR",
#     "@x": "Traceback (most recent call last):\n ...", ...}

Log output format

Every log record is a single JSON object written to stderr. The fields are:

Key Description Example
@t Timestamp in ISO 8601 format "2025-03-06T09:15:00.00Z"
@m Log message "Processing started"
@l Log level "INFO", "ERROR"
module Python module name "my_module"
file Source file path "src/my_module.py"
line Line number 42
environment Value of ENVIRONMENT env var (default nonprod) "production"
team Team name passed to get_logger "physical_operations"
application_name Application name (only when provided) "price_forecast"
job_id Job identifier (only when provided) "run-2025-03-06"
@x Exception traceback (only on exceptions) "Traceback (most recent..."

This format is designed so that Seq (and other structured-logging platforms) can parse and index every field automatically.

get_logger(team='physical_operations', application_name=None, job_id=None)

Return a Seq-tailored loguru logger instance.

This logger is configured to serialize log records into JSON format and includes additional metadata such as environment and team information. The logger is set up to write to standard error (stderr) and uses a custom serialization function to format log messages which Seq can consume.

Parameters:

Name Type Description Default
team str

The team name to include in the log output. Defaults to "physical_operations".

'physical_operations'
application_name str | None

The application name to include in the log output. Only included in the JSON output when explicitly provided. Defaults to None.

None
job_id str | None

The job id to include in the log output. Only included in the JSON output when explicitly provided. Defaults to None.

None

Returns:

Name Type Description
Logger Logger

A loguru logger instance configured for JSON serialization.

Example
from physical_operations_utils.logging_utils import get_logger

logger = get_logger(team="myteam")
logger.info("This is an info message.")
logger.error("This is an error message.")
logger.exception("An exception occurred.")
logger.success("This is a success message.")
logger.critical("This is a critical message.")
logger.debug("This is a debug message.")
logger.warning("This is a warning message.")
Source code in physical_operations_utils/logging_utils.py
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def get_logger(
    team: str = "physical_operations",
    application_name: Optional[str] = None,
    job_id: Optional[str] = None,
) -> Logger:
    """Return a Seq-tailored loguru logger instance.

    This logger is configured to serialize log records into JSON format and
    includes additional metadata such as environment and team information.
    The logger is set up to write to standard error (stderr) and uses a custom
    serialization function to format log messages which Seq can consume.

    Parameters:
        team (str): The team name to include in the log output. Defaults to ``"physical_operations"``.
        application_name (str | None): The application name to include in the log output. Only
            included in the JSON output when explicitly provided. Defaults to ``None``.
        job_id (str | None): The job id to include in the log output. Only included in the
            JSON output when explicitly provided. Defaults to ``None``.

    Returns:
        Logger: A loguru logger instance configured for JSON serialization.

    Example:
        ```python
        from physical_operations_utils.logging_utils import get_logger

        logger = get_logger(team="myteam")
        logger.info("This is an info message.")
        logger.error("This is an error message.")
        logger.exception("An exception occurred.")
        logger.success("This is a success message.")
        logger.critical("This is a critical message.")
        logger.debug("This is a debug message.")
        logger.warning("This is a warning message.")
        ```
    """
    return logger.patch(
        make_patching(team=team, application_name=application_name, job_id=job_id)
    )