Skip to content

🏠 Home | ⚙️ API Reference | 🛠️ Common API

🛠️ Common API (Utilities)

This section describes the shared utility functions of the codex-core library.

Phone Normalization

Tools for correctly formatting phone numbers (supports +, 00, and local 0).

phone

Utilities for normalizing phone numbers to a canonical digit-only format.

Framework-agnostic (zero Django / framework dependencies). Suitable for use in any Python 3.10+ environment including async workers.

The module intentionally does not validate that the resulting number is reachable (no libphonenumber dependency); it only ensures a consistent digit-only representation that can be safely stored, compared, and passed to SMS / telephony APIs.

Functions

normalize_phone(phone, default_country='49')

Normalize a phone number to a digit-only international string.

Handles the three most common input variants encountered in European / German-locale data:

  1. International + prefix — +49 151 1234567
  2. International 00 prefix — 0049 151 1234567
  3. Local 0 prefix — 0151 1234567 (expanded using default_country)
  4. Already normalized — 491511234567 (returned as-is)

Non-digit, non-plus characters (spaces, hyphens, parentheses) are stripped before prefix detection.

Parameters:

Name Type Description Default
phone str

Raw phone string in any common format. Empty string or strings containing no digits/plus return "".

 required
default_country str

ITU-T country code (digits only, no +) prepended when a local 0-prefix number is detected. Defaults to "49" (Germany).

 '49'

Returns:

Type Description
str

Digit-only string representing the international phone number,
str

or an empty string if the input is blank or contains no
str

recognizable digits.
Note

No length or reachability validation is performed. The caller is responsible for validating that the result conforms to the expected E.164 length for the target country.

Example 
normalize_phone("0151 1234567")          # → "491511234567"
normalize_phone("+49 151 1234567")        # → "491511234567"
normalize_phone("0049 151 1234567")       # → "491511234567"
normalize_phone("+1-800-555-0100", "1")   # → "18005550100"
normalize_phone("")                        # → ""
Source code in src/codex_core/common/phone.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def normalize_phone(phone: str, default_country: str = "49") -> str:
    """Normalize a phone number to a digit-only international string.

    Handles the three most common input variants encountered in
    European / German-locale data:

    1. **International ``+`` prefix** — ``+49 151 1234567``
    2. **International ``00`` prefix** — ``0049 151 1234567``
    3. **Local ``0`` prefix** — ``0151 1234567`` (expanded using
       *default_country*)
    4. **Already normalized** — ``491511234567`` (returned as-is)

    Non-digit, non-plus characters (spaces, hyphens, parentheses) are
    stripped before prefix detection.

    Args:
        phone: Raw phone string in any common format.  Empty string or
            strings containing no digits/plus return ``""``.
        default_country: ITU-T country code (digits only, no ``+``)
            prepended when a local ``0``-prefix number is detected.
            Defaults to ``"49"`` (Germany).

    Returns:
        Digit-only string representing the international phone number,
        or an empty string if the input is blank or contains no
        recognizable digits.

    Note:
        No length or reachability validation is performed.  The caller
        is responsible for validating that the result conforms to the
        expected E.164 length for the target country.

    Example:
        ```python
        normalize_phone("0151 1234567")          # → "491511234567"
        normalize_phone("+49 151 1234567")        # → "491511234567"
        normalize_phone("0049 151 1234567")       # → "491511234567"
        normalize_phone("+1-800-555-0100", "1")   # → "18005550100"
        normalize_phone("")                        # → ""
        ```
    """
    if not phone:
        return ""

    # Keep only digits and plus (if present at start)
    cleaned = "".join(c for c in phone if c.isdigit() or c == "+")
    if not cleaned:
        return ""

    # 1. Already has '+' -> remove it and return
    if cleaned.startswith("+"):
        return cleaned.replace("+", "")

    # 2. Starts with '00' (European international format)
    if cleaned.startswith("00"):
        return cleaned[2:]

    # 3. Starts with a single '0' (local format, e.g. German)
    if cleaned.startswith("0"):
        return default_country + cleaned[1:]

    # 4. Already normalized number without plus (e.g. 49151...)
    return cleaned

Text Processing

Tools for name normalization, transliteration, and string cleaning.

text

Text processing utilities for name normalization, transliteration, and SMS safety.

Framework-agnostic (zero Django / framework dependencies). All functions are pure and stateless: they accept a string, return a string, and produce no side effects.

