perf: Optimized code.

Author: amisadmin

fastapi_user_auth/auth/auth.py

@@ -39,10 +39,10 @@
 from .models import BaseUser, Role, User, UserRoleLink
 from .schemas import UserLoginOut
 
-_UserModelT = TypeVar("_UserModelT", bound=BaseUser)
+UserModelT = TypeVar("UserModelT", bound=BaseUser)
 
 
-class AuthBackend(AuthenticationBackend, Generic[_UserModelT]):
+class AuthBackend(AuthenticationBackend, Generic[UserModelT]):
     def __init__(self, auth: "Auth", token_store: BaseTokenStore):
         self.auth = auth
         self.token_store = token_store
@@ -53,23 +53,23 @@ def get_user_token(request: Request) -> Optional[str]:
         scheme, token = get_authorization_scheme_param(authorization)
         return None if not authorization or scheme.lower() != "bearer" else token
 
-    async def authenticate(self, request: Request) -> Tuple["Auth", Optional[_UserModelT]]:
+    async def authenticate(self, request: Request) -> Tuple["Auth", Optional[UserModelT]]:
         return self.auth, await self.auth.get_current_user(request)
 
     def attach_middleware(self, app: FastAPI):
         app.add_middleware(AuthenticationMiddleware, backend=self)  # 添加auth中间件
 
 
-class Auth(Generic[_UserModelT]):
-    user_model: Type[_UserModelT] = None
+class Auth(Generic[UserModelT]):
+    user_model: Type[UserModelT] = None
     db: Union[AsyncDatabase, Database] = None
-    backend: AuthBackend[_UserModelT] = None
+    backend: AuthBackend[UserModelT] = None
 
     def __init__(
         self,
         db: Union[AsyncDatabase, Database],
         token_store: BaseTokenStore = None,
-        user_model: Type[_UserModelT] = User,
+        user_model: Type[UserModelT] = User,
         pwd_context: CryptContext = CryptContext(schemes=["bcrypt"], deprecated="auto"),
     ):
         self.user_model = user_model or self.user_model
@@ -78,7 +78,7 @@ def __init__(
         self.backend = self.backend or AuthBackend(self, token_store or DbTokenStore(self.db))
         self.pwd_context = pwd_context
 
-    async def authenticate_user(self, username: str, password: Union[str, SecretStr]) -> Optional[_UserModelT]:
+    async def authenticate_user(self, username: str, password: Union[str, SecretStr]) -> Optional[UserModelT]:
         user = await self.db.async_scalar(select(self.user_model).where(self.user_model.username == username))
         if user:
             pwd = password.get_secret_value() if isinstance(password, SecretStr) else password
@@ -87,21 +87,17 @@ async def authenticate_user(self, username: str, password: Union[str, SecretStr]
                 return user
         return None
 
-    @cached_property
-    def get_current_user(self):
-        async def _get_current_user(request: Request) -> Optional[_UserModelT]:
-            if request.scope.get("auth"):  # 防止重复授权
-                return request.scope.get("user")
-            request.scope["auth"], request.scope["user"] = self, None
-            token = self.backend.get_user_token(request)
-            if not token:
-                return None
-            token_data = await self.backend.token_store.read_token(token)
-            if token_data is not None:
-                request.scope["user"]: _UserModelT = await self.db.async_get(self.user_model, token_data.id)
-            return request.user
-
-        return _get_current_user
+    async def get_current_user(self, request: Request) -> Optional[UserModelT]:
+        if request.scope.get("auth"):  # 防止重复授权
+            return request.scope.get("user")
+        request.scope["auth"], request.scope["user"] = self, None
+        token = self.backend.get_user_token(request)
+        if not token:
+            return None
+        token_data = await self.backend.token_store.read_token(token)
+        if token_data is not None:
+            request.scope["user"]: UserModelT = await self.db.async_get(self.user_model, token_data.id)
+        return request.user
 
     def requires(
         self,
@@ -116,12 +112,12 @@ def requires(
         roles_ = (roles,) if not roles or isinstance(roles, str) else tuple(roles)
         permissions_ = (permissions,) if not permissions or isinstance(permissions, str) else tuple(permissions)
 
-        async def has_requires(user: _UserModelT) -> bool:
+        async def has_requires(user: UserModelT) -> bool:
             return user and await self.db.async_run_sync(user.has_requires, roles=roles, groups=groups, permissions=permissions)
 
         async def depend(
             request: Request,
-            user: _UserModelT = Depends(self.get_current_user),
+            user: UserModelT = Depends(self.get_current_user),
         ) -> Union[bool, Response]:
             user_auth = request.scope.get("__user_auth__", None)
             if user_auth is None: