refactor: improve type hints for mypy

Author: rexzhang

asgi_webdav/auth.py

@@ -3,7 +3,6 @@
 import hashlib
 import re
 from base64 import b64decode
-from enum import Enum
 from logging import getLogger
 from typing import Any
 from urllib.parse import parse_qs
@@ -16,7 +15,7 @@
     DAVCacheType,
 )
 from asgi_webdav.config import Config
-from asgi_webdav.constants import DAVMethod, DAVUser
+from asgi_webdav.constants import DAVMethod, DAVUpperEnumAbc, DAVUser
 from asgi_webdav.exception import DAVExceptionAuthFailed, DAVExceptionConfig
 from asgi_webdav.request import DAVRequest
 from asgi_webdav.response import DAVResponse
@@ -58,32 +57,22 @@ def _md5(data: str) -> str:
     return hashlib.new("md5", data.encode("utf-8")).hexdigest()
 
 
-class DAVPasswordType(Enum):
-    INVALID = ":", 0
+class DAVPasswordType(DAVUpperEnumAbc):
+    INVALID = "X", -1
+
     RAW = ":", 0
     HASHLIB = ":", 4
     DIGEST = ":", 3
     LDAP = "#", 5
 
-    def __init__(self, *args, **kwds):
-        self._value_ = self._name_
-        self.split_char = args[0]
-        self.split_count = args[1]
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
 
-    @classmethod
-    def _missing_(cls, value):
-        if isinstance(value, str):
-            value = value.upper()
-        else:
-            return cls.INVALID
-
-        try:
-            return cls[value]
+        self.split_char, self.split_count = args
 
-        except KeyError:
-            pass
-
-        return cls.INVALID
+    @classmethod
+    def default_value(cls, value: Any) -> str:
+        return "INVALID"
 
 
 class DAVPassword:
@@ -233,7 +222,7 @@ def check_digest_password(
 
         return False, None
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"{self.type}|{self.data}"
 
 
@@ -443,7 +432,7 @@ def nonce(self) -> str:
         return _md5(f"{uuid4().hex}{self.secret}")
 
     @staticmethod
-    def authorization_str_parser_to_data(authorization: str) -> dict:
+    def authorization_str_parser_to_data(authorization: str) -> dict[str, str]:
         values = authorization.split(",")
         data = dict()
         for value in values:
@@ -720,18 +709,3 @@ def _match_user_agent(rule: str, user_agent: str) -> bool:
             return False
 
         return True
-
-    @staticmethod
-    def _parser_digest_request(authorization: str) -> dict:
-        values = authorization[7:].split(",")
-
-        data = dict()
-        for value in values:
-            value = value.replace('"', "").replace(" ", "")
-            try:
-                k, v = value.split("=")
-                data[k] = v
-            except ValueError:
-                pass
-
-        return data

asgi_webdav/cache.py

@@ -20,10 +20,10 @@ class DAVCacheAbc:  # pragma: no cover
     async def prepare(self) -> None:
         raise NotImplementedError
 
-    async def get(self, key):
+    async def get(self, key: str | bytes) -> Any:
         raise NotImplementedError
 
-    async def set(self, key, value):
+    async def set(self, key: str | bytes, value: Any) -> None:
         raise NotImplementedError
 
     async def purge(self) -> None:
@@ -38,10 +38,10 @@ class DAVCacheBypass(DAVCacheAbc):
     async def prepare(self) -> None:  # pragma: no cover
         pass
 
-    async def get(self, key):
+    async def get(self, key: str | bytes) -> Any:
         return None
 
-    async def set(self, key, value):
+    async def set(self, key: str | bytes, value: Any) -> None:
         pass
 
     async def purge(self) -> None:
@@ -53,7 +53,7 @@ async def close(self) -> None:  # pragma: no cover
 
 class DAVCacheMemory(DAVCacheAbc):
     _lock: Lock
-    _cache: dict
+    _cache: dict[str | bytes, Any]
 
     def __init__(self) -> None:
         self._cache = {}
@@ -62,11 +62,11 @@ def __init__(self) -> None:
     async def prepare(self) -> None:  # pragma: no cover
         pass
 
-    async def get(self, key):
+    async def get(self, key: str | bytes) -> Any:
         async with self._lock:
             return self._cache.get(key)
 
-    async def set(self, key, value):
+    async def set(self, key: str | bytes, value: Any) -> None:
         async with self._lock:
             self._cache[key] = value
 
@@ -80,7 +80,7 @@ async def close(self) -> None:  # pragma: no cover
 
 class DAVCacheExpiring(DAVCacheAbc):
     _lock: Lock
-    _cache: dict[bytes, tuple[Any, datetime]]
+    _cache: dict[str | bytes, tuple[Any, datetime]]
     _cache_expiration_timedelta: timedelta
 
     def __init__(self, cache_expiration: int) -> None:
@@ -94,7 +94,7 @@ def __init__(self, cache_expiration: int) -> None:
     async def prepare(self) -> None:  # pragma: no cover
         pass
 
-    async def get(self, key):
+    async def get(self, key: str | bytes) -> Any:
         async with self._lock:
             cached = self._cache.get(key)
             if cached:
@@ -105,7 +105,7 @@ async def get(self, key):
                 # Cache entry expired
                 self._cache.pop(key, None)
 
-    async def set(self, key, value):
+    async def set(self, key: str | bytes, value: Any) -> None:
         async with self._lock:
             self._cache[key] = (value, datetime.now())
 

asgi_webdav/config.py

@@ -1,9 +1,11 @@
 import json
 import sys
+from collections.abc import Callable
 from dataclasses import dataclass, field
 from enum import Enum
 from logging import getLogger
 from pathlib import Path
+from typing import Any
 
 if sys.version_info >= (3, 11):
     import tomllib
@@ -105,8 +107,8 @@ class GuessTypeExtension:
     enable: bool = True
     enable_default_mapping: bool = True
 
-    filename_mapping: dict = field(default_factory=dict)
-    suffix_mapping: dict = field(default_factory=dict)
+    filename_mapping: dict[str, str] = field(default_factory=dict)
+    suffix_mapping: dict[str, str] = field(default_factory=dict)
 
 
 @dataclass
@@ -305,15 +307,19 @@ def get_config() -> Config:
     return _config
 
 
-def get_config_copy_from_dict(data: dict, complete_config: bool = False) -> Config:
+def get_config_copy_from_dict(
+    data: dict[str, Any], complete_config: bool = False
+) -> Config:
     config = Config.from_dict(data)
     if complete_config:
         config._complete_config()
 
     return config
 
 
-def reinit_config_from_dict(data: dict, complete_config: bool = False) -> Config:
+def reinit_config_from_dict(
+    data: dict[str, Any], complete_config: bool = False
+) -> Config:
     global _config
 
     logger.debug("Load config value from python object(dict)")
@@ -323,6 +329,8 @@ def reinit_config_from_dict(data: dict, complete_config: bool = False) -> Config
 
 
 def reinit_config_from_file(file_name: str, complete_config: bool = False) -> bool:
+    load_func: Callable[[Any], Any]
+
     file = Path(file_name)
     match file.suffix:
         case ".json":

asgi_webdav/constants.py

@@ -4,7 +4,7 @@
 from enum import Enum, IntEnum, auto
 from functools import cache
 from time import time
-from typing import NewType
+from typing import Any, NewType
 from uuid import UUID
 
 import arrow
@@ -23,31 +23,26 @@ class DAVUpperEnumAbc(Enum):
     默认值为空,需要继承实现;默认不会自动匹配默认值
     """
 
-    def __init__(self, *args, **kwargs) -> None:
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         self._value_ = self._name_.upper()
-
         label = args[0]
         if not isinstance(label, str):
             self.label = str(label)
         else:
             self.label = label
 
     @classmethod
-    def _missing_(cls, value):
+    def _missing_(cls, value: Any) -> "DAVUpperEnumAbc":
         if not isinstance(value, str):
             raise ValueError(f"Invalid {cls.__name__} value: {value}")
 
-        if "." in value:
-            # 兼容枚举前后端转换问题
-            value = value.split(".")[1]
-
         try:
             return cls[value.upper()]
         except KeyError:
             return cls[cls.default_value(value).upper()]
 
     @classmethod
-    def default_value(cls, value) -> str:
+    def default_value(cls, value: Any) -> str:
         raise ValueError(f"Invalid {cls.__name__} value: {value}")
 
     @classmethod
@@ -68,6 +63,9 @@ def value_label_mapping(cls) -> dict[str, str]:
 
 # WebDAV protocol ---
 class DAVMethod(DAVUpperEnumAbc):
+    # default/fallback
+    UNKNOWN = auto()
+
     # rfc4918:9.1
     PROPFIND = auto()
     # rfc4918:9.2
@@ -93,11 +91,8 @@ class DAVMethod(DAVUpperEnumAbc):
     # only for inside page
     POST = auto()
 
-    # only for request parser failed
-    UNKNOWN = auto()
-
     @classmethod
-    def default_value(cls, value) -> str:
+    def default_value(cls, value: Any) -> str:
         return "UNKNOWN"
 
     @classmethod
@@ -216,7 +211,10 @@ def add_child(self, child: "DAVPath | str") -> "DAVPath":
     def __hash__(self) -> int:
         return hash(self.raw)
 
-    def __eq__(self, other):
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, DAVPath):
+            return False
+
         return self.raw == other.raw
 
     def __lt__(self, other: "DAVPath") -> bool:

asgi_webdav/lock.py

@@ -1,5 +1,6 @@
 import asyncio
 import pprint
+from collections.abc import Iterable
 from time import time
 from uuid import UUID, uuid4
 
@@ -21,7 +22,7 @@ def __init__(self) -> None:
     def __contains__(self, item: DAVPath) -> bool:
         return item in self.data
 
-    def keys(self):
+    def keys(self) -> Iterable[DAVPath]:
         return self.data.keys()
 
     def get_tokens(self, path: DAVPath) -> list[UUID]:
@@ -129,7 +130,7 @@ async def is_locking(self, path: DAVPath, owner_token: UUID | None = None) -> bo
         return False
 
     async def get_info_by_path(self, path: DAVPath) -> list[DAVLockInfo]:
-        result = list()
+        result: list[DAVLockInfo] = list()
         async with self.lock:
             if path not in self.path2token_map:
                 return result
@@ -170,7 +171,7 @@ async def _release_by_path(self, path: DAVPath) -> None:
             for token in self.path2token_map.get_tokens(path):
                 self._remove_token(path, token)
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         s = "{}\n{}".format(
             pprint.pformat(self.path2token_map), pprint.pformat(self.lock_map)
         )

asgi_webdav/property.py

@@ -14,23 +14,20 @@ class DAVPropertyBasicData:
     last_modified: DAVTime
 
     # resource_type: str = field(init=False)
-    content_type: str | None = field(default=None)
+    content_type: str = ""
     content_charset: str | None = None
-    content_length: int = field(default=0)
+    content_length: int = 0
     content_encoding: str | None = None
 
     def __post_init__(self) -> None:
         # https://developer.mozilla.org/zh-CN/docs/Web/HTTP/Basics_of_HTTP/MIME_types
-        if self.content_type is None:
+        if not self.content_type:
             if self.is_collection:
                 # self.content_type = "httpd/unix-directory"
                 self.content_type = "application/index"
             else:
                 self.content_type = "application/octet-stream"
 
-        if self.content_length is None:
-            self.content_length = 0
-
     @property
     def etag(self) -> str:
         return generate_etag(self.content_length, self.last_modified.timestamp)
@@ -66,8 +63,8 @@ def get_get_head_response_headers(self) -> dict[bytes, bytes]:
 
         return headers
 
-    def as_dict(self) -> dict[str, str]:
-        data = {
+    def as_dict(self) -> dict[str, str | int]:
+        data: dict[str, str | int] = {
             "displayname": self.display_name,
             "getetag": self.etag,
             "creationdate": self.creation_date.dav_creation_date(),

asgi_webdav/provider/webhdfs.py

@@ -137,7 +137,7 @@ async def _create_dav_property_obj(
                 last_modified=DAVTime(float(file_status.get("modificationTime", 0.0))),
                 content_type=content_type,
                 content_charset=charset,
-                content_length=file_status.get("length"),
+                content_length=file_status.get("length", 0),
                 content_encoding=content_encoding,
             )