Functions are organized around three distinct concerns:

  • Normalization — :func:normalize_name, :func:clean_string produce canonical, whitespace-clean representations.
  • Transliteration — :func:transliterate converts Cyrillic to Latin for systems that do not support Unicode input.
  • Sanitization — :func:sanitize_for_sms strips characters illegal or problematic in SMS payloads.

Functions

normalize_name(name)

Capitalize each word in a personal name, preserving hyphens.

Collapses repeated whitespace, then title-cases every name segment separated by a space or hyphen while leaving the delimiters unchanged.

Parameters:

Name Type Description Default
name str

Raw name string in any case with arbitrary spacing (e.g. " ivan ivanov-petrov ").

 required

Returns:

Type Description
str

Normalized name with each segment capitalized
str

(e.g. "Ivan Ivanov-Petrov"), or an empty string if
str

name is falsy.
Example 
normalize_name("ivan ivanov-petrov")  # → "Ivan Ivanov-Petrov"
normalize_name("ANNA MÜLLER")          # → "Anna Müller"
normalize_name("")                     # → ""
Source code in src/codex_core/common/text.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def normalize_name(name: str) -> str:
    """Capitalize each word in a personal name, preserving hyphens.

    Collapses repeated whitespace, then title-cases every name
    segment separated by a space or hyphen while leaving the
    delimiters unchanged.

    Args:
        name: Raw name string in any case with arbitrary spacing
            (e.g. ``"  ivan   ivanov-petrov  "``).

    Returns:
        Normalized name with each segment capitalized
        (e.g. ``"Ivan Ivanov-Petrov"``), or an empty string if
        *name* is falsy.

    Example:
        ```python
        normalize_name("ivan ivanov-petrov")  # → "Ivan Ivanov-Petrov"
        normalize_name("ANNA MÜLLER")          # → "Anna Müller"
        normalize_name("")                     # → ""
        ```
    """
    if not name:
        return ""

    # Normalize spaces (remove extra ones)
    cleaned_name = " ".join(name.strip().split())

    # Split by spaces OR hyphens, PRESERVING the delimiters
    parts = re.split(r"([\s-])", cleaned_name)

    # Capitalize only text parts, leave delimiters as is
    normalized_parts = [p.capitalize() if p not in (" ", "-") else p for p in parts]

    return "".join(normalized_parts)

clean_string(text)

Collapse multiple whitespace characters into a single space.

Strips leading / trailing whitespace and replaces any internal sequence of whitespace (spaces, tabs, newlines) with a single ASCII space. Invisible Unicode whitespace is also collapsed because str.split() is Unicode-aware.

Parameters:

Name Type Description Default
text str

Raw string that may contain irregular whitespace.

 required

Returns:

Type Description
str

Cleaned string with normalized whitespace, or an empty string
str

if text is falsy.
Example 
clean_string("  hello\t  world\n")  # → "hello world"
clean_string("")                       # → ""
Source code in src/codex_core/common/text.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def clean_string(text: str) -> str:
    """Collapse multiple whitespace characters into a single space.

    Strips leading / trailing whitespace and replaces any internal
    sequence of whitespace (spaces, tabs, newlines) with a single
    ASCII space.  Invisible Unicode whitespace is also collapsed
    because ``str.split()`` is Unicode-aware.

    Args:
        text: Raw string that may contain irregular whitespace.

    Returns:
        Cleaned string with normalized whitespace, or an empty string
        if *text* is falsy.

    Example:
        ```python
        clean_string("  hello\\t  world\\n")  # → "hello world"
        clean_string("")                       # → ""
        ```
    """
    if not text:
        return ""
    return " ".join(text.split())

transliterate(text)

Transliterate Cyrillic characters to their Latin equivalents.

Uses a hard-coded character-level mapping (str.maketrans) covering all 33 letters of the Russian alphabet in both cases. Non-Cyrillic characters pass through unchanged.

The mapping follows a simplified scientific transliteration standard (not GOST 7.79-2000) optimised for readability in Latin-script contexts such as SMS gateways and URL slugs.

Parameters:

Name Type Description Default
text str

Input string containing Cyrillic characters.

 required

Returns:

Type Description
str

String with each Cyrillic character replaced by its Latin
str

approximation, or an empty string if text is falsy.
Note

ъ (hard sign) and ь (soft sign) map to an empty string, reducing the output length relative to the input.

