Разделение логики удаления юзера по скоупам (#237)

Author: DaymasS

auth_backend/routes/user.py

@@ -2,11 +2,10 @@
 from datetime import datetime
 from typing import Any, Literal
 
-from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi import APIRouter, Depends, Query
 from fastapi_sqlalchemy import db
 from sqlalchemy import not_
 from sqlalchemy.orm import Session
-from starlette.status import HTTP_403_FORBIDDEN
 
 from auth_backend.auth_method import AuthPluginMeta
 from auth_backend.auth_plugins.email import Email
@@ -144,36 +143,44 @@ async def patch_user(
 @user.delete("/{user_id}", response_model=None)
 async def delete_user(
     user_id: int,
-    current_user: UserSession = Depends(UnionAuth(scopes=[], allow_none=False, auto_error=True)),
+    current_user: UserSession = Depends(UnionAuth(scopes=["auth.user.delete"], allow_none=False, auto_error=True)),
 ) -> None:
     """
-    Scopes: `["auth.user.delete"]` or `["auth.user.selfdelete"]` for self delete
+    Scopes: `["auth.user.delete"]`
     """
-    session_scopes = set([scope.name.lower() for scope in current_user.scopes])
-    if "auth.user.delete" in session_scopes or (
-        "auth.user.selfdelete" in session_scopes and user_id == current_user.user_id
-    ):
-        logger.debug(f'User id={current_user.id} triggered delete_user')
-        old_user = {"user_id": current_user.id}
-        user: User = User.get(user_id, session=db.session)
-
-        for method in user._auth_methods:
-            if method.is_deleted:
-                continue
-            # Сохраняем старое состояние пользователя
-            if method.auth_method not in old_user:
-                old_user[method.auth_method] = {}
-            old_user[method.auth_method][method.param] = method.value
-            # Удаляем AuthMethod
-            AuthMethod.delete(method.id, session=db.session)
-            logger.info(f'{method=} for {user.id=} deleted')
-        User.delete(user_id, session=db.session)
-        # Удаляем сессии
-        db.session.query(UserSession).filter(UserSession.user_id == user_id).filter(not_(UserSession.expired)).update(
-            {"expires": datetime.utcnow()}
-        )
-        db.session.commit()
-        await AuthPluginMeta.user_updated(None, old_user)
-        logger.info(f'{user=} deleted')
-    else:
-        raise HTTPException(status_code=HTTP_403_FORBIDDEN, detail="Not authorized")
+    await delete_user_common(user_triggered_id=current_user.user_id, user_delete_id=user_id)
+
+
+@user.delete("", response_model=None)
+async def delete_self_user(
+    current_user: UserSession = Depends(UnionAuth(scopes=["auth.user.selfdelete"], allow_none=False, auto_error=True))
+) -> None:
+    """
+    Scopes: `["auth.user.selfdelete"]`
+    """
+    await delete_user_common(user_triggered_id=current_user.user_id, user_delete_id=current_user.user_id)
+
+
+async def delete_user_common(user_triggered_id: int, user_delete_id: int):
+    logger.debug(f'User id={user_triggered_id} triggered delete_user')
+    old_user = {"user_id": user_triggered_id}
+    user: User = User.get(user_delete_id, session=db.session)
+
+    for method in user._auth_methods:
+        if method.is_deleted:
+            continue
+        # Сохраняем старое состояние пользователя
+        if method.auth_method not in old_user:
+            old_user[method.auth_method] = {}
+        old_user[method.auth_method][method.param] = method.value
+        # Удаляем AuthMethod
+        AuthMethod.delete(method.id, session=db.session)
+        logger.info(f'{method=} for {user.id=} deleted')
+    User.delete(user_delete_id, session=db.session)
+    # Удаляем сессии
+    db.session.query(UserSession).filter(UserSession.user_id == user_delete_id).filter(
+        not_(UserSession.expired)
+    ).update({"expires": datetime.utcnow()})
+    db.session.commit()
+    await AuthPluginMeta.user_updated(None, old_user)
+    logger.info(f'{user=} deleted')

auth_backend/settings.py

@@ -7,7 +7,6 @@
 
 from annotated_types import Gt
 from pydantic import PostgresDsn
-from pydantic.types import PathType
 from pydantic_settings import BaseSettings, SettingsConfigDict