Corrigir documentação da OpenAPI para mostrar os códigos de status HTTP que de fato podem ser retornados

src/api.py

@@ -14,8 +14,9 @@
 
 import crud
 import crud_auth
-import email_config
 from db_config import DbContextManager, create_db_and_tables
+import email_config
+import response_schemas
 import schemas
 from util import check_permissions
 
@@ -91,8 +92,12 @@ async def docs_redirect(
 @app.post(
     "/token",
     summary="Autentica na API.",
-    response_model=schemas.Token,
     tags=["Auth"],
+    response_model=schemas.Token,
+    responses={
+        **response_schemas.email_validation_error,
+        401: response_schemas.UnauthorizedErrorResponse.docs(),
+    },
 )
 async def login_for_access_token(
     form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
@@ -129,6 +134,8 @@ async def login_for_access_token(
     "/users",
     summary="Lista usuários da API.",
     tags=["Auth"],
+    response_model=list[schemas.UsersGetSchema],
+    responses=response_schemas.not_admin_error,
 )
 async def get_users(
     user_logged: Annotated[  # pylint: disable=unused-argument
@@ -145,6 +152,15 @@ async def get_users(
     "/user/{email}",
     summary="Cria ou altera usuário na API.",
     tags=["Auth"],
+    response_model=schemas.UsersGetSchema,
+    responses={
+        **response_schemas.not_admin_error,
+        422: response_schemas.ValidationErrorResponse.docs(
+            examples=response_schemas.value_response_example(
+                "email deve ser igual na url e no json"
+            )
+        ),
+    },
 )
 async def create_or_update_user(
     user_logged: Annotated[  # pylint: disable=unused-argument
@@ -153,7 +169,7 @@ async def create_or_update_user(
     user: schemas.UsersSchema,
     email: str,
     db: DbContextManager = Depends(DbContextManager),
-) -> dict:
+) -> JSONResponse:
     """Cria um usuário da API ou atualiza os seus dados cadastrais."""
 
     # Validações
@@ -177,26 +193,39 @@ async def create_or_update_user(
         ) from exception
 
     # Call
+    response_status = status.HTTP_200_OK
     try:
         # update
         if await crud_auth.get_user(db, user.email):
             await crud_auth.update_user(db, user)
         # create
         else:
             await crud_auth.create_user(db, user)
+            response_status = status.HTTP_201_CREATED
     except IntegrityError as exception:
         raise HTTPException(
             status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
             detail=f"IntegrityError: {str(exception)}",
         ) from exception
 
-    return user.dict(exclude=["password"])
+    return JSONResponse(
+        content=user.dict(exclude=["password"]), status_code=response_status
+    )
 
 
 @app.get(
     "/user/{email}",
     summary="Consulta um usuário da API.",
     tags=["Auth"],
+    response_model=schemas.UsersGetSchema,
+    responses={
+        **response_schemas.not_admin_error,
+        404: response_schemas.NotFoundErrorResponse.docs(
+            examples=response_schemas.value_response_example(
+                "Usuário `[email protected]` não existe"
+            )
+        ),
+    },
 )
 async def get_user(
     user_logged: Annotated[  # pylint: disable=unused-argument
@@ -223,6 +252,23 @@ async def get_user(
     "/user/forgot_password/{email}",
     summary="Solicita recuperação de acesso à API.",
     tags=["Auth"],
+    responses={
+        **response_schemas.email_validation_error,
+        200: response_schemas.OKMessageResponse.docs(
+            examples={
+                "OK": {
+                    "value": response_schemas.OKMessageResponse(
+                        message="Email enviado!"
+                    ).json(),
+                },
+            }
+        ),
+        404: response_schemas.NotFoundErrorResponse.docs(
+            examples=response_schemas.value_response_example(
+                "Usuário `[email protected]` não existe"
+            )
+        ),
+    },
 )
 async def forgot_password(
     email: str,
@@ -250,6 +296,18 @@ async def forgot_password(
     "/user/reset_password/",
     summary="Criar nova senha a partir do token de acesso.",
     tags=["Auth"],
+    responses={
+        200: response_schemas.OKMessageResponse.docs(
+            examples={
+                "OK": {
+                    "value": response_schemas.OKMessageResponse(
+                        message="Senha do Usuário `[email protected]` atualizada"
+                    ).json(),
+                },
+            }
+        ),
+        400: response_schemas.BadRequestErrorResponse.docs(),
+    },
 )
 async def reset_password(
     access_token: str,
@@ -274,8 +332,16 @@ async def reset_password(
     "/organizacao/{origem_unidade}/{cod_unidade_autorizadora}"
     "/plano_entregas/{id_plano_entregas}",
     summary="Consulta plano de entregas",
-    response_model=schemas.PlanoEntregasSchema,
     tags=["plano de entregas"],
+    response_model=schemas.PlanoEntregasSchema,
+    responses={
+        **response_schemas.outra_unidade_error,
+        404: response_schemas.NotFoundErrorResponse.docs(
+            examples=response_schemas.value_response_example(
+                "Plano de entregas não encontrado"
+            )
+        ),
+    },
 )
 async def get_plano_entrega(
     user: Annotated[schemas.UsersSchema, Depends(crud_auth.get_current_active_user)],
@@ -307,8 +373,9 @@ async def get_plano_entrega(
     "/organizacao/{origem_unidade}/{cod_unidade_autorizadora}"
     "/plano_entregas/{id_plano_entregas}",
     summary="Cria ou substitui plano de entregas",
-    response_model=schemas.PlanoEntregasSchema,
     tags=["plano de entregas"],
+    response_model=schemas.PlanoEntregasSchema,
+    responses=response_schemas.outra_unidade_error,
 )
 async def create_or_update_plano_entregas(
     user: Annotated[schemas.UsersSchema, Depends(crud_auth.get_current_active_user)],
@@ -396,8 +463,16 @@ async def create_or_update_plano_entregas(
     "/organizacao/{origem_unidade}/{cod_unidade_autorizadora}"
     "/plano_trabalho/{id_plano_trabalho}",
     summary="Consulta plano de trabalho",
-    response_model=schemas.PlanoTrabalhoSchema,
     tags=["plano de trabalho"],
+    response_model=schemas.PlanoTrabalhoSchema,
+    responses={
+        **response_schemas.outra_unidade_error,
+        404: response_schemas.NotFoundErrorResponse.docs(
+            examples=response_schemas.value_response_example(
+                "Plano de trabalho não encontrado"
+            )
+        ),
+    },
 )
 async def get_plano_trabalho(
     user: Annotated[schemas.UsersSchema, Depends(crud_auth.get_current_active_user)],
@@ -429,8 +504,9 @@ async def get_plano_trabalho(
     "/organizacao/{origem_unidade}/{cod_unidade_autorizadora}"
     "/plano_trabalho/{id_plano_trabalho}",
     summary="Cria ou substitui plano de trabalho",
-    response_model=schemas.PlanoTrabalhoSchema,
     tags=["plano de trabalho"],
+    response_model=schemas.PlanoTrabalhoSchema,
+    responses=response_schemas.outra_unidade_error,
 )
 async def create_or_update_plano_trabalho(
     user: Annotated[schemas.UsersSchema, Depends(crud_auth.get_current_active_user)],
@@ -527,8 +603,16 @@ async def create_or_update_plano_trabalho(
     "/organizacao/{origem_unidade}/{cod_unidade_autorizadora}"
     "/{cod_unidade_lotacao}/participante/{matricula_siape}",
     summary="Consulta um Participante",
-    response_model=schemas.ParticipanteSchema,
     tags=["participante"],
+    response_model=schemas.ParticipanteSchema,
+    responses={
+        **response_schemas.outra_unidade_error,
+        404: response_schemas.NotFoundErrorResponse.docs(
+            examples=response_schemas.value_response_example(
+                "Participante não encontrado"
+            )
+        ),
+    },
 )
 async def get_participante(
     user: Annotated[schemas.UsersSchema, Depends(crud_auth.get_current_active_user)],
@@ -562,8 +646,9 @@ async def get_participante(
     "/organizacao/{origem_unidade}/{cod_unidade_autorizadora}"
     "/{cod_unidade_lotacao}/participante/{matricula_siape}",
     summary="Envia um participante",
-    response_model=schemas.ParticipanteSchema,
     tags=["participante"],
+    response_model=schemas.ParticipanteSchema,
+    responses=response_schemas.outra_unidade_error,
 )
 async def create_or_update_participante(
     user: Annotated[schemas.UsersSchema, Depends(crud_auth.get_current_active_user)],

src/crud_auth.py

@@ -182,7 +182,8 @@ async def get_current_admin_user(
 
     if not current_user.is_admin:
         raise HTTPException(
-            status_code=401, detail="Usuário não tem permissões de administrador"
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Usuário não tem permissões de administrador",
         )
 
     return current_user

src/response_schemas.py

@@ -0,0 +1,147 @@
+"""Esquemas Pydantic para as respostas da API com mensagens técnicas não
+relacionadas aos requisitos de negócio."""
+
+from abc import ABC
+from typing import Optional
+
+from fastapi import status
+from pydantic import BaseModel
+
+# Classes de respostas de dados da API para mensagens técnicas
+
+
+class ResponseData(ABC, BaseModel):
+    """Classe abstrata para conter os métodos comuns a todos os modelos
+    de dados de resposta da API."""
+
+    _title = ""
+
+    @classmethod
+    def get_title(cls):
+        """Retorna o título da resposta."""
+        return cls._title.default
+
+    @classmethod
+    def docs(cls, examples: Optional[dict] = None) -> dict:
+        """Retorna a documentação da resposta para o método, exibida pelo
+        FastAPI na interface OpenAPI."""
+        docs = {
+            "model": cls,
+            "description": cls.get_title(),
+            "content": {"application/json": {}},
+        }
+        if examples is not None:
+            docs["content"]["application/json"]["examples"] = examples
+        return docs
+
+
+class OKMessageResponse(ResponseData):
+    """Resposta da API para mensagens bem sucedidas."""
+
+    _status_code = status.HTTP_200_OK
+    _title = "OK"
+    message: str
+
+
+class BadRequestErrorResponse(ResponseData):
+    """Resposta da API para erros de autorização."""
+
+    _status_code = status.HTTP_400_BAD_REQUEST
+    _title = "Bad request"
+    detail: str
+
+
+class UnauthorizedErrorResponse(ResponseData):
+    """Resposta da API para erros de autorização."""
+
+    _status_code = status.HTTP_401_UNAUTHORIZED
+    _title = "Unauthorized access"
+    detail: str
+
+
+class ForbiddenErrorResponse(ResponseData):
+    """Resposta da API para erros de permissão."""
+
+    _status_code = status.HTTP_403_FORBIDDEN
+    _title = "Forbidden"
+    detail: str
+
+
+class NotFoundErrorResponse(ResponseData):
+    """Resposta da API para erros de permissão."""
+
+    _status_code = status.HTTP_404_NOT_FOUND
+    _title = "Not found"
+    detail: str
+
+
+class ValidationError(BaseModel):
+    """Estrutura retornada pelo Pydantic para cada erro de validação."""
+
+    type: str
+    loc: list[str]
+    msg: str
+    ctx: dict
+
+
+class ValidationErrorResponse(ResponseData):
+    """Resposta da API para erros de validação."""
+
+    _status_code = status.HTTP_422_UNPROCESSABLE_ENTITY
+    _title = "Unprocessable Entity"
+    detail: list[ValidationError]
+
+
+# Funções auxiliares para documentação de possíveis respostas técnicas
+# na API.
+
+# Documentação de respostas comuns a diversos endpoints
+not_logged_error = {
+    401: UnauthorizedErrorResponse.docs(
+        examples={"Invalid credentials": {"value": {"detail": "Not authenticated"}}}
+    ),
+}
+not_admin_error = {
+    **not_logged_error,
+    403: ForbiddenErrorResponse.docs(
+        examples={
+            "Forbidden": {
+                "value": {"detail": "Usuário não tem permissões de administrador"}
+            }
+        }
+    ),
+}
+outra_unidade_error = {
+    **not_logged_error,
+    403: ForbiddenErrorResponse.docs(
+        examples={
+            "Forbidden": {
+                "value": {
+                    "detail": "Usuário não tem permissão na cod_unidade_autorizadora informada"
+                }
+            }
+        }
+    ),
+}
+email_validation_error = {
+    422: ValidationErrorResponse.docs(
+        examples={
+            "Invalid email format": {
+                "value": ValidationErrorResponse(
+                    detail=[
+                        ValidationError(
+                            type="value_error",
+                            loc=["email"],
+                            msg="value is not a valid email address: "
+                            "An email address must have an @-sign.",
+                            input="my_username",
+                            ctx={"reason": "An email address must have an @-sign."},
+                            url="https://errors.pydantic.dev/2.8/v/value_error",
+                        )
+                    ]
+                ).json()
+            }
+        }
+    )
+}
+value_response_example = lambda message: {"example": {"value": {"detail": message}}}

src/schemas.py

@@ -206,10 +206,7 @@ def validate_data_avaliacao_data_inicio_periodo_avaliativo(
     ) -> "AvaliacaoRegistrosExecucaoSchema":
         """Valida se a data de avaliação dos registros de execução é
         posterior à data de início do período avaliativo."""
-        if (
-            self.data_avaliacao_registros_execucao
-            < self.data_inicio_periodo_avaliativo
-        ):
+        if self.data_avaliacao_registros_execucao < self.data_inicio_periodo_avaliativo:
             raise ValueError(
                 "A data de avaliação de registros de execução deve ser "
                 "igual ou posterior à data de início do período avaliativo."