Example 
transliterate("Привет")        # → "Privet"
transliterate("Щукин")         # → "Shchukin"
transliterate("Hello World")   # → "Hello World"  (passthrough)
Source code in src/codex_core/common/text.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def transliterate(text: str) -> str:
    """Transliterate Cyrillic characters to their Latin equivalents.

    Uses a hard-coded character-level mapping (``str.maketrans``)
    covering all 33 letters of the Russian alphabet in both cases.
    Non-Cyrillic characters pass through unchanged.

    The mapping follows a simplified scientific transliteration
    standard (not GOST 7.79-2000) optimised for readability in
    Latin-script contexts such as SMS gateways and URL slugs.

    Args:
        text: Input string containing Cyrillic characters.

    Returns:
        String with each Cyrillic character replaced by its Latin
        approximation, or an empty string if *text* is falsy.

    Note:
        ``ъ`` (hard sign) and ``ь`` (soft sign) map to an empty
        string, reducing the output length relative to the input.

    Example:
        ```python
        transliterate("Привет")        # → "Privet"
        transliterate("Щукин")         # → "Shchukin"
        transliterate("Hello World")   # → "Hello World"  (passthrough)
        ```
    """
    if not text:
        return ""

    translit_map = str.maketrans(
        {
            "а": "a",
            "б": "b",
            "в": "v",
            "г": "g",
            "д": "d",
            "е": "e",
            "ё": "yo",
            "ж": "zh",
            "з": "z",
            "и": "i",
            "й": "y",
            "к": "k",
            "л": "l",
            "м": "m",
            "н": "n",
            "о": "o",
            "п": "p",
            "р": "r",
            "с": "s",
            "т": "t",
            "у": "u",
            "ф": "f",
            "х": "kh",
            "ц": "ts",
            "ч": "ch",
            "ш": "sh",
            "щ": "shch",
            "ъ": "",
            "ы": "y",
            "ь": "",
            "э": "e",
            "ю": "yu",
            "я": "ya",
            "А": "A",
            "Б": "B",
            "В": "V",
            "Г": "G",
            "Д": "D",
            "Е": "E",
            "Ё": "Yo",
            "Ж": "Zh",
            "З": "Z",
            "И": "I",
            "Й": "Y",
            "К": "K",
            "Л": "L",
            "М": "M",
            "Н": "N",
            "О": "O",
            "П": "P",
            "Р": "R",
            "С": "S",
            "Т": "T",
            "У": "U",
            "Ф": "F",
            "Х": "Kh",
            "Ц": "Ts",
            "Ч": "Ch",
            "Ш": "Sh",
            "Щ": "Shch",
            "Ъ": "",
            "Ы": "Y",
            "Ь": "",
            "Э": "E",
            "Ю": "Yu",
            "Я": "Ya",
        }
    )
    return text.translate(translit_map)

sanitize_for_sms(text, max_length=50)

Produce a safe, length-bounded string suitable for SMS payloads.

Applies two successive regex passes:

  1. Replaces \r, \n, \t sequences with a single space to eliminate line breaks that most SMS gateways convert to encoding errors.
  2. Strips all characters outside the set [\w\s.\-] (word characters, spaces, dots, hyphens).

The result is then truncated to max_length characters and stripped of trailing whitespace.

Parameters:

Name Type Description Default
text str

Arbitrary user-supplied or templated string.

 required
max_length int

Maximum number of characters in the output. Defaults to 50, which fits within a single GSM-7 SMS segment alongside typical prefix text.

 50

Returns:

Type Description
str

Sanitized and truncated string, or an empty string if text
str

is falsy.
Note

Truncation happens before the final strip(), so the effective output length may be slightly less than max_length when the truncation boundary falls on whitespace.

Example 
sanitize_for_sms("Hello\nWorld!!! <script>", 20)
# → "Hello World script"

sanitize_for_sms("Appointment at 10:00", 15)
# → "Appointment at"
Source code in src/codex_core/common/text.py 
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
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def sanitize_for_sms(text: str, max_length: int = 50) -> str:
    """Produce a safe, length-bounded string suitable for SMS payloads.

    Applies two successive regex passes:

    1. Replaces ``\\r``, ``\\n``, ``\\t`` sequences with a single space
       to eliminate line breaks that most SMS gateways convert to
       encoding errors.
    2. Strips all characters outside the set ``[\\w\\s.\\-]``
       (word characters, spaces, dots, hyphens).

    The result is then truncated to *max_length* characters and
    stripped of trailing whitespace.

    Args:
        text: Arbitrary user-supplied or templated string.
        max_length: Maximum number of characters in the output.
            Defaults to ``50``, which fits within a single GSM-7
            SMS segment alongside typical prefix text.

    Returns:
        Sanitized and truncated string, or an empty string if *text*
        is falsy.

    Note:
        Truncation happens *before* the final ``strip()``, so the
        effective output length may be slightly less than *max_length*
        when the truncation boundary falls on whitespace.

    Example:
        ```python
        sanitize_for_sms("Hello\\nWorld!!! <script>", 20)
        # → "Hello World script"

        sanitize_for_sms("Appointment at 10:00", 15)
        # → "Appointment at"
        ```
    """
    if not text:
        return ""
    # Remove newlines and control characters
    clean = re.sub(r"[\r\n\t]+", " ", text)
    # Keep only safe characters: word chars, spaces, dots, dashes
    clean = re.sub(r"[^\w\s.\-]", "", clean)
    return clean[:max_length].strip()

