perf(memory): import storage and passkey SDKs on demand

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/application/security/passkey.py

@@ -6,30 +6,16 @@ import binascii
 import json
 import secrets
 from dataclasses import dataclass
-from typing import Any, Dict, List, Literal, Optional, Protocol, Tuple
+from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Protocol, Tuple
 from urllib.parse import urlparse
 
-from webauthn import (
-    generate_authentication_options,
-    generate_registration_options,
-    options_to_json,
-    verify_authentication_response,
-    verify_registration_response,
-)
-from webauthn.helpers import parse_authentication_credential_json, parse_registration_credential_json
-from webauthn.helpers.cose import COSEAlgorithmIdentifier
-from webauthn.helpers.exceptions import InvalidRegistrationResponse
-from webauthn.helpers.structs import (
-    AuthenticatorSelectionCriteria,
-    AuthenticatorTransport,
-    PublicKeyCredentialDescriptor,
-    ResidentKeyRequirement,
-    UserVerificationRequirement,
-)
-
 from app.application.configuration import get_api_runtime_config_snapshot
 from app.runtime.log import logger
 
+# webauthn 约 14MB，只在 Passkey 注册与登录时才用到，改为方法内导入，避免随 API 路由加载常驻
+if TYPE_CHECKING:
+    from webauthn.helpers.structs import PublicKeyCredentialDescriptor, UserVerificationRequirement
+
 PASSKEY_CHALLENGE_TTL_SECONDS = 5 * 60
 PasskeyChallengePurpose = Literal["authentication", "registration"]
 
@@ -206,6 +192,7 @@ class PassKeyHelper:
         :param credentials: 凭证字典列表
         :return: PublicKeyCredentialDescriptor 列表
         """
+        from webauthn.helpers.structs import AuthenticatorTransport, PublicKeyCredentialDescriptor
         result = []
         for cred in credentials:
             try:
@@ -230,6 +217,7 @@ class PassKeyHelper:
         :param user_verification: 指定的用户验证要求，如果不指定则从配置中读取
         :return: UserVerificationRequirement
         """
+        from webauthn.helpers.structs import UserVerificationRequirement
         if user_verification:
             return UserVerificationRequirement(user_verification)
         return UserVerificationRequirement.REQUIRED if get_api_runtime_config_snapshot().passkey_require_uv \
@@ -267,6 +255,9 @@ class PassKeyHelper:
         :param existing_credentials: 已存在的凭证列表
         :return: (options_json, challenge)
         """
+        from webauthn import generate_registration_options, options_to_json
+        from webauthn.helpers.cose import COSEAlgorithmIdentifier
+        from webauthn.helpers.structs import AuthenticatorSelectionCriteria, ResidentKeyRequirement
         try:
             # 用户信息
             user_id_bytes = str(user_id).encode('utf-8')
@@ -325,6 +316,9 @@ class PassKeyHelper:
         :param expected_rp_id: 期望的RP ID
         :return: (credential_id, public_key, sign_count, aaguid)
         """
+        from webauthn import verify_registration_response
+        from webauthn.helpers import parse_registration_credential_json
+        from webauthn.helpers.exceptions import InvalidRegistrationResponse
         try:
             # 准备验证参数
             origin, rp_id = PassKeyHelper._get_verification_params(expected_origin, expected_rp_id)
@@ -379,6 +373,7 @@ class PassKeyHelper:
         :param user_verification: 用户验证要求，如果不指定则从配置中读取
         :return: (options_json, challenge)
         """
+        from webauthn import generate_authentication_options, options_to_json
         try:
             # 允许的凭证
             allow_credentials = PassKeyHelper._parse_credential_list(existing_credentials) \
@@ -426,6 +421,8 @@ class PassKeyHelper:
         :param expected_rp_id: 期望的RP ID
         :return: (验证成功, 新的签名计数)
         """
+        from webauthn import verify_authentication_response
+        from webauthn.helpers import parse_authentication_credential_json
         try:
             # 准备验证参数
             origin, rp_id = PassKeyHelper._get_verification_params(expected_origin, expected_rp_id)

app/modules/filemanager/storages/smb.py

@@ -4,14 +4,6 @@ import time
 from pathlib import Path
 from typing import List, Optional, Union
 
-import smbclient
-from smbclient import ClientConfig, register_session, reset_connection_cache
-from smbprotocol.exceptions import (
-    SMBAuthenticationError,
-    SMBException,
-    SMBResponseException,
-)
-
 from app.foundation.singleton import WeakSingleton
 from app.modules.filemanager.storages import StorageBase, transfer_process
 from app.runtime.log import logger
