style: modernize python code using ruff

httoop/__init__.py

@@ -1,35 +1,69 @@
-# -*- coding: utf-8 -*-
-"""HTTPOOP - an OOP model of the HTTP protocol.
+"""
+HTTPOOP - an OOP model of the HTTP protocol.
 
 .. seealso:: :rfc:`2616`
 """
 
-from __future__ import absolute_import
 
 from httoop import cache
 from httoop.client import ClientStateMachine
 from httoop.date import Date
-from httoop.exceptions import (
-    DecodeError, EncodeError, InvalidBody, InvalidDate, InvalidHeader, InvalidLine, InvalidURI,
-)
+from httoop.exceptions import DecodeError, EncodeError, InvalidBody, InvalidDate, InvalidHeader, InvalidLine, InvalidURI
 from httoop.header import Headers
 from httoop.messages import Body, Method, Protocol, Request, Response
 from httoop.proxy import ProxyStateMachine
 from httoop.semantic import ComposedRequest, ComposedResponse
 from httoop.server import ServerStateMachine
 from httoop.status import (
-    ACCEPTED, BAD_GATEWAY, BAD_REQUEST, CONFLICT, CONTINUE, CREATED, EXPECTATION_FAILED, FORBIDDEN, FOUND,
-    GATEWAY_TIMEOUT, GONE, HTTP_VERSION_NOT_SUPPORTED, I_AM_A_TEAPOT, INTERNAL_SERVER_ERROR,
-    LENGTH_REQUIRED, METHOD_NOT_ALLOWED, MOVED_PERMANENTLY, MULTIPLE_CHOICES, NO_CONTENT,
-    NON_AUTHORITATIVE_INFORMATION, NOT_ACCEPTABLE, NOT_FOUND, NOT_IMPLEMENTED, NOT_MODIFIED, OK,
-    PARTIAL_CONTENT, PAYLOAD_TOO_LARGE, PAYMENT_REQUIRED, PRECONDITION_FAILED,
-    PROXY_AUTHENTICATION_REQUIRED, RANGE_NOT_SATISFIABLE, REQUEST_TIMEOUT, RESET_CONTENT, SEE_OTHER,
-    SERVICE_UNAVAILABLE, SWITCHING_PROTOCOLS, TEMPORARY_REDIRECT, UNAUTHORIZED, UNPROCESSABLE_ENTITY,
-    UNSUPPORTED_MEDIA_TYPE, URI_TOO_LONG, USE_PROXY, Status, StatusException,
+    ACCEPTED,
+    BAD_GATEWAY,
+    BAD_REQUEST,
+    CONFLICT,
+    CONTINUE,
+    CREATED,
+    EXPECTATION_FAILED,
+    FORBIDDEN,
+    FOUND,
+    GATEWAY_TIMEOUT,
+    GONE,
+    HTTP_VERSION_NOT_SUPPORTED,
+    I_AM_A_TEAPOT,
+    INTERNAL_SERVER_ERROR,
+    LENGTH_REQUIRED,
+    METHOD_NOT_ALLOWED,
+    MOVED_PERMANENTLY,
+    MULTIPLE_CHOICES,
+    NO_CONTENT,
+    NON_AUTHORITATIVE_INFORMATION,
+    NOT_ACCEPTABLE,
+    NOT_FOUND,
+    NOT_IMPLEMENTED,
+    NOT_MODIFIED,
+    OK,
+    PARTIAL_CONTENT,
+    PAYLOAD_TOO_LARGE,
+    PAYMENT_REQUIRED,
+    PRECONDITION_FAILED,
+    PROXY_AUTHENTICATION_REQUIRED,
+    RANGE_NOT_SATISFIABLE,
+    REQUEST_TIMEOUT,
+    RESET_CONTENT,
+    SEE_OTHER,
+    SERVICE_UNAVAILABLE,
+    SWITCHING_PROTOCOLS,
+    TEMPORARY_REDIRECT,
+    UNAUTHORIZED,
+    UNPROCESSABLE_ENTITY,
+    UNSUPPORTED_MEDIA_TYPE,
+    URI_TOO_LONG,
+    USE_PROXY,
+    Status,
+    StatusException,
 )
 from httoop.uri import URI
 from httoop.version import ServerHeader, ServerProtocol, UserAgentHeader, __version__
 
+
 __all__ = [
     'Status', 'Body', 'Headers', 'URI', 'Method',
     'Request', 'Response', 'Protocol', 'Date', 'ServerStateMachine', 'ClientStateMachine',

httoop/__main__.py

@@ -1,13 +1,12 @@
-# -*- coding: utf-8 -*-
-"""httoop CLI tool.
+"""
+httoop CLI tool.
 
 Examples
 --------
 python3 -m httoop compose request -H 'Host: www.example.net'  | python3 -m httoop parse request
 python3 -m httoop compose response  | python3 -m httoop parse response
 """
 
