1from __future__ import annotations
2
3import logging
4import sys
5from collections.abc import Generator, Mapping
6from contextlib import contextmanager
7from typing import Any
8
9from .debug import DebugMode
10
11
12class PlainLogger(logging.Logger):
13 """Enhanced logger that supports structured output and context management."""
14
15 def __init__(self, name: str):
16 super().__init__(name)
17 self.context: dict[str, Any] = {} # Public, mutable context dict
18 self.debug_mode = DebugMode(self)
19
20 @contextmanager
21 def include_context(self, **kwargs: Any) -> Generator[None]:
22 """Context manager for temporary context."""
23 # Store original context
24 original_context = self.context.copy()
25
26 # Add temporary context
27 self.context.update(kwargs)
28
29 try:
30 yield
31 finally:
32 # Restore original context
33 self.context = original_context
34
35 def force_debug(self) -> DebugMode:
36 """Return context manager for temporarily enabling DEBUG level logging."""
37 return self.debug_mode
38
39 # Override logging methods to add context parameter for IDE support
40 def debug(
41 self,
42 msg: object,
43 *args: object,
44 exc_info: Any = None,
45 stack_info: bool = False,
46 stacklevel: int = 1,
47 extra: Mapping[str, object] | None = None,
48 context: dict[str, Any] | None = None,
49 ) -> None:
50 if self.isEnabledFor(logging.DEBUG):
51 self._log(
52 logging.DEBUG,
53 msg,
54 args,
55 exc_info=exc_info,
56 extra=extra,
57 stack_info=stack_info,
58 stacklevel=stacklevel,
59 context=context,
60 )
61
62 def info(
63 self,
64 msg: object,
65 *args: object,
66 exc_info: Any = None,
67 stack_info: bool = False,
68 stacklevel: int = 1,
69 extra: Mapping[str, object] | None = None,
70 context: dict[str, Any] | None = None,
71 ) -> None:
72 if self.isEnabledFor(logging.INFO):
73 self._log(
74 logging.INFO,
75 msg,
76 args,
77 exc_info=exc_info,
78 extra=extra,
79 stack_info=stack_info,
80 stacklevel=stacklevel,
81 context=context,
82 )
83
84 def warning(
85 self,
86 msg: object,
87 *args: object,
88 exc_info: Any = None,
89 stack_info: bool = False,
90 stacklevel: int = 1,
91 extra: Mapping[str, object] | None = None,
92 context: dict[str, Any] | None = None,
93 ) -> None:
94 if self.isEnabledFor(logging.WARNING):
95 self._log(
96 logging.WARNING,
97 msg,
98 args,
99 exc_info=exc_info,
100 extra=extra,
101 stack_info=stack_info,
102 stacklevel=stacklevel,
103 context=context,
104 )
105
106 def error(
107 self,
108 msg: object,
109 *args: object,
110 exc_info: Any = None,
111 stack_info: bool = False,
112 stacklevel: int = 1,
113 extra: Mapping[str, object] | None = None,
114 context: dict[str, Any] | None = None,
115 ) -> None:
116 if self.isEnabledFor(logging.ERROR):
117 self._log(
118 logging.ERROR,
119 msg,
120 args,
121 exc_info=exc_info,
122 extra=extra,
123 stack_info=stack_info,
124 stacklevel=stacklevel,
125 context=context,
126 )
127
128 def critical(
129 self,
130 msg: object,
131 *args: object,
132 exc_info: Any = None,
133 stack_info: bool = False,
134 stacklevel: int = 1,
135 extra: Mapping[str, object] | None = None,
136 context: dict[str, Any] | None = None,
137 ) -> None:
138 if self.isEnabledFor(logging.CRITICAL):
139 self._log(
140 logging.CRITICAL,
141 msg,
142 args,
143 exc_info=exc_info,
144 extra=extra,
145 stack_info=stack_info,
146 stacklevel=stacklevel,
147 context=context,
148 )
149
150 def exception(
151 self,
152 msg: object,
153 *args: object,
154 exc_info: Any = True,
155 stack_info: bool = False,
156 stacklevel: int = 1,
157 extra: Mapping[str, object] | None = None,
158 context: dict[str, Any] | None = None,
159 ) -> None:
160 self.error(
161 msg,
162 *args,
163 exc_info=exc_info,
164 stack_info=stack_info,
165 stacklevel=stacklevel,
166 extra=extra,
167 context=context,
168 )
169
170 def _log(
171 self,
172 level: int,
173 msg: object,
174 args: tuple[object, ...] | Mapping[str, object],
175 exc_info: Any = None,
176 extra: Mapping[str, object] | None = None,
177 stack_info: bool = False,
178 stacklevel: int = 1,
179 context: dict[str, Any] | None = None,
180 ) -> None:
181 """Low-level logging routine which creates a LogRecord and then calls all handlers."""
182
183 # Merge into one dict: persistent context < extra < per-call context.
184 # All keys end up as top-level attributes on the LogRecord.
185 merged_extra: dict[str, object] = {}
186 if self.context:
187 merged_extra.update(self.context)
188 if extra:
189 merged_extra.update(extra)
190 if context:
191 merged_extra.update(context)
192
193 super()._log(
194 level=level,
195 msg=msg,
196 args=args,
197 exc_info=exc_info,
198 extra=merged_extra or None,
199 stack_info=stack_info,
200 stacklevel=stacklevel,
201 )
202
203
204def get_framework_logger(name: str = "") -> logging.Logger:
205 """Get a logger for framework code with auto-derived naming.
206
207 With no arguments, derives the name from the caller's module:
208 plain.postgres.connection → plain.postgres
209 plain.server.workers.entry → plain.server
210
211 With an explicit name, uses it directly:
212 get_framework_logger("plain.server.access")
213 """
214 if not name:
215 caller = sys._getframe(1).f_globals["__name__"]
216 parts = caller.split(".")
217 name = ".".join(parts[:2])
218
219 return logging.getLogger(name)
220
221
222# Create the default app logger instance
223app_logger = PlainLogger("app")