@@ -22,6 +14,9 @@ from app.schemas.file import StorageUsage as _SchemaStorageUsage
 from app.schemas.types import StorageSchema
 from app.schemas.workflow import FileItem as _SchemaFileItem
 
+# SMB SDK（smbclient/smbprotocol/spnego）约 12MB，存储发现会导入全部存储实现，
+# 因此各方法内按需导入，未配置 SMB 时不加载
+
 lock = threading.Lock()
 
 
@@ -65,6 +60,7 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         """
         初始化SMB连接配置
         """
+        from smbclient import ClientConfig, register_session
         try:
             conf = self.get_conf()
             if not conf:
@@ -125,6 +121,8 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         """
         测试SMB连接
         """
+        import smbclient
+        from smbprotocol.exceptions import SMBAuthenticationError, SMBException, SMBResponseException
         try:
             # 尝试列出根目录来测试连接
             smbclient.listdir(self._server_path)
@@ -177,6 +175,7 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         """
         创建文件项
         """
+        import smbclient
         try:
             # 检查是否为目录
             is_directory = smbclient.path.isdir(file_path)
@@ -231,6 +230,7 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         """
         初始化存储
         """
+        from smbclient import reset_connection_cache
         # 重置连接缓存
         reset_connection_cache()
         self._init_connection()
@@ -254,6 +254,8 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         """
         浏览文件
         """
+        import smbclient
+        from smbprotocol.exceptions import SMBException, SMBResponseException
         try:
             self._check_connection()
 
@@ -301,6 +303,7 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         """
         创建目录
         """
+        import smbclient
         try:
             self._check_connection()
 
@@ -354,6 +357,7 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         """
         获取文件或目录，不存在返回None
         """
+        import smbclient
         try:
             self._check_connection()
 
@@ -389,6 +393,7 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         只有 ENOENT/ENOTDIR 才是「确认不存在」，连接中断、认证失败等都无法确认
         目标状态，必须保守失败以免覆盖保护被绕过。
         """
+        import smbclient
         try:
             self._check_connection()
 
@@ -426,6 +431,8 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         """
         删除文件或目录
         """
+        import smbclient
+        from smbprotocol.exceptions import SMBException, SMBResponseException
         try:
             self._check_connection()
 
@@ -465,6 +472,8 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         """
         递归删除目录及其所有内容
         """
+        import smbclient
+        from smbprotocol.exceptions import SMBException, SMBResponseException
         try:
             # 检查路径是否存在
             if not smbclient.path.exists(smb_path):
@@ -528,6 +537,7 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         """
         重命名文件
         """
+        import smbclient
         try:
             self._check_connection()
 
@@ -548,6 +558,7 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         """
         带实时进度显示的下载
         """
+        import smbclient
         local_path = self._build_download_path(fileitem, path or get_runtime_setting('TEMP_PATH'))
         if not local_path:
             return None
@@ -601,6 +612,7 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         """
         带实时进度显示的上传
         """
+        import smbclient
         target_name = new_name or path.name
         target_path = Path(fileitem.path) / target_name
         smb_path = self._normalize_path(str(target_path))
@@ -695,6 +707,8 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         硬链接文件
         Samba服务器需要开启 unix extensions 支持
         """
+        import smbclient
+        from smbprotocol.exceptions import SMBResponseException
         try:
             self._check_connection()
             src_path = self._normalize_path(fileitem.path)
@@ -730,6 +744,7 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         """
         存储使用情况
         """
+        import smbclient
         try:
             self._check_connection()
             volume_stat = smbclient.stat_volume(self._server_path)
@@ -748,6 +763,8 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         """
         try:
             if self._connected:
+                from smbclient import reset_connection_cache
+
                 reset_connection_cache()
         except Exception as e:
             logger.debug(f"【SMB】清理连接失败: {e}")

app/modules/filemanager/storages/u115.py

@@ -6,10 +6,7 @@ from pathlib import Path
 from threading import Lock
 from typing import List, Optional, Tuple, Union
 
-import oss2
 from cryptography.hazmat.primitives import hashes
-from oss2 import SizedFileAdapter, determine_part_size
-from oss2.models import PartInfo
 
 from app.adapters.network.http import HttpRequestError, RequestUtils
 from app.foundation import size as size_tools
@@ -600,6 +597,10 @@ class U115Pan(StorageBase, metaclass=WeakSingleton):
         """
         实现带秒传、断点续传和二次认证的文件上传
         """
+        # oss2 连同阿里云 SDK 约 19MB，存储发现会导入全部存储实现，上传时才按需导入
+        import oss2
+        from oss2 import SizedFileAdapter, determine_part_size
+        from oss2.models import PartInfo
 
         def encode_callback(cb: str) -> str:
             """