Logging

Helpers for structured logging and Loguru configuration.

log_context

Structured logging context: per-request contextvars and per-task wrappers.

Two complementary mechanisms:

  1. contextvars-based context — set_log_context / clear_log_context / get_log_context. Designed for async request/task scoping: set fields once at the start of a request or worker task, and every logger.*() call in that async context automatically includes them via the Loguru patcher configured in :func:~codex_core.common.loguru_setup.setup_logging.

  2. TaskLogContext — a stateful wrapper around logging.Logger for code that uses the standard library directly (see class docstring for details).

Example (contextvars): ```python from codex_core.common.log_context import set_log_context, clear_log_context

set_log_context(request_id="abc-123", char_id=42)
logger.info("CombatMoveSubmitted")  # extra contains request_id, char_id
clear_log_context()
```

Example (TaskLogContext): ```python from codex_core.common.log_context import TaskLogContext

log = TaskLogContext("send_booking_notification", worker="notification_worker")
log.info("Processing appointment", extra={"appointment_id": 123})
```

Classes

TaskLogContext

Structured logging adapter that binds context fields to every record.

Wraps a standard logging.Logger and automatically merges a fixed _base_extra dict (built at construction time) with any per-call extra dict before forwarding the record. This eliminates repetitive extra={"task": ...} boilerplate in worker loops.

The class is stateful in the sense that _base_extra is set once at construction and shared across all log calls. It is not thread-safe to mutate _base_extra after construction; create a new instance per task if context must change.

Parameters:

Name Type Description Default
task_name str

Logical name of the current operation (e.g. "send_booking_notification"). Stored as task in every log record's extra.

 required
logger_name str | None

Name passed to logging.getLogger(). Defaults to task_name when omitted.

 None
**extra Any

Arbitrary keyword arguments added to _base_extra alongside task (e.g. worker="notification_worker").

 {}
