Skip to content

dagster_json_formatter

kedro_dagster.logging.dagster_json_formatter()

Create a JSON formatter for Dagster logging.

This formatter produces structured JSON output suitable for log aggregation systems, monitoring tools, and production environments. Each log entry is a single JSON object with consistent field names and ISO timestamps.

Returns

Type Description
ProcessorFormatter

A formatter configured for JSON output with automatic fallback for older structlog versions.

See Also

kedro_dagster.logging.dagster_rich_formatter : Rich console formatter for development use. kedro_dagster.logging.dagster_colored_formatter : Colored formatter using coloredlogs.

Source Code

View on GitHub

Show/Hide source 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def dagster_json_formatter() -> structlog.stdlib.ProcessorFormatter:
    """Create a JSON formatter for Dagster logging.

    This formatter produces structured JSON output suitable for log aggregation
    systems, monitoring tools, and production environments. Each log entry is
    a single JSON object with consistent field names and ISO timestamps.

    Returns
    -------
    structlog.stdlib.ProcessorFormatter
        A formatter configured for JSON output with automatic fallback
        for older structlog versions.

    See Also
    --------
    `kedro_dagster.logging.dagster_rich_formatter` :
        Rich console formatter for development use.
    `kedro_dagster.logging.dagster_colored_formatter` :
        Colored formatter using coloredlogs.
    """
    foreign_pre_chain = [
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.processors.StackInfoRenderer(),
        structlog.stdlib.ExtraAdder(),
    ]

    json_renderer = structlog.processors.JSONRenderer(sort_keys=True, ensure_ascii=False)
    processors = [
        structlog.stdlib.ProcessorFormatter.remove_processors_meta,
        json_renderer,
    ]

    try:
        # Try the newer API with processors list
        return structlog.stdlib.ProcessorFormatter(
            foreign_pre_chain=foreign_pre_chain,
            processors=processors,
        )
    except TypeError:
        # Fallback to older API with single processor
        return structlog.stdlib.ProcessorFormatter(
            foreign_pre_chain=foreign_pre_chain,
            processor=json_renderer,
        )