tests/test_mfa_passkey_registration_errors.py

@@ -2,10 +2,11 @@ from types import SimpleNamespace
 from unittest.mock import patch
 
 import pytest
+import webauthn
+import webauthn.helpers
 from webauthn.helpers.exceptions import InvalidRegistrationResponse
 
 from app.api.endpoints import mfa as mfa_endpoint
-from app.application.security import passkey as passkey_helper
 from app.application.security.passkey import (
     PASSKEY_CHALLENGE_TTL_SECONDS,
     PasskeyChallengeStore,
@@ -52,12 +53,13 @@ def test_passkey_helper_classifies_origin_mismatch():
         'expected "http://localhost:3000"'
     )
 
+    # passkey 在方法内按需导入 webauthn，替身打在 webauthn 的来源模块上
     with patch.object(
-        passkey_helper,
+        webauthn.helpers,
         "parse_registration_credential_json",
         return_value=object(),
     ), patch.object(
-        passkey_helper,
+        webauthn,
         "verify_registration_response",
         side_effect=library_error,
     ), pytest.raises(PassKeyRegistrationOriginMismatchError) as exc_info:
@@ -75,12 +77,13 @@ def test_passkey_helper_classifies_other_verification_failure():
         "Client data challenge was not expected challenge"
     )
 
+    # passkey 在方法内按需导入 webauthn，替身打在 webauthn 的来源模块上
     with patch.object(
-        passkey_helper,
+        webauthn.helpers,
         "parse_registration_credential_json",
         return_value=object(),
     ), patch.object(
-        passkey_helper,
+        webauthn,
         "verify_registration_response",
         side_effect=library_error,
     ), pytest.raises(PassKeyRegistrationVerificationError) as exc_info:

tests/test_optional_sdk_lazy_import.py

@@ -0,0 +1,46 @@
+"""可选 SDK 的按需导入合同：存储发现与 API 路由加载不得把 115、SMB、Passkey 的 SDK 带入常驻内存。"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+
+# 各 SDK 连同依赖的顶层包；oss2 会牵入阿里云 SDK
+_OPTIONAL_SDKS = ("oss2", "aliyunsdkcore", "aliyunsdkkms", "smbclient", "smbprotocol", "spnego", "webauthn")
+
+
+def _run_isolated(script: str) -> dict:
+    """在全新解释器中执行导入探针，避免当前 pytest 模块缓存干扰。"""
+    completed = subprocess.run(
+        [sys.executable, "-c", script],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    return json.loads(completed.stdout.strip().splitlines()[-1])
+
+
+def test_storage_discovery_and_api_routes_keep_optional_sdks_cold() -> None:
+    """文件整理模块发现存储、注册 API 路由后，可选 SDK 仍未导入。"""
+    result = _run_isolated(
+        f"""
+import json
+import sys
+
+import app.api.apiv1
+from app.foundation.reflection import ModuleHelper
+
+storages = ModuleHelper.load(
+    "app.modules.filemanager.storages",
+    filter_func=lambda _, obj: hasattr(obj, "schema") and obj.schema,
+)
+forbidden = {_OPTIONAL_SDKS!r}
+loaded = sorted(name for name in sys.modules if name.split(".")[0] in forbidden)
+print(json.dumps({{"loaded": loaded, "schemas": sorted(s.schema.value for s in storages)}}))
+"""
+    )
+
+    assert result["loaded"] == []
+    # 按需导入不影响存储发现
+    assert {"smb", "u115"} <= set(result["schemas"])

tests/test_u115_storage.py

@@ -102,10 +102,10 @@ def _upload_with_fakes(storage: U115Pan, target_dir: FileItem, local_file):
     """
 
     with patch(
-        "app.modules.filemanager.storages.u115.oss2.StsAuth",
+        "oss2.StsAuth",
         return_value=object(),
     ), patch(
-        "app.modules.filemanager.storages.u115.oss2.Bucket",
+        "oss2.Bucket",
         _FakeBucket,
     ), patch(
         "app.modules.filemanager.storages.u115.transfer_process",
@@ -158,7 +158,7 @@ def test_upload_uses_dynamic_part_size(tmp_path):
 
     storage = _build_storage()
     with patch(
-        "app.modules.filemanager.storages.u115.determine_part_size",
+        "oss2.determine_part_size",
         return_value=local_file.stat().st_size,
     ) as determine_part_size_mock:
         uploaded_item = _upload_with_fakes(storage, _target_dir(), local_file)