Информация о пользователе для админки тех. поддержки

Makefile

@@ -22,4 +22,4 @@ db:
 	docker run -d -p 5432:5432 -e POSTGRES_HOST_AUTH_METHOD=trust --name db-userdata_api postgres:15
 
 migrate:
-	alembic upgrade head
+	source ./venv/bin/activate && alembic upgrade head

tests/test_routes/test_users_get.py

@@ -1,5 +1,3 @@
-from time import sleep
-
 import pytest
 
 from userdata_api.models.db import Info, Param

userdata_api/routes/admin.py

@@ -0,0 +1,43 @@
+from typing import Any
+
+from auth_lib.fastapi import UnionAuth
+from fastapi import APIRouter, Depends
+
+from userdata_api.schemas.admin import UserCardGet, UserCardUpdate
+from userdata_api.schemas.response_model import StatusResponseModel
+from userdata_api.utils.admin import get_user_info, patch_user_info
+
+
+admin = APIRouter(prefix="/admin", tags=["Admin"])
+
+
+@admin.get("/user/{user_id}", response_model=UserCardGet)
+async def get_user_card(
+    user_id: int,
+    user: dict[str, Any] = Depends(UnionAuth(scopes=["userdata.info.admin"], allow_none=False, auto_error=True)),
+):
+    """
+    Получает профсоюзную информацию пользователя.
+
+    Скоупы: `["userdata.info.admin"]`
+    """
+
+    return await get_user_info(user_id, user)
+
+
+@admin.patch("/user/{user_id}", response_model=StatusResponseModel)
+async def update_user_card(
+    new_info: UserCardUpdate,
+    user_id: int,
+    user: dict[str, Any] = Depends(UnionAuth(scopes=["userdata.info.admin"], allow_none=False, auto_error=True)),
+) -> StatusResponseModel:
+    """
+    Обновить данные в профсоюзной информации пользователя.
+
+    Скоупы: `["userdata.info.admin"]`
+
+     - **user_id**: id пользователя.
+    """
+
+    await patch_user_info(new_info, user_id, user)
+    return StatusResponseModel(status="Success", message="User patch succeeded", ru="Изменение успешно")

userdata_api/routes/base.py

@@ -5,6 +5,7 @@
 from settings import get_settings
 from userdata_api import __version__
 
+from .admin import admin
 from .category import category
 from .param import param
 from .source import source
@@ -41,3 +42,4 @@
 app.include_router(category)
 app.include_router(param)
 app.include_router(user)
+app.include_router(admin)

userdata_api/routes/user.py

@@ -2,9 +2,7 @@
 
 from auth_lib.fastapi import UnionAuth
 from fastapi import APIRouter, Depends, Query
-from fastapi_sqlalchemy import db
 
-from userdata_api.models.db import Category, Info
 from userdata_api.schemas.response_model import StatusResponseModel
 from userdata_api.schemas.user import UserInfoGet, UserInfoUpdate, UsersInfoGet
 from userdata_api.utils.user import get_user_info as get

userdata_api/schemas/admin.py

@@ -0,0 +1,14 @@
+from .base import Base
+
+
+class UserCardGet(Base):
+    user_id: int
+    full_name: str | None = None
+    student_card_number: str | None = None
+    union_card_number: str | None = None
+    is_union_member: str
+
+
+class UserCardUpdate(Base):
+    full_name: str | None = None
+    student_card_number: str | None = None

userdata_api/utils/admin.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+from fastapi_sqlalchemy import db
+
+from userdata_api.exceptions import ObjectNotFound
+from userdata_api.models.db import Info, Param
+from userdata_api.schemas.admin import UserCardGet, UserCardUpdate
+from userdata_api.schemas.user import UserInfo, UserInfoUpdate
+
+from .user import patch_user_info as user_patch
+
+
+async def patch_user_info(new: UserCardUpdate, user_id: int, user: dict[str, int | list[dict[str, str | int]]]) -> None:
+    """
+    Обновить информацию о пользователе в соотетствии с переданным токеном.
+
+    Метод обновляет только информацию из источников `admin`.
+
+    Для обновления от имени админа нужен скоуп `userdata.info.admin`
+
+    :param new: модель запроса, в ней то, на что будет изменена информация о пользователе
+    :param user_id: Айди пользователя
+    :param user: Сессия пользователя выполняющего запрос
+    :return: get_user_info для текущего пользователя с переданными правами
+    """
+    update_info = []
+    if new.full_name is not None:
+        update_info.append(UserInfo(category="Личная информация", param="Полное имя", value=new.full_name))
+    if new.student_card_number is not None:
+        update_info.append(
+            UserInfo(category="Учёба", param="Номер студенческого билета", value=new.student_card_number)
+        )
+    if update_info:
+        update_request = UserInfoUpdate(items=update_info, source="admin")
+        await user_patch(update_request, user_id, user)
+
+
+async def get_user_info(user_id: int, user: dict[str, int | list[dict[str, str | int]]]) -> UserCardGet:
+    """
+    Получить профсоюзную информацию пользователя для админки.
+
+    :param user_id: Айди пользователя, информацию о котором запрашиваем
+    :param user: Сессия пользователя, выполняющего запрос (должен иметь права администратора)
+    :return: Словарь с данными пользователя:
+        - user_id: ID пользователя
+        - full_name: Полное имя (из параметра "Полное имя")
+        - student_card_number: Номер студенческого билета (из параметра "Номер студенческого билета")
+        - union_card_number: Номер профсоюзного билета (из параметра "Номер профсоюзного билета")
+        - is_union_member: Статус мэтчинга (из параметра "Членство в профсоюзе")
+        - last_check_timestamp: Дата последней проверки
+    """
+    users = db.session.query(Info).filter(Info.owner_id == user_id).first()
+    if not users:
+        raise ObjectNotFound(Info, user_id)
+    full_name = (
+        db.session.query(Info)
+        .join(Info.param)
+        .filter(Info.owner_id == user_id, Param.name == "Полное имя")
+        .one_or_none()
+    )
+    is_union_member = (
+        db.session.query(Info)
+        .join(Info.param)
+        .filter(Info.owner_id == user_id, Param.name == "Членство в профсоюзе")
+        .one_or_none()
+    )
+    student_card_number = (
+        db.session.query(Info)
+        .join(Info.param)
+        .filter(Info.owner_id == user_id, Param.name == "Номер студенческого билета")
+        .one_or_none()
+    )
+    union_card_number = (
+        db.session.query(Info)
+        .join(Info.param)
+        .filter(Info.owner_id == user_id, Param.name == "Номер профсоюзного билета")
+        .one_or_none()
+    )
+    result = {
+        "user_id": user_id,
+        "full_name": full_name.value if full_name else None,
+        "student_card_number": student_card_number.value if student_card_number else None,
+        "union_card_number": union_card_number.value if union_card_number else None,
+        "is_union_member": is_union_member.value if is_union_member else "false",
+    }
+    return result

userdata_api/utils/user.py

@@ -3,7 +3,7 @@
 from re import search
 
 from fastapi_sqlalchemy import db
-from sqlalchemy import String, cast, func, not_, or_
+from sqlalchemy import not_, or_
 
 from userdata_api.exceptions import Forbidden, InvalidValidation, ObjectNotFound
 from userdata_api.models.db import Category, Info, Param, Source, ViewType