From b7579c539dcd197ec852549b7e50de56e28f1923 Mon Sep 17 00:00:00 2001 From: GitHub Actions Date: Sat, 4 Apr 2026 03:24:15 +0000 Subject: [PATCH] style: auto-format with ruff [skip ci] --- backend/package/yuxi/services/oidc_service.py | 46 +++++++++---------- 1 file changed, 21 insertions(+), 25 deletions(-) diff --git a/backend/package/yuxi/services/oidc_service.py b/backend/package/yuxi/services/oidc_service.py index e1d3d55c..46f74cb5 100644 --- a/backend/package/yuxi/services/oidc_service.py +++ b/backend/package/yuxi/services/oidc_service.py @@ -8,7 +8,7 @@ import os import secrets import time import urllib.parse -from typing import Any, Optional +from typing import Any from urllib.parse import urlencode import httpx @@ -17,14 +17,14 @@ from fastapi.responses import RedirectResponse from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.exc import IntegrityError - -from server.utils.auth_utils import AuthUtils -from server.utils.common_utils import log_operation from yuxi.repositories.user_repository import UserRepository from yuxi.storage.postgres.models_business import Department, User from yuxi.utils.datetime_utils import utc_now_naive from yuxi.utils.logging_config import logger +from server.utils.auth_utils import AuthUtils +from server.utils.common_utils import log_operation + # 前端 OIDC 回调路由路径（与 web/src/router/index.js 中的路由保持一致） FRONTEND_CALLBACK_PATH = "/auth/oidc/callback" # 登录页路径 @@ -106,11 +106,11 @@ class OIDCProviderMetadata: """OIDC Provider 元数据""" def __init__(self): - self.authorization_endpoint: Optional[str] = None - self.token_endpoint: Optional[str] = None - self.userinfo_endpoint: Optional[str] = None - self.end_session_endpoint: Optional[str] = None - self.last_error: Optional[str] = None + self.authorization_endpoint: str | None = None + self.token_endpoint: str | None = None + self.userinfo_endpoint: str | None = None + self.end_session_endpoint: str | None = None + self.last_error: str | None = None self._loaded = False async def load(self, issuer_url: str) -> bool: @@ -150,12 +150,12 @@ class OIDCProviderMetadata: class OIDCUtils: """OIDC 工具类""" - _metadata: Optional[OIDCProviderMetadata] = None + _metadata: OIDCProviderMetadata | None = None _state_store: dict[str, dict[str, Any]] = {} _login_code_store: dict[str, dict[str, Any]] = {} _state_ttl_seconds = 300 _login_code_ttl_seconds = 60 - _last_metadata_error: Optional[str] = None + _last_metadata_error: str | None = None @classmethod def _cleanup_expired_state(cls) -> None: @@ -172,7 +172,7 @@ class OIDCUtils: cls._login_code_store.pop(key, None) @classmethod - async def get_metadata(cls) -> Optional[OIDCProviderMetadata]: + async def get_metadata(cls) -> OIDCProviderMetadata | None: """获取 OIDC Provider 元数据""" if not oidc_config.enabled or not oidc_config.is_configured(): cls._last_metadata_error = "OIDC 未启用或基础配置不完整" @@ -203,7 +203,7 @@ class OIDCUtils: return cls._metadata @classmethod - def get_last_metadata_error(cls) -> Optional[str]: + def get_last_metadata_error(cls) -> str | None: """获取最近一次 OIDC 元数据加载错误""" return cls._last_metadata_error @@ -219,7 +219,7 @@ class OIDCUtils: return state @classmethod - def verify_state(cls, state: str) -> Optional[dict[str, Any]]: + def verify_state(cls, state: str) -> dict[str, Any] | None: """验证 state 参数""" state_data = cls._state_store.pop(state, None) if not state_data: @@ -240,7 +240,7 @@ class OIDCUtils: return code @classmethod - def consume_login_code(cls, code: str) -> Optional[dict[str, Any]]: + def consume_login_code(cls, code: str) -> dict[str, Any] | None: """消费一次性短期登录 code""" data = cls._login_code_store.pop(code, None) if not data: @@ -255,7 +255,7 @@ class OIDCUtils: return secrets.token_urlsafe(32) @classmethod - async def build_authorization_url(cls, redirect_path: str = "/") -> Optional[str]: + async def build_authorization_url(cls, redirect_path: str = "/") -> str | None: """构建授权 URL""" metadata = await cls.get_metadata() if not metadata or not metadata.authorization_endpoint: @@ -281,7 +281,7 @@ class OIDCUtils: return f"{metadata.authorization_endpoint}?{query_string}" @classmethod - async def exchange_code_for_token(cls, code: str) -> Optional[dict[str, Any]]: + async def exchange_code_for_token(cls, code: str) -> dict[str, Any] | None: """用授权码交换令牌""" metadata = await cls.get_metadata() if not metadata or not metadata.token_endpoint: @@ -313,7 +313,7 @@ class OIDCUtils: return None @classmethod - async def get_userinfo(cls, access_token: str) -> Optional[dict[str, Any]]: + async def get_userinfo(cls, access_token: str) -> dict[str, Any] | None: """获取用户信息""" metadata = await cls.get_metadata() if not metadata or not metadata.userinfo_endpoint: @@ -334,7 +334,7 @@ class OIDCUtils: return None @classmethod - async def build_logout_url(cls, id_token: Optional[str] = None) -> Optional[str]: + async def build_logout_url(cls, id_token: str | None = None) -> str | None: """构建登出 URL""" metadata = await cls.get_metadata() if not metadata or not metadata.end_session_endpoint: @@ -418,9 +418,7 @@ async def find_user_by_oidc_sub(db, sub: str) -> User | None: return user legacy_result = await db.execute( - select(User) - .filter(User.user_id.like(f"{oidc_user_id}:%"), User.is_deleted == 0) - .order_by(User.id.asc()) + select(User).filter(User.user_id.like(f"{oidc_user_id}:%"), User.is_deleted == 0).order_by(User.id.asc()) ) legacy_users = list(legacy_result.scalars().all()) if legacy_users: @@ -441,9 +439,7 @@ async def find_deleted_oidc_user_by_sub(db, sub: str) -> User | None: return deleted_user legacy_result = await db.execute( - select(User) - .filter(User.user_id.like(f"{oidc_user_id}:%"), User.is_deleted == 1) - .order_by(User.id.asc()) + select(User).filter(User.user_id.like(f"{oidc_user_id}:%"), User.is_deleted == 1).order_by(User.id.asc()) ) return legacy_result.scalar_one_or_none()