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

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())

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)

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()

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)

Logging

Helpers for structured logging and Loguru configuration.

log_context

Structured logging context bound to a named task or worker.

Provides :class:TaskLogContext — a thin, stateful wrapper around logging.Logger that automatically enriches every log record with a fixed set of structured fields (task name, worker name, arbitrary extras).

This module intentionally avoids any dependency on Loguru so that it works with any log sink: standard-library logging, Loguru (via :class:~codex_core.common.loguru_setup.InterceptHandler), JSON formatters, ELK, or Loki.

Example 
from codex_core.common.log_context import TaskLogContext

log = TaskLogContext("send_booking_notification", worker="notification_worker")
log.info("Processing appointment", extra={"appointment_id": 123})
# LogRecord contains: task="send_booking_notification",
#   worker="notification_worker", 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 
 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
 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
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 
83
84
85
86
87
88
89
90
91
92
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)
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 
116
117
118
119
120
121
122
123
124
125
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 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
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)
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
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 
105
106
107
108
109
110
111
112
113
114
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)

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.

Three configurable sinks are created by both setup functions:

  1. stdout — colourised, human-readable format for local development.
  2. debug.log — rotating plain-text file, enqueue=True for async-safe writing.
  3. errors.json — rotating JSON-serialised file capturing ERROR and above; suitable for ingestion by ELK / Loki pipelines.

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 
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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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:
        """Forward a single ``logging.LogRecord`` to the active Loguru logger.

        Resolves the Loguru level name from the record's level name,
        falls back to the integer level when the name is unknown, then
        walks up the call stack to find the frame that originally issued
        the log call — ensuring Loguru displays the correct source
        location instead of the handler internals.

        Args:
            record: The log record produced by the standard-library
                logging infrastructure.

        Note:
            If ``loguru`` is not installed (``logger is None``), this
            method returns silently to avoid masking the original
            ``ImportError``.
        """
        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())
Functions
emit(record)

Forward a single logging.LogRecord to the active Loguru logger.

Resolves the Loguru level name from the record's level name, falls back to the integer level when the name is unknown, then walks up the call stack to find the frame that originally issued the log call — ensuring Loguru displays the correct source location instead of the handler internals.

Parameters:

Name Type Description Default
record LogRecord

The log record produced by the standard-library logging infrastructure.

 required
Note

If loguru is not installed (logger is None), this method returns silently to avoid masking the original ImportError.

Source code in src/codex_core/common/loguru_setup.py 
61
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
def emit(self, record: logging.LogRecord) -> None:
    """Forward a single ``logging.LogRecord`` to the active Loguru logger.

    Resolves the Loguru level name from the record's level name,
    falls back to the integer level when the name is unknown, then
    walks up the call stack to find the frame that originally issued
    the log call — ensuring Loguru displays the correct source
    location instead of the handler internals.

    Args:
        record: The log record produced by the standard-library
            logging infrastructure.

    Note:
        If ``loguru`` is not installed (``logger is None``), this
        method returns silently to avoid masking the original
        ``ImportError``.
    """
    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 describing the logging-related subset of settings.

Any settings object that exposes these five attributes satisfies this protocol and can be passed to :func:setup_logging without explicit inheritance. :class:~codex_core.settings.BaseCommonSettings satisfies this protocol out of the box when its subclass adds the required logging fields.

Attributes:

Name Type Description
log_level_console str

Minimum Loguru level for stdout (e.g. "INFO").
log_level_file str

Minimum Loguru level for the rotating debug file (e.g. "DEBUG").
log_rotation str

Loguru rotation threshold (e.g. "10 MB" or "1 day").
log_dir str

Base directory for log files. A service-name subdirectory is appended automatically by :func:setup_logging.
debug bool

When True, enables backtrace and diagnose in the file sink.
Source code in src/codex_core/common/loguru_setup.py 
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
@runtime_checkable
class LoggingSettingsProtocol(Protocol):
    """Structural protocol describing the logging-related subset of settings.

    Any settings object that exposes these five attributes satisfies this
    protocol and can be passed to :func:`setup_logging` without explicit
    inheritance.  :class:`~codex_core.settings.BaseCommonSettings`
    satisfies this protocol out of the box when its subclass adds the
    required logging fields.

    Attributes:
        log_level_console: Minimum Loguru level for stdout
            (e.g. ``"INFO"``).
        log_level_file: Minimum Loguru level for the rotating debug
            file (e.g. ``"DEBUG"``).
        log_rotation: Loguru rotation threshold
            (e.g. ``"10 MB"`` or ``"1 day"``).
        log_dir: Base directory for log files.  A service-name
            subdirectory is appended automatically by
            :func:`setup_logging`.
        debug: When ``True``, enables ``backtrace`` and ``diagnose``
            in the file sink.
    """

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

