Make linter happy

Author: immerrr

uvicorn/logging.py

@@ -121,7 +121,15 @@ def formatMessage(self, record: logging.LogRecord) -> str:
 
 
 class GunicornSafeAtoms(abc.Mapping):
-    """Implement atoms necessary for gunicorn log."""
+    """Implement atoms necessary for gunicorn log.
+
+    This class does a few things:
+    - provide all atoms necessary for gunicorn log formatter
+    - collect response body size for reporting from ASGI messages
+    - provide mapping interface that returns '-' for missing atoms
+    - escapes double quotes found in atom strings
+    """
+
     def __init__(self, scope):
         self.scope = scope
         self.status_code = None
@@ -134,25 +142,23 @@ def __init__(self, scope):
     def request_headers(self):
         if self._request_headers is None:
             self._request_headers = {
-                k.decode('ascii'): v.decode('ascii')
-                for k, v in self.scope['headers']
+                k.decode("ascii"): v.decode("ascii") for k, v in self.scope["headers"]
             }
         return self._request_headers
 
     @property
     def duration(self):
-        d = self.scope['response_end_time'] - self.scope['request_start_time']
+        d = self.scope["response_end_time"] - self.scope["request_start_time"]
         return d
 
     def on_asgi_message(self, message):
-        if message['type'] == 'http.response.start':
-            self.status_code = message['status']
+        if message["type"] == "http.response.start":
+            self.status_code = message["status"]
             self.response_headers = {
-                k.decode('ascii'): v.decode('ascii')
-                for k, v in message['headers']
+                k.decode("ascii"): v.decode("ascii") for k, v in message["headers"]
             }
-        elif message['type'] == 'http.response.body':
-            self.response_length += len(message.get('body', ''))
+        elif message["type"] == "http.response.body":
+            self.response_length += len(message.get("body", ""))
 
     def _request_header(self, key):
         return self.request_headers.get(key.lower())
@@ -168,20 +174,20 @@ def _wsgi_environ_variable(self, key):
     def __getitem__(self, key):
         if key in self.HANDLERS:
             retval = self.HANDLERS[key](self)
-        elif key.startswith('{'):
-            if key.endswith('}i'):
+        elif key.startswith("{"):
+            if key.endswith("}i"):
                 retval = self._request_header(key[1:-2])
-            elif key.endswith('}o'):
+            elif key.endswith("}o"):
                 retval = self._response_header(key[1:-2])
-            elif key.endswith('}e'):
+            elif key.endswith("}e"):
                 retval = self._wsgi_environ_variable(key[1:-2])
             else:
                 retval = None
         else:
             retval = None
 
         if retval is None:
-            return '-'
+            return "-"
         if isinstance(retval, str):
             return retval.replace('"', '\\"')
         return retval
@@ -192,103 +198,104 @@ def _register_handler(key, handlers=HANDLERS):
         def decorator(fn):
             handlers[key] = fn
             return fn
+
         return decorator
 
-    @_register_handler('h')
+    @_register_handler("h")
     def _remote_address(self, *args, **kwargs):
-        return self.scope['client'][0]
+        return self.scope["client"][0]
 
-    @_register_handler('l')
+    @_register_handler("l")
     def _dash(self, *args, **kwargs):
-        return '-'
+        return "-"
 
-    @_register_handler('u')
+    @_register_handler("u")
     def _user_name(self, *args, **kwargs):
         pass
 
-    @_register_handler('t')
+    @_register_handler("t")
     def date_of_the_request(self, *args, **kwargs):
         """Date and time in Apache Common Log Format"""
-        return time.strftime('[%d/%b/%Y:%H:%M:%S %z]')
+        return time.strftime("[%d/%b/%Y:%H:%M:%S %z]")
 
-    @_register_handler('r')
+    @_register_handler("r")
     def status_line(self, *args, **kwargs):
