Информация о пользователе от лк мгу

Author: petrCher

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,42 @@
+from typing import Any
+
+from auth_lib.fastapi import UnionAuth
+from fastapi import APIRouter, Depends
+
+from userdata_api.schemas.admin import UserDebugCardGet, UserDebugCardUpdate
+from userdata_api.schemas.response_model import StatusResponseModel
+from userdata_api.utils.admin import get_user_info as get
+from userdata_api.utils.admin import patch_user_info as patch
+
+
+admin = APIRouter(prefix="/admin", tags=["Admin"])
+
+
+@admin.get("/user/{user_id}", response_model=UserDebugCardGet)
+async def get_user_debug_card(
+    user_id: int,  ##or full_name
+    user: dict[str, Any] = Depends(UnionAuth(scopes=[], allow_none=False, auto_error=True)),  ##scopes
+) -> UserDebugCardGet:
+    """
+    Получает профсоюзную информацию пользователя.
+    """
+
+    return UserDebugCardGet.model_validate(await get(user_id, user))
+
+
+@admin.patch("/user/{user_id}", response_model=StatusResponseModel)
+async def update_user_debug_card(
+    new_info: UserDebugCardUpdate,
+    user_id: int,
+    user: dict[str, Any] = Depends(UnionAuth(scopes=[], allow_none=False, auto_error=True)),  ##scopes
+) -> StatusResponseModel:
+    """
+    Обновить данные в профсоюзной информации пользователя.
+
+     - **user_id**: id пользователя.
+
+    Возвращает **ObjectNotFound** пользователь с указанным user_id не найден.
+    """
+
+    await patch(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,17 @@
+from datetime import datetime
+
+from .base import Base
+
+
+class UserDebugCardGet(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
+    last_check_timestamp: datetime | None = None
+
+
+class UserDebugCardUpdate(Base):
+    full_name: str | None = None
+    student_card_number: str | None = None

userdata_api/schemas/user.py

@@ -34,3 +34,11 @@ class UsersInfoGet(Base):
 
 class UserInfoUpdate(UserInfoGet):
     source: constr(min_length=1)
+
+
+class UserDebugCard(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

userdata_api/utils/admin.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from userdata_api.schemas.admin import UserDebugCardGet, UserDebugCardUpdate
+
+
+async def patch_user_info(
+    new: UserDebugCardUpdate, user_id: int, user: dict[str, int | list[dict[str, str | int]]]
+) -> None:
+    """
+    Обновить информацию о пользователе в соотетствии с переданным токеном.
+
+    Метод обновляет только информацию из источников `admin`, `user` или `dwh`.
+
+    Для обновления от имени админа нужен скоуп `userdata.info.admin`
+
+    Для обновления информации из dwh нужен скоуп `userdata.info.dwh`
+
+    :param new: модель запроса, в ней то, на что будет изменена информация о пользователе
+    :param user_id: Айди пользователя
+    :param user: Сессия пользователя выполняющего запрос
+    :return: get_user_info для текущего пользователя с переданными правами
+    """
+
+
+async def get_user_info(user_id: int, user: dict[str, int | list[dict[str, str | int]]]) -> UserDebugCardGet:
+    """
+    Получить профсоюзную информацию пользователя для админки.
+
+    :param user_id: Айди пользователя, информацию о котором запрашиваем
+    :param user: Сессия пользователя, выполняющего запрос (должен иметь права администратора)
+    :return: Словарь с данными пользователя:
+        - user_id: ID пользователя
+        - full_name: Полное имя (из параметра "Полное имя")
+        - student_card_number: Номер студенческого билета (из параметра "Номер студенческого билета")
+        - union_card_number: Номер профсоюзного билета (из параметра "Номер профсоюзного билета")
+        - is_union_member: Статус мэтчинга (из параметра "Членство в профсоюзе")
+        - last_check_timestamp: Дата последней проверки
+    """

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