From c744bc0bc3d07057f00a073dc2a0ba995eb7a456 Mon Sep 17 00:00:00 2001 From: Von Date: Mon, 6 Apr 2026 13:00:31 +0800 Subject: [PATCH] fix pylance problems --- main.py | 12 ++- src/application/__init__.py | 5 + src/application/app_context.py | 40 +++++++- src/application/application.py | 11 +- src/config/__init__.py | 6 ++ src/config/provider_manager.py | 4 +- src/executors/contracts.py | 3 +- src/executors/registry.py | 8 +- src/external/llm_provider.py | 9 +- src/hooks/contracts.py | 4 +- src/presentation/auth_controller.py | 40 ++++---- src/presentation/decorators.py | 13 ++- src/presentation/provider_controller.py | 85 +++++++++------- src/presentation/proxy_controller.py | 32 +++--- src/presentation/user_controller.py | 128 ++++++++++++++---------- src/presentation/web_controller.py | 97 +++++++++--------- src/repositories/user_repository.py | 4 +- src/services/auth_group_service.py | 1 + src/services/user_service.py | 3 +- src/translators/claude_bridge.py | 7 +- src/translators/registry.py | 19 ++-- src/translators/responses_bridge.py | 2 +- src/utils/compat.py | 65 +++++++----- tests/test_hook_contracts.py | 21 ++-- tests/test_proxy_error_handling.py | 6 +- tests/test_user_model_permissions.py | 31 +++++- 26 files changed, 405 insertions(+), 251 deletions(-) diff --git a/main.py b/main.py index 8175c30..3e292a4 100644 --- a/main.py +++ b/main.py @@ -3,18 +3,20 @@ import argparse from pathlib import Path +from typing import TYPE_CHECKING from gevent import monkey monkey.patch_all() -import urllib3 - -from src.application import Application +import urllib3 # noqa: E402 # 禁用 HTTPS 证书告警 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) +if TYPE_CHECKING: + from src.application import Application + def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="LLM Proxy Service") @@ -24,7 +26,9 @@ def parse_args() -> argparse.Namespace: return parser.parse_args() -def app() -> Application: +def app() -> "Application": + from src.application import Application + args = parse_args() config_path = args.config project_root = Path(__file__).resolve().parent diff --git a/src/application/__init__.py b/src/application/__init__.py index 57e8d65..7a12ffe 100644 --- a/src/application/__init__.py +++ b/src/application/__init__.py @@ -2,8 +2,13 @@ # -*- coding: utf-8 -*- """应用层导出。""" +from typing import TYPE_CHECKING + from .app_context import AppContext, Logger +if TYPE_CHECKING: + from .application import Application + __all__ = ["Application", "AppContext", "Logger"] diff --git a/src/application/app_context.py b/src/application/app_context.py index 23d205d..672780d 100644 --- a/src/application/app_context.py +++ b/src/application/app_context.py @@ -14,13 +14,45 @@ if TYPE_CHECKING: class Logger(Protocol): - def info(self, msg: str) -> None: ... + def info( + self, + msg: object, + *args: object, + exc_info: object = None, + stack_info: bool = False, + stacklevel: int = 1, + extra: object | None = None, + ) -> None: ... - def error(self, msg: str) -> None: ... + def error( + self, + msg: object, + *args: object, + exc_info: object = None, + stack_info: bool = False, + stacklevel: int = 1, + extra: object | None = None, + ) -> None: ... - def warning(self, msg: str) -> None: ... + def warning( + self, + msg: object, + *args: object, + exc_info: object = None, + stack_info: bool = False, + stacklevel: int = 1, + extra: object | None = None, + ) -> None: ... - def debug(self, msg: str) -> None: ... + def debug( + self, + msg: object, + *args: object, + exc_info: object = None, + stack_info: bool = False, + stacklevel: int = 1, + extra: object | None = None, + ) -> None: ... @dataclass(frozen=True) diff --git a/src/application/application.py b/src/application/application.py index a1c97d2..51550f0 100644 --- a/src/application/application.py +++ b/src/application/application.py @@ -4,10 +4,11 @@ import logging from logging.handlers import RotatingFileHandler from pathlib import Path +from typing import Any, cast from flask import request -from .app_context import AppContext +from .app_context import AppContext, Logger from ..config import ( ConfigManager, ProviderManager, @@ -110,8 +111,8 @@ class Application: access_logger.setLevel(level) self._logger = logger - self._flask_app.logger = logger - self._flask_app.access_logger = access_logger + self._access_logger = access_logger + cast(Any, self._flask_app).logger = logger def _setup_request_access_logging(self) -> None: """注册请求访问日志钩子，仅记录 IP、URL 与模型。""" @@ -136,12 +137,12 @@ class Application: log_message = f'{log_message} username={username}' if model is not None: log_message = f'{log_message} model={model}' - self._flask_app.access_logger.info(log_message) + self._access_logger.info(log_message) def _setup_context(self) -> None: """创建全局运行上下文。""" self._ctx = AppContext( - logger=self._logger, + logger=cast(Logger, self._logger), config_manager=self._config_manager, root_path=self._root_path, flask_app=self._flask_app, diff --git a/src/config/__init__.py b/src/config/__init__.py index 10c8ddc..9789cac 100644 --- a/src/config/__init__.py +++ b/src/config/__init__.py @@ -2,6 +2,8 @@ # -*- coding: utf-8 -*- """配置层导出。""" +from typing import TYPE_CHECKING + from .config_manager import ConfigManager from .provider_config import ( AuthEntrySchema, @@ -15,6 +17,10 @@ from .provider_config import ( validate_auth_group_provider_definitions, ) +if TYPE_CHECKING: + from .auth_group_manager import AuthGroupManager + from .provider_manager import ProviderManager + __all__ = [ "ConfigManager", "AuthEntrySchema", diff --git a/src/config/provider_manager.py b/src/config/provider_manager.py index ad55088..30ad5f9 100644 --- a/src/config/provider_manager.py +++ b/src/config/provider_manager.py @@ -142,9 +142,9 @@ class ProviderManager: self._hook_cache[cache_key] = None return None + hook_dir = str(hook_file.parent) + path_inserted = False try: - hook_dir = str(hook_file.parent) - path_inserted = False if hook_dir not in sys.path: sys.path.insert(0, hook_dir) path_inserted = True diff --git a/src/executors/contracts.py b/src/executors/contracts.py index 4b97e64..6ec6151 100644 --- a/src/executors/contracts.py +++ b/src/executors/contracts.py @@ -24,7 +24,8 @@ class OpenedUpstreamResponse: class Executor(Protocol): """Transport-specific executor that opens an upstream response.""" - transport: str + @property + def transport(self) -> str: ... def execute( self, diff --git a/src/executors/registry.py b/src/executors/registry.py index 4ae2519..d345e2e 100644 --- a/src/executors/registry.py +++ b/src/executors/registry.py @@ -157,12 +157,12 @@ class WebSocketExecutor: connection.send(json.dumps(body, ensure_ascii=False)) if requested_stream: - response = WebSocketUpstreamResponse(connection) + stream_response = WebSocketUpstreamResponse(connection) stream_format = resolve_stream_format(None, "", provider.transport) return OpenedUpstreamResponse( - response=response, - status_code=response.status_code, - content_type=(response.headers.get("Content-Type") or "").lower(), + response=stream_response, + status_code=stream_response.status_code, + content_type=(stream_response.headers.get("Content-Type") or "").lower(), is_stream=True, stream_format=stream_format, ) diff --git a/src/external/llm_provider.py b/src/external/llm_provider.py index 257d1bb..e00f209 100644 --- a/src/external/llm_provider.py +++ b/src/external/llm_provider.py @@ -4,7 +4,7 @@ from __future__ import annotations -from typing import Any, Dict, Optional +from typing import Any, Callable, Dict, Optional, cast from ..config.provider_config import ( DEFAULT_PROVIDER_TARGET_FORMAT, @@ -66,7 +66,10 @@ class LLMProvider: if not self.hook: return body - guard = getattr(self.hook, "request_guard", None) + guard = cast( + Optional[Callable[[HookContext, Dict[str, Any]], Optional[Dict[str, Any]]]], + getattr(self.hook, "request_guard", None), + ) if callable(guard): guarded = guard(ctx, body) return body if guarded is None else guarded @@ -76,7 +79,7 @@ class LLMProvider: if not self.hook: return body - guard = getattr(self.hook, "response_guard", None) + guard = cast(Optional[Callable[[HookContext, Any], Any]], getattr(self.hook, "response_guard", None)) if callable(guard): guarded = guard(ctx, body) return body if guarded is None else guarded diff --git a/src/hooks/contracts.py b/src/hooks/contracts.py index 0da0f00..56f1c3e 100644 --- a/src/hooks/contracts.py +++ b/src/hooks/contracts.py @@ -53,7 +53,7 @@ class HookModule(Protocol): def header_hook(self, ctx: HookContext, headers: Dict[str, str]) -> Dict[str, str]: ... - def request_guard(self, ctx: HookContext, body: Dict[str, Any]) -> Dict[str, Any]: + def request_guard(self, ctx: HookContext, body: Dict[str, Any]) -> Optional[Dict[str, Any]]: ... def response_guard(self, ctx: HookContext, body: Any) -> Any: @@ -66,7 +66,7 @@ class BaseHook: def header_hook(self, ctx: HookContext, headers: Dict[str, str]) -> Dict[str, str]: return headers - def request_guard(self, ctx: HookContext, body: Dict[str, Any]) -> Dict[str, Any]: + def request_guard(self, ctx: HookContext, body: Dict[str, Any]) -> Optional[Dict[str, Any]]: return body def response_guard(self, ctx: HookContext, body: Any) -> Any: diff --git a/src/presentation/auth_controller.py b/src/presentation/auth_controller.py index fc35665..b9f0816 100644 --- a/src/presentation/auth_controller.py +++ b/src/presentation/auth_controller.py @@ -2,7 +2,11 @@ # -*- coding: utf-8 -*- """认证控制器。""" -from flask import Response, jsonify, make_response, redirect, render_template, request +from typing import Any + +from flask import jsonify, make_response, redirect, render_template, request +from flask.typing import ResponseReturnValue + from ..application.app_context import AppContext from ..services import AuthenticationService @@ -17,15 +21,20 @@ class AuthenticationController: self._auth_service = auth_service self._register_routes() + @staticmethod + def _get_request_payload() -> dict[str, Any]: + payload = request.get_json(silent=True) + if isinstance(payload, dict): + return dict(payload) + return {} + def _register_routes(self) -> None: - """注册认证相关路由。""" self._app.route('/login')(self.login_page) self._app.route('/api/login', methods=['POST'])(self.api_login) self._app.route('/logout')(self.logout) self._app.route('/api/logout', methods=['POST'])(self.api_logout) - def login_page(self) -> Response: - """返回登录页，已登录用户会被重定向到首页。""" + def login_page(self) -> ResponseReturnValue: if not self._auth_service.is_auth_enabled(): return redirect('/') @@ -35,47 +44,44 @@ class AuthenticationController: return render_template('login.html') - def api_login(self) -> Response: - """处理登录请求并写入 session cookie。""" + def api_login(self) -> ResponseReturnValue: if not self._auth_service.is_auth_enabled(): return jsonify({'message': 'Authentication not enabled'}), 200 - data = request.get_json(silent=True) or {} + data = self._get_request_payload() username = data.get('username') password = data.get('password') - if not username or not password: - self._logger.warning("Login rejected: missing username or password") + if not isinstance(username, str) or not username or not isinstance(password, str) or not password: + self._logger.warning('Login rejected: missing username or password') return jsonify({'error': 'Username and password are required'}), 400 if not self._auth_service.authenticate(username, password): - self._logger.warning(f"Login failed: invalid credentials for username={username!r}") + self._logger.warning('Login failed: invalid credentials for username=%r', username) return jsonify({'error': 'Invalid username or password'}), 401 session_token = self._auth_service.create_session(username) - self._logger.info(f"Login succeeded: username={username!r}") + self._logger.info('Login succeeded: username=%r', username) response = make_response(jsonify({'message': 'Login successful', 'username': username})) cookie_settings = self._auth_service.get_cookie_settings() response.set_cookie('session_token', session_token, **cookie_settings) return response - def logout(self) -> Response: - """处理页面登出并清理 cookie。""" + def logout(self) -> ResponseReturnValue: session_token = request.cookies.get('session_token') if session_token: self._auth_service.destroy_session(session_token) - self._logger.info("Logout succeeded via page route") + self._logger.info('Logout succeeded via page route') response = make_response(redirect('/login')) response.delete_cookie('session_token') return response - def api_logout(self) -> Response: - """处理 API 登出并清理 cookie。""" + def api_logout(self) -> ResponseReturnValue: session_token = request.cookies.get('session_token') if session_token: self._auth_service.destroy_session(session_token) - self._logger.info("Logout succeeded via API route") + self._logger.info('Logout succeeded via API route') response = make_response(jsonify({'message': 'Logout successful'})) response.delete_cookie('session_token') diff --git a/src/presentation/decorators.py b/src/presentation/decorators.py index bb1b56f..56ed5bf 100644 --- a/src/presentation/decorators.py +++ b/src/presentation/decorators.py @@ -3,19 +3,24 @@ """认证装饰器。""" from functools import wraps -from typing import Callable +from typing import Callable, ParamSpec from flask import jsonify, redirect, request +from flask.typing import ResponseReturnValue from ..services import AuthenticationService +P = ParamSpec("P") -def require_authentication(auth_service: AuthenticationService): + +def require_authentication( + auth_service: AuthenticationService, +) -> Callable[[Callable[P, ResponseReturnValue]], Callable[P, ResponseReturnValue]]: """生成认证校验装饰器。""" - def decorator(func: Callable) -> Callable: + def decorator(func: Callable[P, ResponseReturnValue]) -> Callable[P, ResponseReturnValue]: @wraps(func) - def decorated_function(*args, **kwargs): + def decorated_function(*args: P.args, **kwargs: P.kwargs) -> ResponseReturnValue: if not auth_service.is_auth_enabled(): return func(*args, **kwargs) diff --git a/src/presentation/provider_controller.py b/src/presentation/provider_controller.py index fba706b..c74f803 100644 --- a/src/presentation/provider_controller.py +++ b/src/presentation/provider_controller.py @@ -2,7 +2,10 @@ # -*- coding: utf-8 -*- """Provider 管理 API。""" -from flask import Response, jsonify, request +from typing import Any + +from flask import jsonify, request +from flask.typing import ResponseReturnValue from ..application.app_context import AppContext from ..services import ( @@ -16,7 +19,7 @@ from .decorators import require_authentication class ProviderController: - """处理 provider 配置的增删改查、模型拉取与配置开关更新。""" + """处理 provider 配置、认证分组与相关管理 API。""" def __init__( self, @@ -36,6 +39,19 @@ class ProviderController: self._auth_service = auth_service self._register_routes() + @staticmethod + def _get_request_payload() -> dict[str, Any]: + payload = request.get_json(silent=True) + if isinstance(payload, dict): + return dict(payload) + return {} + + @staticmethod + def _coerce_name_list(value: Any) -> list[str]: + if not isinstance(value, list): + raise ValueError('Provider names must be a non-empty list') + return [str(item) for item in value] + def _register_routes(self) -> None: auth = require_authentication(self._auth_service) @@ -75,14 +91,14 @@ class ProviderController: auth(self.restore_auth_group_entry) ) - def get_providers(self) -> Response: + def get_providers(self) -> ResponseReturnValue: try: return jsonify(self._provider_service.list_providers()) except Exception as exc: self._logger.error('Error getting providers: %s', exc) return jsonify({'error': str(exc)}), 500 - def get_provider(self, name: str) -> Response: + def get_provider(self, name: str) -> ResponseReturnValue: try: provider = self._provider_service.get_provider(name) if provider is None: @@ -92,9 +108,9 @@ class ProviderController: self._logger.error('Error getting provider: %s', exc) return jsonify({'error': str(exc)}), 500 - def create_provider(self) -> Response: + def create_provider(self) -> ResponseReturnValue: try: - payload = request.get_json(silent=True) or {} + payload = self._get_request_payload() provider = self._provider_service.create_provider(payload) self._logger.info('Provider created: %s', provider.get('name')) return jsonify(provider), 201 @@ -104,9 +120,9 @@ class ProviderController: self._logger.error('Error creating provider: %s', exc) return jsonify({'error': str(exc)}), 500 - def update_provider(self, name: str) -> Response: + def update_provider(self, name: str) -> ResponseReturnValue: try: - payload = request.get_json(silent=True) or {} + payload = self._get_request_payload() provider = self._provider_service.update_provider(name, payload) self._logger.info('Provider updated: %s -> %s', name, provider.get('name')) return jsonify(provider) @@ -118,7 +134,7 @@ class ProviderController: self._logger.error('Error updating provider: %s', exc) return jsonify({'error': str(exc)}), 500 - def delete_provider(self, name: str) -> Response: + def delete_provider(self, name: str) -> ResponseReturnValue: try: self._provider_service.delete_provider(name) self._logger.info('Provider deleted: %s', name) @@ -131,7 +147,7 @@ class ProviderController: self._logger.error('Error deleting provider: %s', exc) return jsonify({'error': str(exc)}), 500 - def disable_provider(self, name: str) -> Response: + def disable_provider(self, name: str) -> ResponseReturnValue: try: provider = self._provider_service.set_provider_enabled(name, enabled=False) self._logger.info('Provider disabled: %s', name) @@ -144,7 +160,7 @@ class ProviderController: self._logger.error('Error disabling provider: %s', exc) return jsonify({'error': str(exc)}), 500 - def enable_provider(self, name: str) -> Response: + def enable_provider(self, name: str) -> ResponseReturnValue: try: provider = self._provider_service.set_provider_enabled(name, enabled=True) self._logger.info('Provider enabled: %s', name) @@ -157,11 +173,11 @@ class ProviderController: self._logger.error('Error enabling provider: %s', exc) return jsonify({'error': str(exc)}), 500 - def batch_providers(self) -> Response: + def batch_providers(self) -> ResponseReturnValue: try: - payload = request.get_json(silent=True) or {} + payload = self._get_request_payload() action = str(payload.get('action') or '').strip().lower() - names = payload.get('names') + names = self._coerce_name_list(payload.get('names')) if action == 'enable': result = self._provider_service.batch_set_provider_enabled(names, enabled=True) elif action == 'disable': @@ -181,14 +197,14 @@ class ProviderController: self._logger.error('Error applying provider batch action: %s', exc) return jsonify({'error': str(exc)}), 500 - def get_auth_groups(self) -> Response: + def get_auth_groups(self) -> ResponseReturnValue: try: return jsonify(self._auth_group_service.list_auth_groups()) except Exception as exc: self._logger.error('Error getting auth groups: %s', exc) return jsonify({'error': str(exc)}), 500 - def get_auth_group(self, name: str) -> Response: + def get_auth_group(self, name: str) -> ResponseReturnValue: try: auth_group = self._auth_group_service.get_auth_group(name) if auth_group is None: @@ -198,9 +214,9 @@ class ProviderController: self._logger.error('Error getting auth group: %s', exc) return jsonify({'error': str(exc)}), 500 - def create_auth_group(self) -> Response: + def create_auth_group(self) -> ResponseReturnValue: try: - payload = request.get_json(silent=True) or {} + payload = self._get_request_payload() auth_group = self._auth_group_service.create_auth_group(payload) self._logger.info('Auth group created: %s', auth_group.get('name')) return jsonify(auth_group), 201 @@ -210,10 +226,11 @@ class ProviderController: self._logger.error('Error creating auth group: %s', exc) return jsonify({'error': str(exc)}), 500 - def import_auth_group_entries(self) -> Response: + def import_auth_group_entries(self) -> ResponseReturnValue: try: - payload = request.get_json(silent=True) or {} - entries = self._auth_group_service.import_auth_entries(payload.get('yaml', '')) + payload = self._get_request_payload() + yaml_text = str(payload.get('yaml', '') or '') + entries = self._auth_group_service.import_auth_entries(yaml_text) return jsonify({'entries': entries}) except ValueError as exc: return jsonify({'error': str(exc)}), 400 @@ -221,9 +238,9 @@ class ProviderController: self._logger.error('Error importing auth group entries: %s', exc) return jsonify({'error': str(exc)}), 500 - def update_auth_group(self, name: str) -> Response: + def update_auth_group(self, name: str) -> ResponseReturnValue: try: - payload = request.get_json(silent=True) or {} + payload = self._get_request_payload() auth_group = self._auth_group_service.update_auth_group(name, payload) self._logger.info('Auth group updated: %s -> %s', name, auth_group.get('name')) return jsonify(auth_group) @@ -235,7 +252,7 @@ class ProviderController: self._logger.error('Error updating auth group: %s', exc) return jsonify({'error': str(exc)}), 500 - def delete_auth_group(self, name: str) -> Response: + def delete_auth_group(self, name: str) -> ResponseReturnValue: try: self._auth_group_service.delete_auth_group(name) self._logger.info('Auth group deleted: %s', name) @@ -248,7 +265,7 @@ class ProviderController: self._logger.error('Error deleting auth group: %s', exc) return jsonify({'error': str(exc)}), 500 - def get_auth_group_runtime(self, name: str) -> Response: + def get_auth_group_runtime(self, name: str) -> ResponseReturnValue: try: return jsonify(self._auth_group_service.get_auth_group_runtime(name)) except ValueError as exc: @@ -257,7 +274,7 @@ class ProviderController: self._logger.error('Error getting auth group runtime: %s', exc) return jsonify({'error': str(exc)}), 500 - def clear_auth_group_entry_cooldown(self, name: str, entry_id: str) -> Response: + def clear_auth_group_entry_cooldown(self, name: str, entry_id: str) -> ResponseReturnValue: try: self._auth_group_service.clear_entry_cooldown(name, entry_id) self._logger.info('Auth group entry cooldown cleared: %s/%s', name, entry_id) @@ -270,7 +287,7 @@ class ProviderController: self._logger.error('Error clearing auth group entry cooldown: %s', exc) return jsonify({'error': str(exc)}), 500 - def disable_auth_group_entry(self, name: str, entry_id: str) -> Response: + def disable_auth_group_entry(self, name: str, entry_id: str) -> ResponseReturnValue: try: self._auth_group_service.set_entry_disabled(name, entry_id, disabled=True) self._logger.info('Auth group entry disabled: %s/%s', name, entry_id) @@ -283,7 +300,7 @@ class ProviderController: self._logger.error('Error disabling auth group entry: %s', exc) return jsonify({'error': str(exc)}), 500 - def enable_auth_group_entry(self, name: str, entry_id: str) -> Response: + def enable_auth_group_entry(self, name: str, entry_id: str) -> ResponseReturnValue: try: self._auth_group_service.set_entry_disabled(name, entry_id, disabled=False) self._logger.info('Auth group entry enabled: %s/%s', name, entry_id) @@ -296,7 +313,7 @@ class ProviderController: self._logger.error('Error enabling auth group entry: %s', exc) return jsonify({'error': str(exc)}), 500 - def reset_auth_group_entry_minute_usage(self, name: str, entry_id: str) -> Response: + def reset_auth_group_entry_minute_usage(self, name: str, entry_id: str) -> ResponseReturnValue: try: self._auth_group_service.reset_entry_minute_usage(name, entry_id) self._logger.info('Auth group entry minute usage reset: %s/%s', name, entry_id) @@ -309,7 +326,7 @@ class ProviderController: self._logger.error('Error resetting auth group entry minute usage: %s', exc) return jsonify({'error': str(exc)}), 500 - def reset_auth_group_entry_runtime(self, name: str, entry_id: str) -> Response: + def reset_auth_group_entry_runtime(self, name: str, entry_id: str) -> ResponseReturnValue: try: self._auth_group_service.reset_entry_runtime(name, entry_id) self._logger.info('Auth group entry runtime reset: %s/%s', name, entry_id) @@ -322,7 +339,7 @@ class ProviderController: self._logger.error('Error resetting auth group entry runtime: %s', exc) return jsonify({'error': str(exc)}), 500 - def restore_auth_group_entry(self, name: str, entry_id: str) -> Response: + def restore_auth_group_entry(self, name: str, entry_id: str) -> ResponseReturnValue: try: self._auth_group_service.restore_entry(name, entry_id) self._logger.info('Auth group entry restored: %s/%s', name, entry_id) @@ -335,7 +352,7 @@ class ProviderController: self._logger.error('Error restoring auth group entry: %s', exc) return jsonify({'error': str(exc)}), 500 - def fetch_models(self) -> Response: + def fetch_models(self) -> ResponseReturnValue: try: result = self._model_discovery_service.fetch_models_preview( api=request.args.get('api', ''), @@ -356,9 +373,9 @@ class ProviderController: self._logger.error('Error fetching provider models: %s', exc) return jsonify({'error': str(exc)}), 500 - def update_chat_whitelist(self) -> Response: + def update_chat_whitelist(self) -> ResponseReturnValue: try: - payload = request.get_json(silent=True) or {} + payload = self._get_request_payload() enabled = self._settings_service.update_chat_whitelist_enabled(payload.get('enabled')) self._logger.info('Chat whitelist updated: enabled=%s', enabled) return jsonify({'enabled': enabled}) diff --git a/src/presentation/proxy_controller.py b/src/presentation/proxy_controller.py index d2d446a..d12b8b1 100644 --- a/src/presentation/proxy_controller.py +++ b/src/presentation/proxy_controller.py @@ -7,6 +7,7 @@ from __future__ import annotations from typing import Any, Dict, Iterable, Optional from flask import Response, jsonify, request +from flask.typing import ResponseReturnValue from ..application.app_context import AppContext from ..config import ConfigManager, ProviderManager @@ -56,7 +57,7 @@ class ProxyController: *, error_format: str, ) -> tuple[Optional[Dict[str, Any]], Optional[tuple[Response, int]]]: - """在开启白名单时解析当前请求对应用户。""" + """在启用白名单时解析当前请求对应的用户。""" if not self._is_whitelist_required(): return None, None @@ -151,21 +152,21 @@ class ProxyController: return normalized[0] return ", ".join(normalized) - def chat_completions(self) -> Response: + def chat_completions(self) -> ResponseReturnValue: return self._proxy_completion_request( route_name="chat_completions", expected_target_formats=("openai_chat",), inspect_stream_usage=True, ) - def responses(self) -> Response: + def responses(self) -> ResponseReturnValue: return self._proxy_completion_request( route_name="responses", expected_target_formats=("openai_responses", "codex"), inspect_stream_usage=False, ) - def messages(self) -> Response: + def messages(self) -> ResponseReturnValue: return self._proxy_completion_request( route_name="messages", expected_target_formats=("claude_chat",), @@ -179,7 +180,7 @@ class ProxyController: expected_target_formats: Iterable[str], inspect_stream_usage: bool, error_format: Optional[str] = None, - ) -> Response: + ) -> ResponseReturnValue: normalized_expected_target_formats = tuple( str(item or "").strip().lower() for item in expected_target_formats if str(item or "").strip() ) @@ -196,10 +197,10 @@ class ProxyController: if denial_response is not None: return denial_response - request_data = request.get_json(silent=True) - if request_data is None: - request_data = {} - if not isinstance(request_data, dict): + raw_request_data = request.get_json(silent=True) + if raw_request_data is None: + request_data: Dict[str, Any] = {} + elif not isinstance(raw_request_data, dict): self._logger.warning("Proxy rejected: request body is not a JSON object route=%s", route_name) return self._error_response( "Request body must be a JSON object", @@ -208,9 +209,11 @@ class ProxyController: code="invalid_request_body", error_format=resolved_error_format, ) + else: + request_data = dict(raw_request_data) - request_data = dict(request_data) - if "model" not in request_data: + model_name_value = request_data.get("model") + if not isinstance(model_name_value, str) or not model_name_value.strip(): self._logger.warning("Proxy rejected: missing model in request body route=%s", route_name) return self._error_response( "Missing 'model' in request body", @@ -220,7 +223,7 @@ class ProxyController: error_format=resolved_error_format, ) - model_name = request_data["model"] + model_name = model_name_value.strip() provider = self._provider_manager.get_provider_for_model(model_name) if not provider: self._logger.warning("Proxy rejected: unknown model=%r route=%s", model_name, route_name) @@ -390,7 +393,7 @@ class ProxyController: error_format=resolved_error_format, ) - def list_models(self) -> Response: + def list_models(self) -> ResponseReturnValue: try: client_ip = normalize_ip(request.remote_addr) user, denial_response = self._get_authorized_user_for_request( @@ -453,3 +456,6 @@ class ProxyController: "upgrade", } return {k: v for k, v in headers.items() if k.lower() not in excluded} + + + diff --git a/src/presentation/user_controller.py b/src/presentation/user_controller.py index 4666f71..f1e5201 100644 --- a/src/presentation/user_controller.py +++ b/src/presentation/user_controller.py @@ -2,7 +2,10 @@ # -*- coding: utf-8 -*- """用户控制器。""" -from flask import Response, jsonify, request +from typing import Any + +from flask import jsonify, request +from flask.typing import ResponseReturnValue from ..application.app_context import AppContext from ..services import AuthenticationService, UserService @@ -22,8 +25,14 @@ class UserController: self._auth_service = auth_service self._register_routes() + @staticmethod + def _get_request_payload() -> dict[str, Any]: + payload = request.get_json(silent=True) + if isinstance(payload, dict): + return dict(payload) + return {} + def _register_routes(self) -> None: - """注册用户管理路由。""" auth = require_authentication(self._auth_service) self._app.route('/api/users', methods=['GET'])(auth(self.get_users)) @@ -34,14 +43,13 @@ class UserController: self._app.route('/api/users/', methods=['DELETE'])(auth(self.delete_user)) self._app.route('/api/users//toggle', methods=['POST'])(auth(self.toggle_user)) - def get_users(self) -> Response: - """分页查询用户列表。""" + def get_users(self) -> ResponseReturnValue: try: page = request.args.get('page', 1, type=int) page_size = request.args.get('page_size', 50, type=int) keyword = (request.args.get('keyword', '', type=str) or '').strip() self._logger.debug( - "List users requested: page=%s, page_size=%s, keyword=%r", + 'List users requested: page=%s, page_size=%s, keyword=%r', page, page_size, keyword, @@ -50,71 +58,84 @@ class UserController: users = self._user_service.get_users(page=page, page_size=page_size, keyword=keyword) total = self._user_service.get_total_users_count(keyword=keyword) - return jsonify({ - 'users': users, - 'total': total, - 'page': page, - 'page_size': page_size, - 'total_pages': (total + page_size - 1) // page_size, - 'keyword': keyword, - 'available_models': self._user_service.get_available_models(), - }) + return jsonify( + { + 'users': users, + 'total': total, + 'page': page, + 'page_size': page_size, + 'total_pages': (total + page_size - 1) // page_size, + 'keyword': keyword, + 'available_models': self._user_service.get_available_models(), + } + ) except Exception as exc: - self._logger.error(f'Error getting users: {exc}') + self._logger.error('Error getting users: %s', exc) return jsonify({'error': str(exc)}), 500 - def get_user(self, user_id: int) -> Response: - """查询单个用户。""" + def get_user(self, user_id: int) -> ResponseReturnValue: try: user = self._user_service.get_user_by_id(user_id) if not user: - self._logger.warning(f"Get user failed: user_id={user_id} not found") + self._logger.warning('Get user failed: user_id=%s not found', user_id) return jsonify({'error': 'User not found'}), 404 return jsonify(user) except Exception as exc: - self._logger.error(f'Error getting user: {exc}') + self._logger.error('Error getting user: %s', exc) return jsonify({'error': str(exc)}), 500 - def create_user(self) -> Response: - """创建用户。""" + def create_user(self) -> ResponseReturnValue: try: - data = request.get_json(silent=True) or {} + data = self._get_request_payload() username = data.get('username') ip_address = data.get('ip_address') - if not username: - self._logger.warning("Create user rejected: username is empty") + if not isinstance(username, str) or not username.strip(): + self._logger.warning('Create user rejected: username is empty') return jsonify({'error': 'Username is required'}), 400 - if not ip_address: - self._logger.warning("Create user rejected: ip_address is empty") + if not isinstance(ip_address, str) or not ip_address.strip(): + self._logger.warning('Create user rejected: ip_address is empty') return jsonify({'error': 'IP address is required'}), 400 + normalized_username = username.strip() normalized_ip = normalize_ip(ip_address) if not is_valid_ip(normalized_ip): - self._logger.warning(f"Create user rejected: invalid ip_address={ip_address!r}") + self._logger.warning('Create user rejected: invalid ip_address=%r', ip_address) return jsonify({'error': 'Invalid IP address'}), 400 - user_id = self._user_service.create_user(username, normalized_ip) + user_id = self._user_service.create_user(normalized_username, normalized_ip) if not user_id: - self._logger.warning(f"Create user failed: username={username!r}, ip={normalized_ip}") + self._logger.warning('Create user failed: username=%r, ip=%s', normalized_username, normalized_ip) return jsonify({'error': 'Failed to create user or IP already exists'}), 400 - self._logger.info(f"Create user succeeded: user_id={user_id}, username={username!r}, ip={normalized_ip}") + self._logger.info( + 'Create user succeeded: user_id=%s, username=%r, ip=%s', + user_id, + normalized_username, + normalized_ip, + ) return jsonify({'id': user_id, 'message': 'User created successfully'}), 201 except Exception as exc: - self._logger.error(f'Error creating user: {exc}') + self._logger.error('Error creating user: %s', exc) return jsonify({'error': str(exc)}), 500 - def update_user(self, user_id: int) -> Response: - """更新用户信息。""" + def update_user(self, user_id: int) -> ResponseReturnValue: try: - data = request.get_json(silent=True) + data = self._get_request_payload() if not data: - self._logger.warning(f"Update user rejected: no payload, user_id={user_id}") + self._logger.warning('Update user rejected: no payload, user_id=%s', user_id) return jsonify({'error': 'No data provided'}), 400 - username = data.get('username') - ip_address = data.get('ip_address') + username_value = data.get('username') + if username_value is not None and not isinstance(username_value, str): + return jsonify({'error': 'Username must be a string'}), 400 + username = username_value + + ip_address_value = data.get('ip_address') + if ip_address_value is not None and not isinstance(ip_address_value, str): + return jsonify({'error': 'IP address must be a string'}), 400 + ip_address = ip_address_value + whitelist_access_enabled = data.get('whitelist_access_enabled') model_permissions_provided = 'model_permissions' in data model_permissions = data.get('model_permissions') @@ -123,7 +144,7 @@ class UserController: if ip_address is not None: normalized_ip = normalize_ip(ip_address) if not is_valid_ip(normalized_ip): - self._logger.warning(f"Update user rejected: invalid ip={ip_address!r}, user_id={user_id}") + self._logger.warning('Update user rejected: invalid ip=%r, user_id=%s', ip_address, user_id) return jsonify({'error': 'Invalid IP address'}), 400 success = self._user_service.update_user( @@ -135,21 +156,20 @@ class UserController: model_permissions=model_permissions, ) if not success: - self._logger.warning(f"Update user failed: user_id={user_id}") + self._logger.warning('Update user failed: user_id=%s', user_id) return jsonify({'error': 'Failed to update user'}), 400 - self._logger.info(f"Update user succeeded: user_id={user_id}") + self._logger.info('Update user succeeded: user_id=%s', user_id) return jsonify({'message': 'User updated successfully'}) except ValueError as exc: return jsonify({'error': str(exc)}), 400 except Exception as exc: - self._logger.error(f'Error updating user: {exc}') + self._logger.error('Error updating user: %s', exc) return jsonify({'error': str(exc)}), 500 - def batch_users(self) -> Response: - """执行用户批量操作。""" + def batch_users(self) -> ResponseReturnValue: try: - payload = request.get_json(silent=True) or {} + payload = self._get_request_payload() action = str(payload.get('action') or '').strip().lower() if action != 'set_model_permissions': raise ValueError(f'Unsupported user batch action: {action or ""}') @@ -169,31 +189,29 @@ class UserController: status_code = 404 if 'not found' in message.lower() else 400 return jsonify({'error': message}), status_code except Exception as exc: - self._logger.error(f'Error applying user batch action: {exc}') + self._logger.error('Error applying user batch action: %s', exc) return jsonify({'error': str(exc)}), 500 - def delete_user(self, user_id: int) -> Response: - """删除用户。""" + def delete_user(self, user_id: int) -> ResponseReturnValue: try: if not self._user_service.delete_user(user_id): - self._logger.warning(f"Delete user failed: user_id={user_id}") + self._logger.warning('Delete user failed: user_id=%s', user_id) return jsonify({'error': 'Failed to delete user'}), 400 - self._logger.info(f"Delete user succeeded: user_id={user_id}") + self._logger.info('Delete user succeeded: user_id=%s', user_id) return jsonify({'message': 'User deleted successfully'}) except Exception as exc: - self._logger.error(f'Error deleting user: {exc}') + self._logger.error('Error deleting user: %s', exc) return jsonify({'error': str(exc)}), 500 - def toggle_user(self, user_id: int) -> Response: - """切换用户白名单状态。""" + def toggle_user(self, user_id: int) -> ResponseReturnValue: try: if not self._config_manager.is_chat_whitelist_enabled(): return jsonify({'error': 'Whitelist control is disabled'}), 400 if not self._user_service.toggle_user_status(user_id): - self._logger.warning(f"Toggle user whitelist failed: user_id={user_id}") + self._logger.warning('Toggle user whitelist failed: user_id=%s', user_id) return jsonify({'error': 'Failed to toggle user status'}), 400 - self._logger.info(f"Toggle user whitelist succeeded: user_id={user_id}") + self._logger.info('Toggle user whitelist succeeded: user_id=%s', user_id) return jsonify({'message': 'User status toggled successfully'}) except Exception as exc: - self._logger.error(f'Error toggling user status: {exc}') + self._logger.error('Error toggling user status: %s', exc) return jsonify({'error': str(exc)}), 500 diff --git a/src/presentation/web_controller.py b/src/presentation/web_controller.py index 9b1b38a..5e69222 100644 --- a/src/presentation/web_controller.py +++ b/src/presentation/web_controller.py @@ -2,7 +2,8 @@ # -*- coding: utf-8 -*- """页面与统计控制器。""" -from flask import Response, jsonify, render_template, request +from flask import jsonify, render_template, request +from flask.typing import ResponseReturnValue from ..application.app_context import AppContext from ..services import AuthenticationService, LogService @@ -24,27 +25,27 @@ class WebController: def _register_routes(self) -> None: auth = require_authentication(self._auth_service) - self._app.route("/")(auth(self.index)) - self._app.route("/users")(auth(self.users_page)) - self._app.route("/providers")(auth(self.providers_page)) + self._app.route('/')(auth(self.index)) + self._app.route('/users')(auth(self.users_page)) + self._app.route('/providers')(auth(self.providers_page)) - self._app.route("/api/statistics", methods=["GET"])(auth(self.get_statistics)) - self._app.route("/api/request-logs", methods=["GET"])(auth(self.get_request_logs)) - self._app.route("/api/usernames", methods=["GET"])(auth(self.get_usernames)) - self._app.route("/api/request-models", methods=["GET"])(auth(self.get_request_models)) + self._app.route('/api/statistics', methods=['GET'])(auth(self.get_statistics)) + self._app.route('/api/request-logs', methods=['GET'])(auth(self.get_request_logs)) + self._app.route('/api/usernames', methods=['GET'])(auth(self.get_usernames)) + self._app.route('/api/request-models', methods=['GET'])(auth(self.get_request_models)) def index(self) -> str: return render_template( - "index.html", - active_page="index", + 'index.html', + active_page='index', current_username=self._get_current_username(), auth_enabled=self._auth_service.is_auth_enabled(), ) def users_page(self) -> str: return render_template( - "users.html", - active_page="users", + 'users.html', + active_page='users', chat_whitelist_enabled=self._config_manager.is_chat_whitelist_enabled(), current_username=self._get_current_username(), auth_enabled=self._auth_service.is_auth_enabled(), @@ -52,8 +53,8 @@ class WebController: def providers_page(self) -> str: return render_template( - "providers.html", - active_page="providers", + 'providers.html', + active_page='providers', chat_whitelist_enabled=self._config_manager.is_chat_whitelist_enabled(), current_username=self._get_current_username(), auth_enabled=self._auth_service.is_auth_enabled(), @@ -61,66 +62,66 @@ class WebController: def _get_current_username(self) -> str: if not self._auth_service.is_auth_enabled(): - return "" + return '' - session_token = request.cookies.get("session_token") - return self._auth_service.get_session_username(session_token) or "" + session_token = request.cookies.get('session_token') + return self._auth_service.get_session_username(session_token) or '' - def get_statistics(self) -> Response: + def get_statistics(self) -> ResponseReturnValue: try: self._logger.debug( - "Statistics queried: start_date=%s end_date=%s username=%s request_model=%s", - request.args.get("start_date"), - request.args.get("end_date"), - request.args.get("username"), - request.args.get("request_model"), + 'Statistics queried: start_date=%s end_date=%s username=%s request_model=%s', + request.args.get('start_date'), + request.args.get('end_date'), + request.args.get('username'), + request.args.get('request_model'), ) stats = self._log_service.get_statistics( - request.args.get("start_date"), - request.args.get("end_date"), - request.args.get("username"), - request.args.get("request_model"), + request.args.get('start_date'), + request.args.get('end_date'), + request.args.get('username'), + request.args.get('request_model'), ) return jsonify(stats) except Exception as exc: - self._logger.error(f"Error getting statistics: {exc}") - return jsonify({"error": str(exc)}), 500 + self._logger.error('Error getting statistics: %s', exc) + return jsonify({'error': str(exc)}), 500 - def get_request_logs(self) -> Response: + def get_request_logs(self) -> ResponseReturnValue: try: - page = max(int(request.args.get("page", 1)), 1) - page_size = min(max(int(request.args.get("page_size", 50)), 1), 200) - self._logger.debug(f"Request logs queried: page={page}, page_size={page_size}") + page = max(int(request.args.get('page', 1)), 1) + page_size = min(max(int(request.args.get('page_size', 50)), 1), 200) + self._logger.debug('Request logs queried: page=%s, page_size=%s', page, page_size) logs = self._log_service.get_request_logs( page, page_size, - request.args.get("start_date"), - request.args.get("end_date"), - request.args.get("username"), - request.args.get("request_model"), + request.args.get('start_date'), + request.args.get('end_date'), + request.args.get('username'), + request.args.get('request_model'), ) return jsonify(logs) except ValueError: - return jsonify({"error": "page and page_size must be integers"}), 400 + return jsonify({'error': 'page and page_size must be integers'}), 400 except Exception as exc: - self._logger.error(f"Error getting request logs: {exc}") - return jsonify({"error": str(exc)}), 500 + self._logger.error('Error getting request logs: %s', exc) + return jsonify({'error': str(exc)}), 500 - def get_usernames(self) -> Response: + def get_usernames(self) -> ResponseReturnValue: try: usernames = self._log_service.get_unique_usernames() - self._logger.debug(f"Usernames queried: count={len(usernames)}") + self._logger.debug('Usernames queried: count=%s', len(usernames)) return jsonify(usernames) except Exception as exc: - self._logger.error(f"Error getting usernames: {exc}") - return jsonify({"error": str(exc)}), 500 + self._logger.error('Error getting usernames: %s', exc) + return jsonify({'error': str(exc)}), 500 - def get_request_models(self) -> Response: + def get_request_models(self) -> ResponseReturnValue: try: models = self._log_service.get_unique_request_models() - self._logger.debug(f"Request models queried: count={len(models)}") + self._logger.debug('Request models queried: count=%s', len(models)) return jsonify(models) except Exception as exc: - self._logger.error(f"Error getting request models: {exc}") - return jsonify({"error": str(exc)}), 500 + self._logger.error('Error getting request models: %s', exc) + return jsonify({'error': str(exc)}), 500 diff --git a/src/repositories/user_repository.py b/src/repositories/user_repository.py index 410a590..6b30349 100644 --- a/src/repositories/user_repository.py +++ b/src/repositories/user_repository.py @@ -158,8 +158,8 @@ class UserRepository: model_permissions: Optional[str] = None, ) -> bool: """更新用户记录。""" - updates = [] - params = [] + updates: list[str] = [] + params: list[Any] = [] if username is not None: updates.append("username = ?") diff --git a/src/services/auth_group_service.py b/src/services/auth_group_service.py index 8177929..714026b 100644 --- a/src/services/auth_group_service.py +++ b/src/services/auth_group_service.py @@ -218,6 +218,7 @@ class AuthGroupService: @staticmethod def _normalize_import_payload(payload: Any) -> List[Mapping[str, Any]]: + entries: Any if isinstance(payload, list): entries = payload elif isinstance(payload, Mapping): diff --git a/src/services/user_service.py b/src/services/user_service.py index ad591b4..268998e 100644 --- a/src/services/user_service.py +++ b/src/services/user_service.py @@ -231,10 +231,11 @@ class UserService: try: available_models = self._get_available_model_names() users = self._repository.get(page=page, page_size=page_size, keyword=keyword) - return [ + decorated_users = [ self._decorate_user(user, available_models=available_models) for user in users ] + return [user for user in decorated_users if user is not None] except Exception as exc: self._logger.error(f"Failed to get users: {exc}") return [] diff --git a/src/translators/claude_bridge.py b/src/translators/claude_bridge.py index 90bb3d8..4a84dc3 100644 --- a/src/translators/claude_bridge.py +++ b/src/translators/claude_bridge.py @@ -324,7 +324,7 @@ def convert_openai_chat_response_to_claude( if not isinstance(payload, dict): return payload - message = {} + message: Dict[str, Any] = {} finish_reason = None for choice in payload.get("choices") or []: if not isinstance(choice, dict): @@ -350,13 +350,14 @@ def convert_openai_chat_response_to_claude( "stop_sequence": None, } if isinstance(payload.get("usage"), dict): - response["usage"] = { + usage_payload: Dict[str, Any] = { "input_tokens": int(payload["usage"].get("prompt_tokens") or 0), "output_tokens": int(payload["usage"].get("completion_tokens") or 0), } + response["usage"] = usage_payload cached_tokens = int(((payload["usage"].get("prompt_tokens_details") or {}).get("cached_tokens")) or 0) if cached_tokens > 0: - response["usage"]["cache_read_input_tokens"] = cached_tokens + usage_payload["cache_read_input_tokens"] = cached_tokens return response diff --git a/src/translators/registry.py b/src/translators/registry.py index f65488a..63c5938 100644 --- a/src/translators/registry.py +++ b/src/translators/registry.py @@ -26,8 +26,11 @@ from .responses_bridge import ( class Translator(Protocol): - source_format: str - target_format: str + @property + def source_format(self) -> str: ... + + @property + def target_format(self) -> str: ... def translate_request(self, model_name: str, body: Dict[str, Any], stream: bool) -> Dict[str, Any]: ... @@ -1047,7 +1050,7 @@ def build_default_translator_registry() -> TranslatorRegistry: codex_passthrough = CodexPassthroughTranslator() openai_chat_codex = OpenAIChatCodexTranslator() - for translator in ( + builtin_translators: tuple[Translator, ...] = ( openai_chat, openai_responses, openai_responses_passthrough, @@ -1058,17 +1061,19 @@ def build_default_translator_registry() -> TranslatorRegistry: codex_chat, codex_passthrough, openai_chat_codex, - ): + ) + for translator in builtin_translators: registry.register(translator) - for translator in ( + composed_translators: tuple[Translator, ...] = ( ComposedTranslator("openai_responses", "claude_chat", openai_responses, openai_chat_claude), ComposedTranslator("openai_responses", "codex", openai_responses, openai_chat_codex), ComposedTranslator("claude_chat", "openai_responses", claude_chat, openai_chat_responses), ComposedTranslator("claude_chat", "codex", claude_chat, openai_chat_codex), ComposedTranslator("codex", "openai_responses", codex_chat, openai_chat_responses), ComposedTranslator("codex", "claude_chat", codex_chat, openai_chat_claude), - ): + ) + for translator in composed_translators: registry.register(translator) return registry @@ -1486,7 +1491,7 @@ def _extract_claude_usage(payload: Any) -> Dict[str, Any]: return {} prompt_tokens = int(payload.get("input_tokens") or 0) completion_tokens = int(payload.get("output_tokens") or 0) - usage = { + usage: Dict[str, Any] = { "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "total_tokens": prompt_tokens + completion_tokens, diff --git a/src/translators/responses_bridge.py b/src/translators/responses_bridge.py index 8cee87a..671c281 100644 --- a/src/translators/responses_bridge.py +++ b/src/translators/responses_bridge.py @@ -625,7 +625,7 @@ def convert_openai_chat_response_to_responses( if not isinstance(payload, dict): return payload - message = {} + message: Dict[str, Any] = {} finish_reason = None for choice in payload.get("choices") or []: if not isinstance(choice, dict): diff --git a/src/utils/compat.py b/src/utils/compat.py index b7a827d..fe76637 100644 --- a/src/utils/compat.py +++ b/src/utils/compat.py @@ -1,46 +1,59 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""Compatibility helpers for supported Python runtimes.""" +"""兼容不同 Python 运行时的辅助工具。""" + +from __future__ import annotations import sys from dataclasses import dataclass as _dataclass from enum import Enum -from typing import Any +from typing import TYPE_CHECKING, Any, Callable, TypeVar, overload -try: +if TYPE_CHECKING: from enum import StrEnum -except ImportError: - class StrEnum(str, Enum): - """Fallback StrEnum for Python versions before 3.11.""" - - pass - -try: - from typing import Literal -except ImportError: + from typing import Literal, Protocol + from typing import dataclass_transform as _dataclass_transform +else: try: - from typing_extensions import Literal # type: ignore + from typing import Literal, Protocol except ImportError: - class _LiteralCompat(object): - def __getitem__(self, _item): - return Any + from typing_extensions import Literal, Protocol # type: ignore - Literal = _LiteralCompat() - -try: - from typing import Protocol -except ImportError: try: - from typing_extensions import Protocol # type: ignore + from typing import dataclass_transform as _dataclass_transform except ImportError: - class Protocol(object): - """Fallback Protocol base when typing_extensions is unavailable.""" + try: + from typing_extensions import dataclass_transform as _dataclass_transform # type: ignore + except ImportError: + def _dataclass_transform(*_args: Any, **_kwargs: Any) -> Callable[[Any], Any]: + def decorator(func: Any) -> Any: + return func + + return decorator + + try: + from enum import StrEnum + except ImportError: + class StrEnum(str, Enum): + """Python 3.11 之前的 StrEnum 兼容实现。""" pass -def dataclass(_cls=None, **kwargs): - """Mirror dataclasses.dataclass while skipping unsupported keywords.""" +T = TypeVar("T") + + +@_dataclass_transform() +@overload +def dataclass(_cls: type[T], **kwargs: Any) -> type[T]: ... + + +@overload +def dataclass(_cls: None = None, **kwargs: Any) -> Callable[[type[T]], type[T]]: ... + + +def dataclass(_cls: type[T] | None = None, **kwargs: Any) -> Any: + """兼容旧版本参数的 dataclass 包装器。""" if sys.version_info < (3, 10): kwargs.pop("slots", None) diff --git a/tests/test_hook_contracts.py b/tests/test_hook_contracts.py index 9180082..e1283b8 100644 --- a/tests/test_hook_contracts.py +++ b/tests/test_hook_contracts.py @@ -1,6 +1,7 @@ import sys import unittest from pathlib import Path +from typing import Any, cast sys.path.insert(0, str(Path(__file__).resolve().parents[1])) @@ -9,17 +10,17 @@ from src.hooks import BaseHook, HookContext class FakeLogger: - def info(self, msg: str, *args) -> None: - del msg, args + def info(self, msg: object, *args: object, **kwargs: object) -> None: + del msg, args, kwargs - def warning(self, msg: str, *args) -> None: - del msg, args + def warning(self, msg: object, *args: object, **kwargs: object) -> None: + del msg, args, kwargs - def error(self, msg: str, *args) -> None: - del msg, args + def error(self, msg: object, *args: object, **kwargs: object) -> None: + del msg, args, kwargs - def debug(self, msg: str, *args) -> None: - del msg, args + def debug(self, msg: object, *args: object, **kwargs: object) -> None: + del msg, args, kwargs class LegacyOnlyHook: @@ -83,7 +84,7 @@ class HookContractsTests(unittest.TestCase): self.assertEqual(response_body, hook.response_guard(ctx, response_body)) def test_provider_does_not_call_legacy_hook_methods(self) -> None: - provider = LLMProvider(name="demo", api="https://example.com", hook=LegacyOnlyHook()) + provider = LLMProvider(name="demo", api="https://example.com", hook=cast(Any, LegacyOnlyHook())) ctx = self._ctx() request_body = {"messages": [{"role": "user", "content": "hello"}]} response_body = {"message": "ok"} @@ -101,7 +102,7 @@ class HookContractsTests(unittest.TestCase): def test_none_from_guard_keeps_original_body(self) -> None: provider = LLMProvider(name="demo", api="https://example.com", hook=NoneReturningGuard()) ctx = self._ctx() - request_body = {"messages": []} + request_body: dict[str, Any] = {"messages": []} response_body = {"message": "ok"} self.assertEqual(request_body, provider.apply_request_guard(ctx, request_body)) diff --git a/tests/test_proxy_error_handling.py b/tests/test_proxy_error_handling.py index ea405ea..3dd901a 100644 --- a/tests/test_proxy_error_handling.py +++ b/tests/test_proxy_error_handling.py @@ -127,8 +127,8 @@ class StubProxyService: class RecordingProxyService: def __init__(self, proxy_result) -> None: self._proxy_result = proxy_result - self.last_args = None - self.last_kwargs = None + self.last_args: tuple[object, ...] | None = None + self.last_kwargs: dict[str, object] | None = None def proxy_request(self, *args, **kwargs): self.last_args = args @@ -455,6 +455,8 @@ class ProxyControllerErrorFormatTests(unittest.TestCase): ) self.assertEqual(200, response.status_code) + self.assertIsNotNone(proxy_service.last_kwargs) + assert proxy_service.last_kwargs is not None self.assertEqual("openai_chat", proxy_service.last_kwargs["resolved_target_format"]) def test_responses_route_allows_codex_target(self) -> None: diff --git a/tests/test_user_model_permissions.py b/tests/test_user_model_permissions.py index ae637b7..47ce9ef 100644 --- a/tests/test_user_model_permissions.py +++ b/tests/test_user_model_permissions.py @@ -3,10 +3,13 @@ import tempfile import unittest from copy import deepcopy from pathlib import Path -from types import SimpleNamespace +from typing import Any, cast + +from flask import Flask sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from src.application.app_context import AppContext from src.repositories import UserRepository from src.services import UserService from src.utils.database import create_connection_factory @@ -49,9 +52,11 @@ class UserModelPermissionTests(unittest.TestCase): } ] } - self._ctx = SimpleNamespace( + self._ctx = AppContext( logger=DummyLogger(), - config_manager=FakeConfigManager(self._config), + config_manager=cast(Any, FakeConfigManager(self._config)), + root_path=Path(self._tempdir.name), + flask_app=Flask(__name__), ) self._repository = UserRepository(self._get_connection) self._service = UserService(self._ctx, self._repository) @@ -61,13 +66,19 @@ class UserModelPermissionTests(unittest.TestCase): def test_user_repository_defaults_model_permissions_to_wildcard(self) -> None: user_id = self._repository.create("alice", "127.0.0.1") + self.assertIsNotNone(user_id) + assert user_id is not None user = self._repository.get_by_id(user_id) + self.assertIsNotNone(user) + assert user is not None self.assertEqual("*", user["model_permissions"]) def test_update_user_model_permissions_supports_explicit_models(self) -> None: user_id = self._repository.create("alice", "127.0.0.1") + self.assertIsNotNone(user_id) + assert user_id is not None updated = self._service.update_user( user_id, @@ -77,12 +88,16 @@ class UserModelPermissionTests(unittest.TestCase): user = self._service.get_user_by_id(user_id) self.assertTrue(updated) + self.assertIsNotNone(user) + assert user is not None self.assertEqual(["demo/gpt-4.1-mini"], user["model_permissions"]) self.assertFalse(self._service.can_user_access_model(user, "demo/gpt-4.1")) self.assertTrue(self._service.can_user_access_model(user, "demo/gpt-4.1-mini")) def test_sync_model_permissions_prunes_deleted_models_from_explicit_users(self) -> None: user_id = self._repository.create("alice", "127.0.0.1") + self.assertIsNotNone(user_id) + assert user_id is not None self._service.update_user( user_id, model_permissions_provided=True, @@ -94,11 +109,17 @@ class UserModelPermissionTests(unittest.TestCase): user = self._service.get_user_by_id(user_id) self.assertEqual(1, updated_count) + self.assertIsNotNone(user) + assert user is not None self.assertEqual(["demo/gpt-4.1"], user["model_permissions"]) def test_batch_update_model_permissions_updates_multiple_users(self) -> None: alice_id = self._repository.create("alice", "127.0.0.1") bob_id = self._repository.create("bob", "127.0.0.2") + self.assertIsNotNone(alice_id) + self.assertIsNotNone(bob_id) + assert alice_id is not None + assert bob_id is not None result = self._service.batch_update_model_permissions( [alice_id, bob_id], @@ -107,6 +128,10 @@ class UserModelPermissionTests(unittest.TestCase): alice = self._service.get_user_by_id(alice_id) bob = self._service.get_user_by_id(bob_id) + self.assertIsNotNone(alice) + self.assertIsNotNone(bob) + assert alice is not None + assert bob is not None self.assertEqual(2, result["count"]) self.assertEqual(["demo/gpt-4.1-mini"], result["model_permissions"]) self.assertEqual(["demo/gpt-4.1-mini"], alice["model_permissions"])