Example 
log = TaskLogContext(
    "slot_calculation",
    logger_name="codex.booking",
    worker="slot_worker",
    tenant_id=42,
)
log.debug("Starting slot search")
log.error("Slot not found", extra={"slot_id": 7})
Source code in src/codex_core/common/log_context.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
class TaskLogContext:
    """Structured logging adapter that binds context fields to every record.

    Wraps a standard ``logging.Logger`` and automatically merges a
    fixed ``_base_extra`` dict (built at construction time) with any
    per-call ``extra`` dict before forwarding the record.  This
    eliminates repetitive ``extra={"task": ...}`` boilerplate in
    worker loops.

    The class is **stateful** in the sense that ``_base_extra`` is
    set once at construction and shared across all log calls.  It is
    **not** thread-safe to mutate ``_base_extra`` after construction;
    create a new instance per task if context must change.

    Args:
        task_name: Logical name of the current operation
            (e.g. ``"send_booking_notification"``).  Stored as
            ``task`` in every log record's ``extra``.
        logger_name: Name passed to ``logging.getLogger()``.
            Defaults to *task_name* when omitted.
        **extra: Arbitrary keyword arguments added to ``_base_extra``
            alongside ``task`` (e.g. ``worker="notification_worker"``).

    Example:
        ```python
        log = TaskLogContext(
            "slot_calculation",
            logger_name="codex.booking",
            worker="slot_worker",
            tenant_id=42,
        )
        log.debug("Starting slot search")
        log.error("Slot not found", extra={"slot_id": 7})
        ```
    """

    def __init__(self, task_name: str, logger_name: str | None = None, **extra: Any) -> None:
        self._logger = logging.getLogger(logger_name or task_name)
        self._base_extra = {"task": task_name, **extra}

    def _merge_extra(self, extra: dict[str, Any] | None) -> dict[str, Any]:
        """Merge per-call extra fields with the base context.

        Args:
            extra: Optional per-call fields.  Keys in *extra* take
                precedence over identically named keys in
                ``_base_extra``.

        Returns:
            A new ``dict`` containing the merged fields.
        """
        if extra:
            return {**self._base_extra, **extra}
        return self._base_extra

    def debug(self, msg: str, *args: Any, extra: dict[str, Any] | None = None, **kwargs: Any) -> None:
        """Emit a ``DEBUG`` record enriched with the bound context fields.

        Args:
            msg: Log message format string.
            *args: Positional arguments passed to ``Logger.debug``.
            extra: Per-call structured fields merged with ``_base_extra``.
            **kwargs: Additional keyword arguments forwarded to ``Logger.debug``.
        """
        self._logger.debug(msg, *args, extra=self._merge_extra(extra), **kwargs)

    def info(self, msg: str, *args: Any, extra: dict[str, Any] | None = None, **kwargs: Any) -> None:
        """Emit an ``INFO`` record enriched with the bound context fields.

        Args:
            msg: Log message format string.
            *args: Positional arguments passed to ``Logger.info``.
            extra: Per-call structured fields merged with ``_base_extra``.
            **kwargs: Additional keyword arguments forwarded to ``Logger.info``.
        """
        self._logger.info(msg, *args, extra=self._merge_extra(extra), **kwargs)

    def warning(self, msg: str, *args: Any, extra: dict[str, Any] | None = None, **kwargs: Any) -> None:
        """Emit a ``WARNING`` record enriched with the bound context fields.

        Args:
            msg: Log message format string.
            *args: Positional arguments passed to ``Logger.warning``.
            extra: Per-call structured fields merged with ``_base_extra``.
            **kwargs: Additional keyword arguments forwarded to ``Logger.warning``.
        """
        self._logger.warning(msg, *args, extra=self._merge_extra(extra), **kwargs)

    def error(self, msg: str, *args: Any, extra: dict[str, Any] | None = None, **kwargs: Any) -> None:
        """Emit an ``ERROR`` record enriched with the bound context fields.

        Args:
            msg: Log message format string.
            *args: Positional arguments passed to ``Logger.error``.
            extra: Per-call structured fields merged with ``_base_extra``.
            **kwargs: Additional keyword arguments forwarded to ``Logger.error``.
        """
        self._logger.error(msg, *args, extra=self._merge_extra(extra), **kwargs)

    def exception(self, msg: str, *args: Any, extra: dict[str, Any] | None = None, **kwargs: Any) -> None:
        """Emit an ``ERROR`` record with exception traceback and bound context.

        Equivalent to :meth:`error` but always captures the current
        exception info (``exc_info=True`` is implicit).  Call inside an
        ``except`` block.

        Args:
            msg: Log message format string.
            *args: Positional arguments passed to ``Logger.exception``.
            extra: Per-call structured fields merged with ``_base_extra``.
            **kwargs: Additional keyword arguments forwarded to ``Logger.exception``.
        """
        self._logger.exception(msg, *args, extra=self._merge_extra(extra), **kwargs)
Functions
debug(msg, *args, extra=None, **kwargs)

Emit a DEBUG record enriched with the bound context fields.

Parameters:

Name Type Description Default
msg str

Log message format string.

 required
*args Any

Positional arguments passed to Logger.debug.

 ()
extra dict[str, Any] | None

Per-call structured fields merged with _base_extra.

 None
**kwargs Any

Additional keyword arguments forwarded to Logger.debug.

 {}
Source code in src/codex_core/common/log_context.py 
117
118
119
120
121
122
123
124
125
126
def debug(self, msg: str, *args: Any, extra: dict[str, Any] | None = None, **kwargs: Any) -> None:
    """Emit a ``DEBUG`` record enriched with the bound context fields.

    Args:
        msg: Log message format string.
        *args: Positional arguments passed to ``Logger.debug``.
        extra: Per-call structured fields merged with ``_base_extra``.
        **kwargs: Additional keyword arguments forwarded to ``Logger.debug``.
    """
    self._logger.debug(msg, *args, extra=self._merge_extra(extra), **kwargs)
info(msg, *args, extra=None, **kwargs)

Emit an INFO record enriched with the bound context fields.

Parameters:

Name Type Description Default
msg str

Log message format string.

 required
*args Any

Positional arguments passed to Logger.info.

 ()
extra dict[str, Any] | None

Per-call structured fields merged with _base_extra.

 None
**kwargs Any

Additional keyword arguments forwarded to Logger.info.

 {}