-        full_raw_path = (self.scope['raw_path'] + self.scope['query_string'])
-        full_path = full_raw_path.decode('ascii')
-        return '{method} {full_path} HTTP/{http_version}'.format(
+        full_raw_path = self.scope["raw_path"] + self.scope["query_string"]
+        full_path = full_raw_path.decode("ascii")
+        return "{method} {full_path} HTTP/{http_version}".format(
             full_path=full_path, **self.scope
         )
 
-    @_register_handler('m')
+    @_register_handler("m")
     def request_method(self, *args, **kwargs):
-        return self.scope['method']
+        return self.scope["method"]
 
-    @_register_handler('U')
+    @_register_handler("U")
     def url_path(self, *args, **kwargs):
-        return self.scope['raw_path'].decode('ascii')
+        return self.scope["raw_path"].decode("ascii")
 
-    @_register_handler('q')
+    @_register_handler("q")
     def query_string(self, *args, **kwargs):
-        return self.scope['query_string'].decode('ascii')
+        return self.scope["query_string"].decode("ascii")
 
-    @_register_handler('H')
+    @_register_handler("H")
     def protocol(self, *args, **kwargs):
-        return 'HTTP/%s' % self.scope['http_version']
+        return "HTTP/%s" % self.scope["http_version"]
 
-    @_register_handler('s')
+    @_register_handler("s")
     def status(self, *args, **kwargs):
-        return self.status_code or '-'
+        return self.status_code or "-"
 
-    @_register_handler('B')
+    @_register_handler("B")
     def response_length(self, *args, **kwargs):
         return self.response_length
 
-    @_register_handler('b')
+    @_register_handler("b")
     def response_length_or_dash(self, *args, **kwargs):
-        return self.response_length or '-'
+        return self.response_length or "-"
 
-    @_register_handler('f')
+    @_register_handler("f")
     def referer(self, *args, **kwargs):
-        val = self.request_headers.get(b'referer')
+        val = self.request_headers.get(b"referer")
         if val is None:
             return None
-        return val.decode('ascii')
+        return val.decode("ascii")
 
-    @_register_handler('a')
+    @_register_handler("a")
     def user_agent(self, *args, **kwargs):
-        val = self.request_headers.get(b'user-agent')
+        val = self.request_headers.get(b"user-agent")
         if val is None:
             return None
-        return val.decode('ascii')
+        return val.decode("ascii")
 
-    @_register_handler('T')
+    @_register_handler("T")
     def request_time_seconds(self, *args, **kwargs):
         return int(self.duration)
 
-    @_register_handler('D')
+    @_register_handler("D")
     def request_time_microseconds(self, *args, **kwargs):
         return int(self.duration * 1_000_000)
 
-    @_register_handler('L')
+    @_register_handler("L")
     def request_time_decimal_seconds(self, *args, **kwargs):
         return "%.6f" % self.duration
 
-    @_register_handler('p')
+    @_register_handler("p")
     def process_id(self, *args, **kwargs):
         return "<%s>" % getpid()
 
     def __iter__(self):
         # FIXME: add WSGI environ
         yield from self.HANDLERS
-        for k, _ in self.scope['headers']:
-            yield '{%s}i' % k.lower()
+        for k, _ in self.scope["headers"]:
+            yield "{%s}i" % k.lower()
         for k in self.response_headers:
-            yield '{%s}o' % k.lower()
+            yield "{%s}o" % k.lower()
 
     def __len__(self):
         # FIXME: add WSGI environ
         return (
             len(self.HANDLERS)
-            + len(self.scope['headers'] or ())
+            + len(self.scope["headers"] or ())
             + len(self.response_headers)
         )

uvicorn/protocols/http/h11_impl.py

@@ -490,8 +490,8 @@ async def send(self, message):
                 event = h11.EndOfMessage()
                 output = self.conn.send(event)
 
-                duration_extension = self.scope['extensions']['uvicorn_request_duration']
-                duration_extension["response_end_time"] = time.monotonic()
+                duration_scope = self.scope["extensions"]["uvicorn_request_duration"]
+                duration_scope["response_end_time"] = time.monotonic()
                 if self.gunicorn_log is not None:
                     try:
                         self.gunicorn_log.access_log.info(

uvicorn/protocols/http/httptools_impl.py

@@ -5,8 +5,6 @@
 import time
 import traceback
 import urllib
-from collections import abc
-from os import getpid
 from typing import Callable
 
 import httptools
@@ -528,8 +526,8 @@ async def send(self, message):
                 if self.expected_content_length != 0:
                     raise RuntimeError("Response content shorter than Content-Length")
                 self.response_complete = True
-                duration_extension = self.scope['extensions']['uvicorn_request_duration']
-                duration_extension['response_end_time'] = time.monotonic()
+                duration_scope = self.scope["extensions"]["uvicorn_request_duration"]
+                duration_scope["response_end_time"] = time.monotonic()
 
                 if self.gunicorn_log is not None:
                     try:

uvicorn/workers.py

@@ -38,7 +38,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
             "callback_notify": self.callback_notify,
             "limit_max_requests": self.max_requests,
             "forwarded_allow_ips": self.cfg.forwarded_allow_ips,
-            'gunicorn_log': self.log,
+            "gunicorn_log": self.log,
         }
 
         if self.cfg.is_ssl: