codekingpro/portable-devtools
115k
1import datetime
2import functools
3import logging
4import os
5import re
6import time as time_mod
7from collections import namedtuple
8from collections.abc import Iterable
9from typing import Callable, ClassVar, Dict, List, Optional, Tuple
10
11from .abc import AbstractAccessLogger
12from .web_request import BaseRequest
13from .web_response import StreamResponse
14
15KeyMethod = namedtuple("KeyMethod", "key method")
16
17
18class AccessLogger(AbstractAccessLogger):
19 """Helper object to log access.
20
21 Usage:
22 log = logging.getLogger("spam")
23 log_format = "%a %{User-Agent}i"
24 access_logger = AccessLogger(log, log_format)
25 access_logger.log(request, response, time)
26
27 Format:
28 %% The percent sign
29 %a Remote IP-address (IP-address of proxy if using reverse proxy)
30 %t Time when the request was started to process
31 %P The process ID of the child that serviced the request
32 %r First line of request
33 %s Response status code
34 %b Size of response in bytes, including HTTP headers
35 %T Time taken to serve the request, in seconds
36 %Tf Time taken to serve the request, in seconds with floating fraction
37 in .06f format
38 %D Time taken to serve the request, in microseconds
39 %{FOO}i request.headers['FOO']
40 %{FOO}o response.headers['FOO']
41 %{FOO}e os.environ['FOO']
42
43 """
44
45 LOG_FORMAT_MAP = {
46 "a": "remote_address",
47 "t": "request_start_time",
48 "P": "process_id",
49 "r": "first_request_line",
50 "s": "response_status",
51 "b": "response_size",
52 "T": "request_time",
53 "Tf": "request_time_frac",
54 "D": "request_time_micro",
55 "i": "request_header",
56 "o": "response_header",
57 }
58
59 LOG_FORMAT = '%a %t "%r" %s %b "%{Referer}i" "%{User-Agent}i"'
60 FORMAT_RE = re.compile(r"%(\{([A-Za-z0-9\-_]+)\}([ioe])|[atPrsbOD]|Tf?)")
61 CLEANUP_RE = re.compile(r"(%[^s])")
62 _FORMAT_CACHE: Dict[str, Tuple[str, List[KeyMethod]]] = {}
63
64 _cached_tz: ClassVar[Optional[datetime.timezone]] = None
65 _cached_tz_expires: ClassVar[float] = 0.0
66
67 def __init__(self, logger: logging.Logger, log_format: str = LOG_FORMAT) -> None:
68 """Initialise the logger.
69
70 logger is a logger object to be used for logging.
71 log_format is a string with apache compatible log format description.
72
73 """
74 super().__init__(logger, log_format=log_format)
75
76 _compiled_format = AccessLogger._FORMAT_CACHE.get(log_format)
77 if not _compiled_format:
78 _compiled_format = self.compile_format(log_format)
79 AccessLogger._FORMAT_CACHE[log_format] = _compiled_format
80
81 self._log_format, self._methods = _compiled_format
82
83 def compile_format(self, log_format: str) -> Tuple[str, List[KeyMethod]]:
84 """Translate log_format into form usable by modulo formatting
85
86 All known atoms will be replaced with %s
87 Also methods for formatting of those atoms will be added to
88 _methods in appropriate order
89
90 For example we have log_format = "%a %t"
91 This format will be translated to "%s %s"
92 Also contents of _methods will be
93 [self._format_a, self._format_t]
94 These method will be called and results will be passed
95 to translated string format.
96
97 Each _format_* method receive 'args' which is list of arguments
98 given to self.log
99
100 Exceptions are _format_e, _format_i and _format_o methods which
101 also receive key name (by functools.partial)
102
103 """
104 # list of (key, method) tuples, we don't use an OrderedDict as users
105 # can repeat the same key more than once
106 methods = list()
107
108 for atom in self.FORMAT_RE.findall(log_format):
109 if atom[1] == "":
110 format_key1 = self.LOG_FORMAT_MAP[atom[0]]
111 m = getattr(AccessLogger, "_format_%s" % atom[0])
112 key_method = KeyMethod(format_key1, m)
113 else:
114 format_key2 = (self.LOG_FORMAT_MAP[atom[2]], atom[1])
115 m = getattr(AccessLogger, "_format_%s" % atom[2])
116 key_method = KeyMethod(format_key2, functools.partial(m, atom[1]))
117
118 methods.append(key_method)
119
120 log_format = self.FORMAT_RE.sub(r"%s", log_format)
121 log_format = self.CLEANUP_RE.sub(r"%\1", log_format)
122 return log_format, methods
123
124 @staticmethod
125 def _format_i(
126 key: str, request: BaseRequest, response: StreamResponse, time: float
127 ) -> str:
128 if request is None:
129 return "(no headers)"
130
131 # suboptimal, make istr(key) once
132 return request.headers.get(key, "-")
133
134 @staticmethod
135 def _format_o(
136 key: str, request: BaseRequest, response: StreamResponse, time: float
137 ) -> str:
138 # suboptimal, make istr(key) once
139 return response.headers.get(key, "-")
140
141 @staticmethod
142 def _format_a(request: BaseRequest, response: StreamResponse, time: float) -> str:
143 if request is None:
144 return "-"
145 ip = request.remote
146 return ip if ip is not None else "-"
147
148 @classmethod
149 def _get_local_time(cls) -> datetime.datetime:
150 if cls._cached_tz is None or time_mod.time() >= cls._cached_tz_expires:
151 gmtoff = time_mod.localtime().tm_gmtoff
152 cls._cached_tz = tz = datetime.timezone(datetime.timedelta(seconds=gmtoff))
153
154 now = datetime.datetime.now(tz)
155 # Expire at every 30 mins, as any DST change should occur at 0/30 mins past.
156 d = now + datetime.timedelta(minutes=30)
157 d = d.replace(minute=30 if d.minute >= 30 else 0, second=0, microsecond=0)
158 cls._cached_tz_expires = d.timestamp()
159 return now
160
161 return datetime.datetime.now(cls._cached_tz)
162
163 @staticmethod
164 def _format_t(request: BaseRequest, response: StreamResponse, time: float) -> str:
165 now = AccessLogger._get_local_time()
166 start_time = now - datetime.timedelta(seconds=time)
167 return start_time.strftime("[%d/%b/%Y:%H:%M:%S %z]")
168
169 @staticmethod
170 def _format_P(request: BaseRequest, response: StreamResponse, time: float) -> str:
171 return "<%s>" % os.getpid()
172
173 @staticmethod
174 def _format_r(request: BaseRequest, response: StreamResponse, time: float) -> str:
175 if request is None:
176 return "-"
177 return "{} {} HTTP/{}.{}".format(
178 request.method,
179 request.path_qs,
180 request.version.major,
181 request.version.minor,
182 )
183
184 @staticmethod
185 def _format_s(request: BaseRequest, response: StreamResponse, time: float) -> int:
186 return response.status
187
188 @staticmethod
189 def _format_b(request: BaseRequest, response: StreamResponse, time: float) -> int:
190 return response.body_length
191
192 @staticmethod
193 def _format_T(request: BaseRequest, response: StreamResponse, time: float) -> str:
194 return str(round(time))
195
196 @staticmethod
197 def _format_Tf(request: BaseRequest, response: StreamResponse, time: float) -> str:
198 return "%06f" % time
199
200 @staticmethod
201 def _format_D(request: BaseRequest, response: StreamResponse, time: float) -> str:
202 return str(round(time * 1000000))
203
204 def _format_line(
205 self, request: BaseRequest, response: StreamResponse, time: float
206 ) -> Iterable[Tuple[str, Callable[[BaseRequest, StreamResponse, float], str]]]:
207 return [(key, method(request, response, time)) for key, method in self._methods]
208
209 @property
210 def enabled(self) -> bool:
211 """Check if logger is enabled."""
212 # Avoid formatting the log line if it will not be emitted.
213 return self.logger.isEnabledFor(logging.INFO)
214
215 def log(self, request: BaseRequest, response: StreamResponse, time: float) -> None:
216 try:
217 fmt_info = self._format_line(request, response, time)
218
219 values = list()
220 extra = dict()
221 for key, value in fmt_info:
222 values.append(value)
223
224 if key.__class__ is str:
225 extra[key] = value
226 else:
227 k1, k2 = key # type: ignore[misc]
228 dct = extra.get(k1, {}) # type: ignore[var-annotated,has-type]
229 dct[k2] = value # type: ignore[index,has-type]
230 extra[k1] = dct # type: ignore[has-type,assignment]
231
232 self.logger.info(self._log_format % tuple(values), extra=extra)
233 except Exception:
234 self.logger.exception("Error in logging")
235