Source code in src/codex_core/common/log_context.py 
128
129
130
131
132
133
134
135
136
137
def info(self, msg: str, *args: Any, extra: dict[str, Any] | None = None, **kwargs: Any) -> None:
    """Emit an ``INFO`` record enriched with the bound context fields.

    Args:
        msg: Log message format string.
        *args: Positional arguments passed to ``Logger.info``.
        extra: Per-call structured fields merged with ``_base_extra``.
        **kwargs: Additional keyword arguments forwarded to ``Logger.info``.
    """
    self._logger.info(msg, *args, extra=self._merge_extra(extra), **kwargs)
warning(msg, *args, extra=None, **kwargs)

Emit a WARNING record enriched with the bound context fields.

Parameters:

Name Type Description Default
msg str

Log message format string.

 required
*args Any

Positional arguments passed to Logger.warning.

 ()
extra dict[str, Any] | None

Per-call structured fields merged with _base_extra.

 None
**kwargs Any

Additional keyword arguments forwarded to Logger.warning.

 {}
Source code in src/codex_core/common/log_context.py 
139
140
141
142
143
144
145
146
147
148
def warning(self, msg: str, *args: Any, extra: dict[str, Any] | None = None, **kwargs: Any) -> None:
    """Emit a ``WARNING`` record enriched with the bound context fields.

    Args:
        msg: Log message format string.
        *args: Positional arguments passed to ``Logger.warning``.
        extra: Per-call structured fields merged with ``_base_extra``.
        **kwargs: Additional keyword arguments forwarded to ``Logger.warning``.
    """
    self._logger.warning(msg, *args, extra=self._merge_extra(extra), **kwargs)
error(msg, *args, extra=None, **kwargs)

Emit an ERROR record enriched with the bound context fields.

Parameters:

Name Type Description Default
msg str

Log message format string.

 required
*args Any

Positional arguments passed to Logger.error.

 ()
extra dict[str, Any] | None

Per-call structured fields merged with _base_extra.

 None
**kwargs Any

Additional keyword arguments forwarded to Logger.error.

 {}
Source code in src/codex_core/common/log_context.py 
150
151
152
153
154
155
156
157
158
159
def error(self, msg: str, *args: Any, extra: dict[str, Any] | None = None, **kwargs: Any) -> None:
    """Emit an ``ERROR`` record enriched with the bound context fields.

    Args:
        msg: Log message format string.
        *args: Positional arguments passed to ``Logger.error``.
        extra: Per-call structured fields merged with ``_base_extra``.
        **kwargs: Additional keyword arguments forwarded to ``Logger.error``.
    """
    self._logger.error(msg, *args, extra=self._merge_extra(extra), **kwargs)
exception(msg, *args, extra=None, **kwargs)

Emit an ERROR record with exception traceback and bound context.

Equivalent to :meth:error but always captures the current exception info (exc_info=True is implicit). Call inside an except block.

Parameters:

Name Type Description Default
msg str

Log message format string.

 required
*args Any

Positional arguments passed to Logger.exception.

 ()
extra dict[str, Any] | None

Per-call structured fields merged with _base_extra.

 None
**kwargs Any

Additional keyword arguments forwarded to Logger.exception.

 {}
Source code in src/codex_core/common/log_context.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def exception(self, msg: str, *args: Any, extra: dict[str, Any] | None = None, **kwargs: Any) -> None:
    """Emit an ``ERROR`` record with exception traceback and bound context.

    Equivalent to :meth:`error` but always captures the current
    exception info (``exc_info=True`` is implicit).  Call inside an
    ``except`` block.

    Args:
        msg: Log message format string.
        *args: Positional arguments passed to ``Logger.exception``.
        extra: Per-call structured fields merged with ``_base_extra``.
        **kwargs: Additional keyword arguments forwarded to ``Logger.exception``.
    """
    self._logger.exception(msg, *args, extra=self._merge_extra(extra), **kwargs)

Functions

set_log_context(**kwargs)

Merge kwargs into the current async-local log context.

Source code in src/codex_core/common/log_context.py 
45
46
47
48
49
def set_log_context(**kwargs: Any) -> None:
    """Merge *kwargs* into the current async-local log context."""
    current = _log_context.get().copy()
    current.update(kwargs)
    _log_context.set(current)

clear_log_context()

Reset the async-local log context to empty.

Source code in src/codex_core/common/log_context.py 
52
53
54
def clear_log_context() -> None:
    """Reset the async-local log context to empty."""
    _log_context.set({})

get_log_context()

Return a shallow copy of the current async-local log context.

