[add] users routing

app/api/endpoints/user.py

@@ -3,25 +3,43 @@
 from fastapi import APIRouter, Body, Depends, HTTPException, status
 from fastapi_users import InvalidPasswordException
 from sqlalchemy.ext.asyncio import AsyncSession
+from starlette.responses import Response
 
 from app.api.endpoints import register
 from app.api.validators import check_obj_exists
 from app.api_docs_responses.user import (
-    USER_CONFIRM_DESCRIPTION, add_router_doc,
+    USER_CONFIRM_DESCRIPTION,
+    USERS_DESCRIPTION,
+    add_router_doc,
 )
 from app.api_docs_responses.utils_docs import USER_VALUE
 from app.core.db import get_async_session
 from app.core.user import (
-    UserManager, auth_backend_cookie, auth_backend_jwt, current_user,
-    fastapi_users, get_user_manager,
+    UserManager,
+    auth_backend_cookie,
+    auth_backend_jwt,
+    current_user,
+    fastapi_users,
+    get_user_manager,
 )
 from app.crud.hasher import Hasher
 from app.crud.user import user_crud
 from app.models import User
 from app.schemas.user import (
-    UserChangePassword, UserCreate, UserRead, UserReadRegister, UserUpdate,
+    UserChangePassword,
+    UserCreate,
+    UserList,
+    UserRead,
+    UserReadRegister,
+    UserUpdate,
 )
 from app.services.token_generator.tokens import token_generator
+from app.services.utils import (
+    Pagination,
+    add_response_headers,
+    get_pagination_params,
+    paginated,
+)
 
 router = APIRouter()
 
@@ -39,23 +57,41 @@
 
 router.include_router(
     register.get_register_router(
-        UserReadRegister,
-        Annotated[UserCreate, Body(example=USER_VALUE)]
+        UserReadRegister, Annotated[UserCreate, Body(example=USER_VALUE)]
     ),
     prefix='/auth',
     tags=['auth'],
 )
 
+
 router.include_router(
     fastapi_users.get_users_router(
-        UserRead,
-        Annotated[UserUpdate, Body(example=USER_VALUE)]
+        UserRead, Annotated[UserUpdate, Body(example=USER_VALUE)]
     ),
     prefix='/users',
     tags=['users'],
 )
 
 
+@router.get(
+    "/users",
+    response_model=list[UserList],
+    tags=["users"],
+    summary='Получение всех пользователей.',
+    status_code=status.HTTP_200_OK,
+    description=USERS_DESCRIPTION,
+    dependencies=[Depends(current_user)],
+)
+async def get_users(
+    response: Response,
+    session: AsyncSession = Depends(get_async_session),
+    pagination: Pagination = Depends(get_pagination_params),
+):
+    users = await user_crud.get_users(session)
+    add_response_headers(response, users, pagination)
+    return paginated(users, pagination)
+
+
 @router.post(
     '/users/{user_id}/{confirm_code}',
     tags=['users'],
@@ -140,4 +176,5 @@ async def change_password(
 
     return {'detail': 'Пароль был успешно изменен.'}
 
+
 add_router_doc(router)

app/api_docs_responses/user.py

@@ -199,6 +199,17 @@ class RouteEnum(enum.IntEnum):
         Только суперпользователь.\n
 """
 
+USERS_DESCRIPTION = """
+    Получение информации о пользователях.
+
+    Args:\n
+        None.\n
+    Returns:\n
+        list: dict: Пользователь.\n
+    Permissions:\n
+        Только зарегистрированный пользователь.\n
+"""
+
 USER_CONFIRM_DESCRIPTION = """
     По полученной пользователем ссылке на почту происходит подтверждение
     а верификация пользователя с его ящиком.

app/crud/user.py

@@ -1,4 +1,4 @@
-from sqlalchemy import select
+from sqlalchemy import desc, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.crud.base import CRUDBase
@@ -41,5 +41,14 @@ async def get_user_by_credentials(
             return None
         return res
 
+    async def get_users(
+        self,
+        session: AsyncSession,
+    ):
+        db_objs = await session.execute(
+            select(self.model).order_by(desc('id'))
+        )
+        return db_objs.scalars().all()
+
 
 user_crud = CRUDUser(User)

app/schemas/user.py

@@ -1,13 +1,21 @@
 from typing import Optional
 
 from fastapi_users import schemas
-from pydantic import WithJsonSchema, field_serializer
+from pydantic import BaseModel, WithJsonSchema, field_serializer
 from sqlalchemy_utils import Choice
 from typing_extensions import Annotated
 
 from app.core.constants import Role
 
 
+class RoleMixin:
+    role: Annotated[Choice, WithJsonSchema({'type': 'str'})]
+
+    @field_serializer('role')
+    def serialize_role(self, role: Choice, _info):
+        return role.code
+
+
 class UserReadRegister(schemas.BaseUser[int]):
     username: str
     tariff_id: Optional[int]
@@ -17,15 +25,17 @@ class Config:
         arbitrary_types_allowed = True
 
 
-class UserRead(UserReadRegister):
-    role: Annotated[
-        Choice,
-        WithJsonSchema({'type': 'str'})
-    ]
+class UserRead(RoleMixin, UserReadRegister):
+    pass
 
-    @field_serializer('role')
-    def serialize_role(self, role: Choice, _info):
-        return role.code
+
+class UserList(RoleMixin, BaseModel):
+    id: int
+    email: str
+    username: str
+
+    class Config:
+        arbitrary_types_allowed = True
 
 
 class UserCreate(schemas.BaseUserCreate):