chore: update xmltodict to v1.0.2

Author: rexzhang

.pre-commit-config.yaml

@@ -1,20 +1,20 @@
 repos:
   - repo: https://github.com/asottile/pyupgrade
-    rev: v3.19.1
+    rev: v3.21.2
     hooks:
       - id: pyupgrade
         args: [ --py310-plus ]
   - repo: https://github.com/psf/black
-    rev: 25.1.0
+    rev: 25.12.0
     hooks:
       - id: black
         args: [ "--target-version", "py310" ]
   - repo: https://github.com/pycqa/isort
-    rev: 6.0.1
+    rev: 7.0.0
     hooks:
       - id: isort
         name: isort (python)
   - repo: https://github.com/PyCQA/flake8
-    rev: 7.2.0
+    rev: 7.3.0
     hooks:
       - id: flake8

TODO

@@ -51,13 +51,11 @@
         ☐ 文档自动部署失效问题
     - 每年定期更新上游依赖
         ☐ aiofiles
-        ☐ xmltodict
+        ✔ xmltodict v1.0.2 @done(25-12-09)
 
 待实现:
     ☐ 基于配置,检查依赖安装情况,以便在启动时告警
     ☐ 兼容 Py3.14  @block dataclass-wizard 不兼容 py3.14
-    ☐ 更新 xmltodict 到最新版本
-        这个上游库变化非常大, 之前版本与现在版本不兼容
     ☐ 浏览器前台界面 @low
         ☐ 更干净的目录过滤机制
         ☐ 更好看的界面

asgi_webdav/config.py

@@ -33,6 +33,7 @@
 class EnvConfig(EnvWizard):
     class _(EnvWizard.Meta):
         env_prefix = "WEBDAV_"
+        env_file = True
 
     username: str | None = None
     password: str | None = None

asgi_webdav/helpers.py

@@ -5,6 +5,7 @@
 from logging import getLogger
 from mimetypes import guess_type as orig_guess_type
 from pathlib import Path
+from typing import Any
 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 import aiofiles
@@ -147,13 +148,20 @@ def dav_dict2xml(data: dict) -> bytes:
     )
 
 
-def dav_xml2dict(data: bytes) -> dict | None:
+def get_dav_property_data_from_xml(data: bytes, propert_type: str) -> dict[str, Any]:
     try:
         result = xmltodict.parse(data, process_namespaces=True)
 
     except (xmltodict.ParsingInterrupted, xml.parsers.expat.ExpatError) as e:
-        logger.warning(f"parser XML failed, {e}, {data}")
-        return None
+        logger.warning(f"parser XML {propert_type} failed: {e}, xml: {data}")
+        return {}
+
+    try:
+        result = result[f"DAV::{propert_type}"]
+
+    except (ValueError, KeyError) as e:
+        logger.warning(f"parser XML {propert_type} failed: {e}, xml: {data}")
+        return {}
 
     return result
 

asgi_webdav/request.py

@@ -2,6 +2,7 @@
 import urllib.parse
 from collections.abc import Callable
 from dataclasses import dataclass, field
+from logging import getLogger
 from pyexpat import ExpatError
 from uuid import UUID
 
@@ -18,7 +19,15 @@
     DAVPropertyPatches,
     DAVUser,
 )
-from asgi_webdav.helpers import dav_xml2dict, receive_all_data_in_one_call
+from asgi_webdav.helpers import (
+    get_dav_property_data_from_xml,
+    receive_all_data_in_one_call,
+)
+
+logger = getLogger(__name__)
+
+
+_XML_NAME_SPACE_TAG = "@xmlns"
 
 
 @dataclass
@@ -325,7 +334,6 @@ def _cut_ns_key(ns_key: str) -> tuple[str, str]:
             return ns_key[:index], ns_key[index + 1 :]
 
     async def _parser_body_propfind(self) -> bool:
-        self.body = await receive_all_data_in_one_call(self.receive)
         """
         A client may choose not to submit a request body.  An empty PROPFIND
            request body MUST be treated as if it were an 'allprop' request.
@@ -334,25 +342,24 @@ async def _parser_body_propfind(self) -> bool:
             # allprop
             return True
 
-        data = dav_xml2dict(self.body)
-        if data is None:
+        data = get_dav_property_data_from_xml(self.body, "propfind")
+        if not data:
             return False
 
-        find_symbol = "DAV::propfind"
-        if "propname" in data[find_symbol]:
+        if "propname" in data:
             self.propfind_only_fetch_property_name = True
             return True
 
-        if "DAV::allprop" in data[find_symbol]:
+        if "DAV::allprop" in data:
             return True
         else:
             self.propfind_fetch_all_property = False
 
-        if "DAV::prop" not in data[find_symbol]:
+        if "DAV::prop" not in data:
             # TODO error
             return False
 
-        for ns_key in data[find_symbol]["DAV::prop"]:
+        for ns_key in data["DAV::prop"]:
             ns, key = self._cut_ns_key(ns_key)
             if key in DAV_PROPERTY_BASIC_KEYS:
                 self.propfind_basic_keys.add(key)
@@ -365,33 +372,42 @@ async def _parser_body_propfind(self) -> bool:
         return True
 
     async def _parser_body_proppatch(self) -> bool:
-        self.body = await receive_all_data_in_one_call(self.receive)
-        data = dav_xml2dict(self.body)
-        if data is None:
+        data = get_dav_property_data_from_xml(self.body, "propertyupdate")
+        if not data:
             return False
 
-        update_symbol = "DAV::propertyupdate"
-        for action in data[update_symbol]:
-            _, key = self._cut_ns_key(action)
-            if key == "set":
+        for action, action_data in data.items():
+            if action == _XML_NAME_SPACE_TAG:
+                continue
+
+            _, action_method = self._cut_ns_key(action)
+            if action_method == "set":
                 method = True
             else:
+                # remove
                 method = False
 
-            for item in data[update_symbol][action]:
-                if isinstance(item, dict):
-                    ns_key, value = item["DAV::prop"].popitem()
-                else:
-                    ns_key, value = data[update_symbol][action][item].popitem()
-                    if isinstance(value, dict):
-                        # value namespace: drop namespace info # TODO ???
-                        value, _ = value.popitem()
-                        _, value = self._cut_ns_key(value)
-                        # value = "<{} xmlns='{}'>".format(vns_key, vns_ns)
+            if isinstance(action_data, dict):
+                # 当 action 只有一条的时候, actions_data 是一个 dict, 需要转换为 list 以便后续处理
+                action_data = [action_data]
 
+            for action_item in action_data:
+
+                ns_key, dav_prop_data = action_item["DAV::prop"].popitem()
                 ns, key = self._cut_ns_key(ns_key)
+
+                # value = value.get("#text")
+                value = None
+                for prop_key, prop_value in dav_prop_data.items():
+                    if prop_key == _XML_NAME_SPACE_TAG:
+                        continue
+                    if prop_key == "#text":
+                        value = prop_value
+                    else:
+                        _, value = self._cut_ns_key(prop_key)
+
                 if not isinstance(value, str):
-                    value = str(value)
+                    value = str(value)  # TODO: 可能不需要转换?
 
                 self.proppatch_entries.append(
                     DAVPropertyPatches([DAVPropertyIdentity((ns, key)), value, method])
@@ -400,36 +416,39 @@ async def _parser_body_proppatch(self) -> bool:
         return True
 
     async def _parser_body_lock(self) -> bool:
-        self.body = await receive_all_data_in_one_call(self.receive)
         if len(self.body) == 0:
             # LOCK accept empty body
             return True
 
-        data = dav_xml2dict(self.body)
-        if data is None:
+        data = get_dav_property_data_from_xml(self.body, "lockinfo")
+        if not data:
             return False
 
-        if "DAV::exclusive" in data["DAV::lockinfo"]["DAV::lockscope"]:
+        if "DAV::exclusive" in data["DAV::lockscope"]:
             self.lock_scope = DAVLockScope.exclusive
         else:
             self.lock_scope = DAVLockScope.shared
 
-        lock_owner = data["DAV::lockinfo"]["DAV::owner"]
+        lock_owner = data["DAV::owner"]
         self.lock_owner = str(lock_owner)
         return True
 
     async def parser_body(self) -> bool:
-        if self.method == DAVMethod.PROPFIND:
-            self.body_is_parsed_success = await self._parser_body_propfind()
+        match self.method:
+            case DAVMethod.PROPFIND:
+                self.body = await receive_all_data_in_one_call(self.receive)
+                self.body_is_parsed_success = await self._parser_body_propfind()
 
-        elif self.method == DAVMethod.PROPPATCH:
-            self.body_is_parsed_success = await self._parser_body_proppatch()
+            case DAVMethod.PROPPATCH:
+                self.body = await receive_all_data_in_one_call(self.receive)
+                self.body_is_parsed_success = await self._parser_body_proppatch()
 
-        elif self.method == DAVMethod.LOCK:
-            self.body_is_parsed_success = await self._parser_body_lock()
+            case DAVMethod.LOCK:
+                self.body = await receive_all_data_in_one_call(self.receive)
+                self.body_is_parsed_success = await self._parser_body_lock()
 
-        else:
-            self.body_is_parsed_success = False
+            case _:
+                self.body_is_parsed_success = False
 
         return self.body_is_parsed_success
 

docs/changelog.en.md

@@ -5,6 +5,7 @@
 - Broken change:
   - remove brotli support
 - feat: support zstd compression in response
+- chore: update xmltodict to v1.0.2
 
 ## 1.6.1 - 20251101
 

requirements.d/basic.txt

@@ -4,10 +4,10 @@ aiofiles~=23.2.0
 ASGIMiddlewareStaticFile~=0.6.0
 
 # data
-xmltodict~=0.13.0
+xmltodict~=1.0.2
 
 # config
-dataclass-wizard<1.0
+dataclass-wizard[dotenv]<1.0
 tomli; python_version < '3.11'
 
 # misc

tests/test_helpers.py

@@ -6,6 +6,7 @@
 from asgi_webdav.helpers import (
     detect_charset,
     get_data_generator_from_content,
+    get_dav_property_data_from_xml,
     guess_type,
     is_browser_user_agent,
     paser_timezone_key,
@@ -123,6 +124,35 @@ async def test_func_get_data_generator_from_content():
     assert len(data_new) == 100
 
 
+def test_get_dav_property_data_from_xml():
+    # all good
+    assert get_dav_property_data_from_xml(
+        data=b'<?xml version="1.0" encoding="utf-8" ?>\n<D:propertyupdate xmlns:D="DAV:"><D:set><D:prop><random xmlns="http://webdav.org/neon/litmus/">foobar</random></D:prop></D:set>\n</D:propertyupdate>\n',
+        propert_type="propertyupdate",
+    ) == {
+        "@xmlns": {"D": "DAV:"},
+        "DAV::set": {
+            "DAV::prop": {
+                "http://webdav.org/neon/litmus/:random": {
+                    "@xmlns": {"": "http://webdav.org/neon/litmus/"},
+                    "#text": "foobar",
+                }
+            }
+        },
+    }
+
+    # bad
+    assert get_dav_property_data_from_xml(data=b"", propert_type="propertyupdate") == {}
+
+    assert (
+        get_dav_property_data_from_xml(
+            data=b'<?xml version="1.0" encoding="utf-8" ?>\n<D:propertyupdate xmlns:D="DAV:"><D:set><D:prop><random xmlns="http://webdav.org/neon/litmus/">foobar</random></D:prop></D:set>\n</D:propertyupdate>\n',
+            propert_type="bad",
+        )
+        == {}
+    )
+
+
 def test_get_timezone_from_env():
     assert paser_timezone_key("Asia/Shanghai") == "Asia/Shanghai"