Source code in src/codex_core/common/log_context.py 
57
58
59
def get_log_context() -> dict[str, Any]:
    """Return a shallow copy of the current async-local log context."""
    return _log_context.get().copy()

loguru_setup

Application-level Loguru configuration helpers (optional dependency).

Provides opinionated, zero-boilerplate Loguru setup for codex_tools applications. The codex_core library itself never calls these helpers; it uses the standard logging module exclusively so that consumers retain full control over their log infrastructure.

Dev mode (debug=True): three sinks — colourised stdout, rotating plain-text debug.log, and JSON errors.json.

Prod mode (debug=False): a single JSON-serialised stdout sink designed for ingestion by Grafana Alloy / Loki / ELK. File sinks are omitted because container runtimes capture stdout natively.

Both modes inject a patcher that enriches every log record with:

  • service — the service name passed to the setup function.
  • Any fields set via :func:~codex_core.common.log_context.set_log_context (request_id, char_id, correlation_id, etc.).

A built-in healthcheck filter suppresses records whose extra contains healthcheck=True, preventing /health endpoint noise from reaching log aggregators.

Standard-library logging records are bridged via :class:InterceptHandler so that third-party libraries (SQLAlchemy, httpx, aiogram, etc.) are automatically captured by Loguru.

Availability

loguru is an optional dependency. Both :func:setup_logging and :func:setup_universal_logging raise :exc:ImportError with an actionable message when loguru is not installed.

Classes

InterceptHandler

Bases: Handler

Bridge standard-library logging records to the Loguru sink.

Install this handler on the root logger (or any named logger) to forward all logging-based records into Loguru transparently. The handler resolves the correct call-stack depth so that Loguru reports the original call site rather than the handler frame.

This class is stateless and thread-safe; a single instance may be shared across all intercepted loggers.

Example 
import logging
from codex_core.common.loguru_setup import InterceptHandler

logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)
Source code in src/codex_core/common/loguru_setup.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 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
class InterceptHandler(logging.Handler):
    """Bridge standard-library ``logging`` records to the Loguru sink.

    Install this handler on the root logger (or any named logger) to
    forward all ``logging``-based records into Loguru transparently.
    The handler resolves the correct call-stack depth so that Loguru
    reports the *original* call site rather than the handler frame.

    This class is stateless and thread-safe; a single instance may be
    shared across all intercepted loggers.

    Example:
        ```python
        import logging
        from codex_core.common.loguru_setup import InterceptHandler

        logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)
        ```
    """

    def emit(self, record: logging.LogRecord) -> None:
        if logger is None:
            return

        level: str | int
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        frame: FrameType | None = logging.currentframe()
        depth = 6
        while frame and frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())

LoggingSettingsProtocol

Bases: Protocol

Structural protocol for the logging-related subset of settings.

Source code in src/codex_core/common/loguru_setup.py 
207
208
209
210
211
212
213
214
215
@runtime_checkable
class LoggingSettingsProtocol(Protocol):
    """Structural protocol for the logging-related subset of settings."""

    log_level_console: str
    log_level_file: str
    log_rotation: str
    log_dir: str
    debug: bool

Functions

setup_universal_logging(log_dir, service_name='App', console_level='INFO', file_level='DEBUG', rotation='10 MB', is_debug=False)

Configure Loguru with three sinks and standard-library interception.

Intended for applications that do not use :class:~codex_core.settings.BaseCommonSettings but still want the full codex_core logging stack. Callers supply raw configuration values directly rather than a settings object.

Parameters:

Name Type Description Default
log_dir Path

Directory where log files are created.

 required
service_name str

Label embedded in the console format string.

 'App'
console_level str

Minimum level for stdout output.

 'INFO'
file_level str

Minimum level for the debug log file.

 'DEBUG'
rotation str

Loguru rotation threshold string.

 '10 MB'
is_debug bool

When True, enables backtrace and diagnose.

 False

Raises:

Type Description
ImportError