-from __future__ import print_function
 
 import sys
 from argparse import ArgumentParser, FileType
@@ -17,52 +16,52 @@
 from httoop.server import ServerStateMachine
 
 
-class CLI(object):
+class CLI:
     """httoop CLI tool."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         self.message = None
         self.parser = ArgumentParser(name, description=self.__doc__, epilog='https://github.com/spaceone/httoop/')
         self.parent_parser = ArgumentParser(add_help=False)
         self.parser.add_argument('-v', '--version', action='version', version='%%(prog)s %s' % (version,))
         self.add_subparsers()
         self.parse_arguments()
 
-    def add_subparsers(self):
-        action_subparsers = self.parser.add_subparsers(title="action", dest="action", required=True)
-        parse_parser = action_subparsers.add_parser("parse", parents=[self.parent_parser])
-        compose_parser = action_subparsers.add_parser("compose", parents=[self.parent_parser])
+    def add_subparsers(self) -> None:
+        action_subparsers = self.parser.add_subparsers(title='action', dest='action', required=True)
+        parse_parser = action_subparsers.add_parser('parse', parents=[self.parent_parser])
+        compose_parser = action_subparsers.add_parser('compose', parents=[self.parent_parser])
 
-        compose_message_subparsers = compose_parser.add_subparsers(title="message", dest="message")
-        request = compose_message_subparsers.add_parser("request", parents=[self.parent_parser])
+        compose_message_subparsers = compose_parser.add_subparsers(title='message', dest='message')
+        request = compose_message_subparsers.add_parser('request', parents=[self.parent_parser])
         request.set_defaults(func=self.compose_request)
         add = request.add_argument
         add('-m', '--method')
         add('-u', '--uri')
         self.add_common_arguments(add)
 
-        response = compose_message_subparsers.add_parser("response", parents=[self.parent_parser])
+        response = compose_message_subparsers.add_parser('response', parents=[self.parent_parser])
         response.set_defaults(func=self.compose_response)
         add = response.add_argument
         add('-s', '--status')
         add('--reason')
         self.add_common_arguments(add)
 
-        parse_message_subparsers = parse_parser.add_subparsers(title="message", dest="message")
-        request = parse_message_subparsers.add_parser("request", parents=[self.parent_parser])
+        parse_message_subparsers = parse_parser.add_subparsers(title='message', dest='message')
+        request = parse_message_subparsers.add_parser('request', parents=[self.parent_parser])
         request.set_defaults(func=self.parse_request)
         add = request.add_argument
         add('--file', default='-', type=FileType('rb'))
         add('--scheme', default='http')
         add('--host', default='www.example.net')
         add('--port', default=80, type=int)
 
-        response = parse_message_subparsers.add_parser("response", parents=[self.parent_parser])
+        response = parse_message_subparsers.add_parser('response', parents=[self.parent_parser])
         add = response.add_argument
         response.set_defaults(func=self.parse_response)
         add('--file', default='-', type=FileType('rb'))
 
-    def parse_arguments(self):
+    def parse_arguments(self) -> None:
         self.arguments = self.parser.parse_args()
 
         if self.arguments.action == 'parse' and hasattr(self.arguments.file, 'buffer'):
@@ -75,14 +74,14 @@ def add_common_arguments(self, add) -> None:
         add('-H', '--header', action='append', default=[])
         add('-b', '--body', default='')
 
-    def parse_request(self):
+    def parse_request(self) -> None:
         server = ServerStateMachine(self.arguments.scheme, self.arguments.host, self.arguments.port)
-        for request, response in server.parse(self.arguments.file.read()):
+        for _request, response in server.parse(self.arguments.file.read()):
             print(repr(response))
             print(repr(response.headers))
             print(repr(response.body))
 
-    def parse_response(self):
+    def parse_response(self) -> None:
         client = ClientStateMachine()
         client.request = Request()
         for response in client.parse(self.arguments.file.read()):
@@ -97,15 +96,15 @@ def parse_response(self):
             print(repr(client.message.body))
             print(repr(client.buffer))
 
-    def compose_request(self):
+    def compose_request(self) -> None:
         self.message = Request()
         if self.arguments.method:
             self.message.method = self.arguments.method
         if self.arguments.uri:
             self.message.uri = self.arguments.uri
         self.common()
 
-    def compose_response(self):
+    def compose_response(self) -> None:
         self.message = Response()
         status = self.message.status.code
         if self.arguments.status:
@@ -115,7 +114,7 @@ def compose_response(self):
         self.message.status = status
         self.common()
 
-    def common(self):
+    def common(self) -> None:
         if self.arguments.protocol:
             protocol = self.arguments.protocol
             try:

httoop/authentication/__init__.py

@@ -1,7 +1,7 @@
-# -*- coding: utf-8 -*-
 from __future__ import annotations
+
 import re
-from typing import Any, Dict, List, Tuple, Union
+from typing import TYPE_CHECKING, Any
 
 from httoop.authentication.basic import BasicAuthRequestScheme, BasicAuthResponseScheme
 from httoop.authentication.digest import DigestAuthRequestScheme, DigestAuthResponseScheme
@@ -10,6 +10,10 @@
 from httoop.util import _
 
 
+if TYPE_CHECKING:
+    from httoop.header.auth import WWWAuthenticate
+
+
 class AuthElement(HeaderElement):
 
     schemes = {}
@@ -23,38 +27,38 @@ def sanitize(self) -> None:
                 self.params[key] = type(value)(x.encode('UTF-8') if not isinstance(x, bytes) and isinstance(x, str) else x for x in value)
 
     @classmethod
-    def parseparams(cls, elementstr: bytes) -> Union[Tuple[bytes, Dict[bytes, str]], Tuple[bytes, Dict[str, bytes]], Tuple[bytes, Dict[str, Union[bytes, List[bytes], bool]]], Tuple[bytes, Dict[str, Union[bytes, List[bytes]]]], Tuple[bytes, Dict[bytes, Union[str, bytes]]]]:
+    def parseparams(cls, elementstr: bytes) -> tuple[bytes, dict[bytes, str]] | tuple[bytes, dict[str, bytes]] | tuple[bytes, dict[str, bytes | list[bytes] | bool]] | tuple[bytes, dict[str, bytes | list[bytes]]] | tuple[bytes, dict[bytes, str | bytes]]:
         try:
             scheme, authinfo = elementstr.split(b' ', 1)
         except ValueError:
-            raise InvalidHeader(_(u'Authorization headers must contain authentication scheme'))
+            raise InvalidHeader(_('Authorization headers must contain authentication scheme'))
         try:
             parser = cls.schemes[scheme.decode('ISO8859-1').lower()]
         except KeyError:
-            raise InvalidHeader(_(u'Unsupported authentication scheme: %r'), scheme)
+            raise InvalidHeader(_('Unsupported authentication scheme: %r'), scheme)
 
         try:
             authinfo = parser.parse(authinfo)
         except KeyError as key:
-            raise InvalidHeader(_(u'Missing parameter %r for authentication scheme %r'), str(key), scheme)
+            raise InvalidHeader(_('Missing parameter %r for authentication scheme %r'), str(key), scheme)
 
         return scheme.title(), authinfo
 
     def compose(self) -> bytes:
         try:
             scheme = self.schemes[self.value.lower()]
         except KeyError:
-            raise InvalidHeader(_(u'Unsupported authentication scheme: %r'), self.value)
+            raise InvalidHeader(_('Unsupported authentication scheme: %r'), self.value)
 
         try:
             authinfo = scheme.compose(self.params)
         except KeyError as key:
-            raise InvalidHeader(_(u'Missing parameter %r for authentication scheme %r'), str(key), self.value)
+            raise InvalidHeader(_('Missing parameter %r for authentication scheme %r'), str(key), self.value)
 
         return b'%s %s' % (self.value.encode('ASCII').title(), authinfo)
 
     @classmethod
-    def split(cls, value: bytes) -> List[Union[bytes, Any]]:
+    def split(cls, value: bytes) -> list[bytes | Any]:
         value = cls.RE_SPACE_SPLIT.split(value)
         indexes = [i for i, val in enumerate(value) if val != b',' and b'=' not in val]
         return [b' '.join(value[a:b]) for a, b in zip(indexes, indexes[1:] + [None])]
@@ -78,7 +82,7 @@ def username(self):
         return self.params.get('username').decode(self.encoding)
 
     @username.setter
-    def username(self, username):
+    def username(self, username) -> None:
         self.params['username'] = username.encode(self.encoding)
 
     @property
@@ -87,7 +91,7 @@ def password(self):
             return self.params.get('password').decode(self.encoding)
 
     @password.setter
-    def password(self, password):
+    def password(self, password) -> None:
         if self.scheme == 'basic':
             self.params['password'] = password.encode(self.encoding)
 
@@ -100,20 +104,20 @@ class AuthResponseElement(AuthElement):
     }
 
     @classmethod
-    def sorted(cls, elements: List[Union["WWWAuthenticate", Any]]) -> List[Union["WWWAuthenticate", Any]]:
-        return list(sorted(elements, key=lambda e: {'basic': u'\xff'}.get(e.value.lower(), e.value)))
+    def sorted(cls, elements: list[WWWAuthenticate | Any]) -> list[WWWAuthenticate | Any]:
+        return sorted(elements, key=lambda e: {'basic': '\xff'}.get(e.value.lower(), e.value))
 
     @classmethod
-    def join(cls, values: List[bytes]) -> bytes:
+    def join(cls, values: list[bytes]) -> bytes:
         return b' '.join(values)
 
     @property
     def realm(self):
         return self.params[b'realm'].decode('ASCII')
 
     @realm.setter
-    def realm(self, realm):
-        self.params['realm'] = realm.replace(u'"', u'').encode('ASCII', 'ignore')
+    def realm(self, realm) -> None:
+        self.params['realm'] = realm.replace('"', '').encode('ASCII', 'ignore')
 
 
 class AuthInfoElement(HeaderElement):

httoop/authentication/basic.py

@@ -1,33 +1,32 @@
-# -*- coding: utf-8 -*-
-
 from __future__ import annotations
+
+import base64
 from binascii import Error as Base64Error
-from typing import Dict, Union
 
 from httoop.exceptions import InvalidHeader
 from httoop.header.element import HeaderElement
-from httoop.util import ByteUnicodeDict, _, decode_base64, encode_base64
+from httoop.util import ByteUnicodeDict, _
 
 
-class BasicAuthRequestScheme(object):
+class BasicAuthRequestScheme:
 
     @staticmethod
-    def parse(authinfo: bytes) -> Dict[str, bytes]:
-        #try:
+    def parse(authinfo: bytes) -> dict[str, bytes]:
+        # try:
         #    authinfo = authinfo.encode('ascii')
-        #except ValueError:
+        # except ValueError:
         #    raise InvalidHeader(_(u'Invalid base64 in basic authentication'))
 
         try:
-            username, password = decode_base64(authinfo.strip()).split(b':')
+            username, password = base64.b64decode(authinfo.strip()).split(b':')
         except Base64Error:
-            raise InvalidHeader(_(u'Basic authentication contains invalid base64'))
+            raise InvalidHeader(_('Basic authentication contains invalid base64'))
         except ValueError:
-            raise InvalidHeader(_(u'No username:password provided'))
+            raise InvalidHeader(_('No username:password provided'))
 
         authinfo = {
-            #'username': username.decode('ISO8859-1'),
-            #'password': password.decode('ISO8859-1')
+            # 'username': username.decode('ISO8859-1'),
+            # 'password': password.decode('ISO8859-1')
             'username': username,
             'password': password,
         }
@@ -37,15 +36,15 @@ def parse(authinfo: bytes) -> Dict[str, bytes]:
     def compose(authinfo: ByteUnicodeDict) -> bytes:
         username = authinfo['username']
         password = authinfo['password']
-        #username = username.encode('ISO8859-1')
-        #password = password.encode('ISO8859-1')
-        return encode_base64(b'%s:%s' % (username, password)).strip()
+        # username = username.encode('ISO8859-1')
+        # password = password.encode('ISO8859-1')
+        return base64.b64encode(b'%s:%s' % (username, password))
 
 
-class BasicAuthResponseScheme(object):
+class BasicAuthResponseScheme:
 
     @staticmethod
-    def parse(authinfo: bytes) -> Dict[bytes, Union[str, bytes]]:
+    def parse(authinfo: bytes) -> dict[bytes, str | bytes]:
         params = HeaderElement.parseparams(b'X;%s' % authinfo)[1]
         params.setdefault(b'realm', b'')
         return params