Functions

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

Configure Loguru from a settings object conforming to :class:LoggingSettingsProtocol.

Preferred entry-point for applications that use :class:~codex_core.settings.BaseCommonSettings. Configuration is read from settings rather than raw arguments, enabling environment-driven log levels without code changes.

PII masking is not handled here; it is the responsibility of :class:~codex_core.core.base_dto.BaseDTO.__repr__ at the DTO level. Do not log raw user input through this logger.

Side effects
  • Removes all existing Loguru handlers (logger.remove()).
  • Creates <settings.log_dir>/<service_name>/ directory tree.
  • Replaces the root logging handler with :class:InterceptHandler.
  • Optionally replaces handlers on loggers listed in intercept_loggers.
  • Optionally sets log levels on loggers listed in log_levels.

Parameters:

Name Type Description Default
settings LoggingSettingsProtocol

Any object satisfying :class:LoggingSettingsProtocol. Typically a subclass of :class:~codex_core.settings.BaseCommonSettings.

 required
service_name str

Identifies the service in log output and is used as the leaf directory name under settings.log_dir.

 required
intercept_loggers list[str] | None

Optional list of logger names whose handlers should be replaced with :class:InterceptHandler (e.g. ["aiogram", "sqlalchemy.engine"]).

 None
log_levels dict[str, int] | None

Optional mapping of {logger_name: level_int} for silencing verbose third-party libraries (e.g. {"httpx": logging.WARNING}).

 None

Raises:

Type Description
ImportError

If loguru is not installed in the environment.
Example 
import logging
from codex_core.common.loguru_setup import setup_logging

