add bots

common_lib/utils/auth.py

@@ -1,5 +1,5 @@
 import httpx
-from fastapi import Depends, Request, HTTPException, status
+from fastapi import Depends, Request, HTTPException, status, Header
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from typing import List
 
@@ -15,14 +15,17 @@ auth_scheme = HTTPBearer()
 @dataclass
 class CurrentUser:
     token: str
+    is_bot: str
     user_id: str
     session_id: str
     permissions: List[str]
 
+
 async def _fetch_current_user(
     request: Request,
     credentials: HTTPAuthorizationCredentials,
-    require_permissions: bool
+    require_permissions: bool,
+    is_bot=False
 ) -> CurrentUser:
     token = credentials.credentials
     ip = request.client.host or "(unknown)"
@@ -36,7 +39,8 @@ async def _fetch_current_user(
                 "token": token,
                 "ip": ip,
                 "user_agent": user_agent,
-                "require_permissions": require_permissions
+                "require_permissions": require_permissions,
+                "is_bot": is_bot
             },
         )
     except httpx.RequestError as e:
@@ -54,6 +58,7 @@ async def _fetch_current_user(
 
     return CurrentUser(
         token=token,
+        is_bot=data["is_bot"],
         user_id=data["user_id"],
         session_id=data["session_id"],
         permissions=data["permissions"]
@@ -120,6 +125,7 @@ async def fetch_user_for_sio(
 
     return CurrentUser(
         token=token,
+        is_bot=data["is_bot"],
         user_id=data["user_id"],
         session_id=data["session_id"],
         permissions=data["permissions"]
@@ -139,6 +145,16 @@ async def get_current_user_with_permissions(
 ) -> CurrentUser:
     return await _fetch_current_user(request, credentials, require_permissions=True)
 
+
+async def get_current_user_or_bot(
+    request: Request,
+    is_bot: str | None = Header(default=None),
+    credentials: HTTPAuthorizationCredentials = Depends(auth_scheme)
+) -> CurrentUser:
+    is_bot_flag = str(is_bot).lower() in ("true", "1", "yes") if is_bot else False
+    return await _fetch_current_user(request, credentials, require_permissions=False, is_bot=is_bot_flag)
+
+
 def validate_username(value: str,
                       field_name: str = "login",
                       with_httpexception=False,

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "common-lib"
-version = "0.0.30"
+version = "0.0.31"
 description = "Библиотека общих компонентов для микросервисов yobble"
 authors = [{ name = "cheykrym", email = "you@example.com" }]
 license = "MIT"