If loguru is not installed.
Source code in src/codex_core/common/loguru_setup.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
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
def setup_universal_logging(
    log_dir: Path,
    service_name: str = "App",
    console_level: str = "INFO",
    file_level: str = "DEBUG",
    rotation: str = "10 MB",
    is_debug: bool = False,
) -> None:
    """Configure Loguru with three sinks and standard-library interception.

    Intended for applications that do not use
    :class:`~codex_core.settings.BaseCommonSettings` but still want the
    full codex_core logging stack.  Callers supply raw configuration
    values directly rather than a settings object.

    Args:
        log_dir: Directory where log files are created.
        service_name: Label embedded in the console format string.
        console_level: Minimum level for stdout output.
        file_level: Minimum level for the debug log file.
        rotation: Loguru rotation threshold string.
        is_debug: When ``True``, enables ``backtrace`` and ``diagnose``.

    Raises:
        ImportError: If ``loguru`` is not installed.
    """
    if logger is None:
        raise ImportError("loguru is not installed. Please install it manually: pip install loguru")

    logger.remove()
    log_dir.mkdir(parents=True, exist_ok=True)

    logger.configure(patcher=_make_context_patcher(service_name))

    logger.add(
        sink=sys.stdout,
        level=console_level,
        colorize=True,
        filter=_healthcheck_filter,
        format=_CONSOLE_FORMAT,
    )

    logger.add(
        sink=str(log_dir / "debug.log"),
        level=file_level,
        rotation=rotation,
        compression="zip",
        format=_FILE_FORMAT,
        enqueue=True,
        backtrace=is_debug,
        diagnose=is_debug,
    )

    logger.add(
        sink=str(log_dir / "errors.json"),
        level="ERROR",
        serialize=True,
        rotation=rotation,
        compression="zip",
        enqueue=True,
    )

    logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)

    logger.info("LoguruSetupComplete log_dir={}", log_dir)

setup_logging(settings, service_name, intercept_loggers=None, log_levels=None)

Configure Loguru from a settings object.

Dev (settings.debug is True): colourised console + file sinks. Prod (settings.debug is False): JSON-serialised stdout only.

Both modes apply a patcher (service name + contextvars) and a healthcheck filter.

Parameters:

Name Type Description Default
settings LoggingSettingsProtocol

Object satisfying :class:LoggingSettingsProtocol.

 required
service_name str

Identifies the service in log output.

 required
intercept_loggers list[str] | None

Logger names whose handlers are replaced with :class:InterceptHandler.

 None
log_levels dict[str, int] | None

{logger_name: level_int} for silencing noisy libraries.

 None

Raises:

Type Description
ImportError

If loguru is not installed.
Source code in src/codex_core/common/loguru_setup.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
def setup_logging(
    settings: LoggingSettingsProtocol,
    service_name: str,
    intercept_loggers: list[str] | None = None,
    log_levels: dict[str, int] | None = None,
) -> None:
    """Configure Loguru from a settings object.

    **Dev** (``settings.debug is True``): colourised console + file sinks.
    **Prod** (``settings.debug is False``): JSON-serialised stdout only.

    Both modes apply a patcher (service name + contextvars) and a
    healthcheck filter.

    Args:
        settings: Object satisfying :class:`LoggingSettingsProtocol`.
        service_name: Identifies the service in log output.
        intercept_loggers: Logger names whose handlers are replaced
            with :class:`InterceptHandler`.
        log_levels: ``{logger_name: level_int}`` for silencing noisy
            libraries.

    Raises:
        ImportError: If ``loguru`` is not installed.
    """
    if logger is None:
        raise ImportError("loguru is not installed. Please install it manually: pip install loguru")

    logger.remove()

    logger.configure(patcher=_make_context_patcher(service_name))

    if settings.debug:
        # ── Dev: colourised console + file sinks ──
        log_dir = Path(settings.log_dir) / service_name
        log_dir.mkdir(parents=True, exist_ok=True)

        logger.add(
            sink=sys.stdout,
            level=settings.log_level_console,
            colorize=True,
            filter=_healthcheck_filter,
            format=_CONSOLE_FORMAT,
        )

        logger.add(
            sink=str(log_dir / "debug.log"),
            level=settings.log_level_file,
            rotation=settings.log_rotation,
            compression="zip",
            format=_FILE_FORMAT,
            enqueue=True,
            backtrace=True,
            diagnose=True,
        )

        logger.add(
            sink=str(log_dir / "errors.json"),
            level="ERROR",
            serialize=True,
            rotation=settings.log_rotation,
            compression="zip",
            enqueue=True,
        )
    else:
        # ── Prod: JSON stdout only (Alloy / Loki / ELK ingestion) ──
        logger.add(
            sink=sys.stdout,
            level=settings.log_level_console,
            serialize=True,
            colorize=False,
            filter=_healthcheck_filter,
            enqueue=True,
        )

    # Intercept standard logging
    logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)

    if intercept_loggers:
        for name in intercept_loggers:
            logging.getLogger(name).handlers = [InterceptHandler()]

    if log_levels:
        for name, level in log_levels.items():
            logging.getLogger(name).setLevel(level)