setup_logging(
    settings=app_settings,
    service_name="api",
    intercept_loggers=["uvicorn", "sqlalchemy.engine"],
    log_levels={"httpx": logging.WARNING},
)
Source code in src/codex_core/common/loguru_setup.py 
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
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
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 conforming to :class:`LoggingSettingsProtocol`.

    Preferred entry-point for applications that use
    :class:`~codex_core.settings.BaseCommonSettings`.  Configuration
    is read from *settings* rather than raw arguments, enabling
    environment-driven log levels without code changes.

    PII masking is **not** handled here; it is the responsibility of
    :class:`~codex_core.core.base_dto.BaseDTO.__repr__` at the DTO
    level.  Do not log raw user input through this logger.

    Side effects:
        - Removes all existing Loguru handlers (``logger.remove()``).
        - Creates ``<settings.log_dir>/<service_name>/`` directory tree.
        - Replaces the root ``logging`` handler with
          :class:`InterceptHandler`.
        - Optionally replaces handlers on loggers listed in
          *intercept_loggers*.
        - Optionally sets log levels on loggers listed in *log_levels*.

    Args:
        settings: Any object satisfying :class:`LoggingSettingsProtocol`.
            Typically a subclass of
            :class:`~codex_core.settings.BaseCommonSettings`.
        service_name: Identifies the service in log output and is used
            as the leaf directory name under ``settings.log_dir``.
        intercept_loggers: Optional list of logger names whose handlers
            should be replaced with :class:`InterceptHandler`
            (e.g. ``["aiogram", "sqlalchemy.engine"]``).
        log_levels: Optional mapping of ``{logger_name: level_int}``
            for silencing verbose third-party libraries
            (e.g. ``{"httpx": logging.WARNING}``).

    Raises:
        ImportError: If ``loguru`` is not installed in the environment.

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

        setup_logging(
            settings=app_settings,
            service_name="api",
            intercept_loggers=["uvicorn", "sqlalchemy.engine"],
            log_levels={"httpx": logging.WARNING},
        )
        ```
    """
    if logger is None:
        raise ImportError(
            "loguru is not installed. Please install it manually: "
            "pip install loguru"
        )

    logger.remove()

    log_dir = Path(settings.log_dir) / service_name
    log_dir.mkdir(parents=True, exist_ok=True)

    # Console
    logger.add(
        sink=sys.stdout,
        level=settings.log_level_console,
        colorize=True,
        format=(
            "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
            "<level>{level: <8}</level> | "
            f"<magenta>{service_name}</magenta> | "
            "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
            "<level>{message}</level>"
        ),
    )

    # Debug file
    logger.add(
        sink=str(log_dir / "debug.log"),
        level=settings.log_level_file,
        rotation=settings.log_rotation,
        compression="zip",
        format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
        enqueue=True,
        backtrace=settings.debug,
        diagnose=settings.debug,
    )

    # Errors file (JSON)
    logger.add(
        sink=str(log_dir / "errors.json"),
        level="ERROR",
        serialize=True,
        rotation=settings.log_rotation,
        compression="zip",
        enqueue=True,
    )

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

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

    # Set levels for noisy libraries
    if log_levels:
        for name, level in log_levels.items():
            logging.getLogger(name).setLevel(level)

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.

Sink layout after the call:

  • stdout — colourised, console_level threshold.
  • /debug.log — rotating plain text, file_level threshold, backtrace / diagnose enabled when is_debug=True.
  • /errors.json — rotating JSON, ERROR threshold, async-enqueued.
  • Root logger — intercepted via :class:InterceptHandler.

Parameters:

Name Type Description Default
log_dir Path

Directory where log files are created. Created recursively if it does not exist.

 required
service_name str

Label embedded in the console format string to distinguish output from multiple co-running services.

 'App'
console_level str

Minimum level for stdout output (e.g. "INFO").

 'INFO'
file_level str

Minimum level for the debug log file (e.g. "DEBUG").

 'DEBUG'
rotation str

Loguru rotation threshold string (e.g. "10 MB" or "1 day").

 '10 MB'
is_debug bool

When True, enables full backtrace and diagnose output in the debug file sink.

 False

Raises:

Type Description
ImportError

If loguru is not installed in the environment.
Example 
from pathlib import Path
from codex_core.common.loguru_setup import setup_universal_logging

setup_universal_logging(
    log_dir=Path("/var/log/myapp"),
    service_name="booking-worker",
    is_debug=True,
)
Source code in src/codex_core/common/loguru_setup.py 
 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
187
188
189
190
191
192
193
194
195
196
197
198
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.

    Sink layout after the call:

    - **stdout** — colourised, ``console_level`` threshold.
    - **<log_dir>/debug.log** — rotating plain text, ``file_level``
      threshold, ``backtrace`` / ``diagnose`` enabled when
      ``is_debug=True``.
    - **<log_dir>/errors.json** — rotating JSON, ``ERROR`` threshold,
      async-enqueued.
    - **Root logger** — intercepted via :class:`InterceptHandler`.

    Args:
        log_dir: Directory where log files are created.  Created
            recursively if it does not exist.
        service_name: Label embedded in the console format string to
            distinguish output from multiple co-running services.
        console_level: Minimum level for stdout output (e.g. ``"INFO"``).
        file_level: Minimum level for the debug log file
            (e.g. ``"DEBUG"``).
        rotation: Loguru rotation threshold string (e.g. ``"10 MB"`` or
            ``"1 day"``).
        is_debug: When ``True``, enables full ``backtrace`` and
            ``diagnose`` output in the debug file sink.

    Raises:
        ImportError: If ``loguru`` is not installed in the environment.

    Example:
        ```python
        from pathlib import Path
        from codex_core.common.loguru_setup import setup_universal_logging

        setup_universal_logging(
            log_dir=Path("/var/log/myapp"),
            service_name="booking-worker",
            is_debug=True,
        )
        ```
    """
    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)

    # 1. Console output
    logger.add(
        sink=sys.stdout,
        level=console_level,
        colorize=True,
        format=(
            "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
            "<level>{level: <8}</level> | "
            f"<magenta>{service_name}</magenta> | "
            "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
            "<level>{message}</level>"
        ),
    )

    # 2. Debug file
    logger.add(
        sink=str(log_dir / "debug.log"),
        level=file_level,
        rotation=rotation,
        compression="zip",
        format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
        enqueue=True,
        backtrace=is_debug,
        diagnose=is_debug,
    )

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

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

    logger.info(f"Loguru setup complete for {service_name}. Logs: {log_dir}")