282 lines
13 KiB
Python
282 lines
13 KiB
Python
# Importações de bibliotecas padrão
|
|
from typing import Annotated, Any, Sequence, List, Optional, Dict, Union
|
|
from enum import Enum
|
|
|
|
# Importações de bibliotecas de terceiros
|
|
from fastapi import APIRouter, Depends
|
|
from starlette import status
|
|
from starlette.responses import Response
|
|
from sqlalchemy import Row, RowMapping
|
|
from pydantic import BaseModel, Field
|
|
|
|
# Importações do seu próprio projeto
|
|
from app.routers.dependencies import get_repository
|
|
from app.database.RepositoryBase import RepositoryBase
|
|
from app.database.RelationalTableRepository import RelationalTableRepository
|
|
from app.rbac.permissions import verify_permissions
|
|
from app.database.models import RbacUser
|
|
from app.config import COLUNA
|
|
from app.database.formatar_retorno_bd import formatters_map
|
|
|
|
|
|
# Esquema para o corpo da requisição
|
|
class FilterCondition(BaseModel):
|
|
"""
|
|
Representa uma condição de filtro individual.
|
|
"""
|
|
column: str = Field(..., description="Caminho completo da coluna (exemplo: 'relacao_setor.setor_nome').")
|
|
value: Any = Field(..., description="Valor a ser usado no filtro.")
|
|
operator: str = Field("==", description="Operador lógico (exemplo: '==', '!=', '<', '>', '>=', '<=').")
|
|
logical: str = Field("AND", description="Operador lógico para combinações ('AND', 'OR' etc).")
|
|
|
|
|
|
class FilterRequest(BaseModel):
|
|
"""
|
|
Representa a requisição contendo filtros dinâmicos.
|
|
"""
|
|
filters: Optional[List[FilterCondition]] = Field(
|
|
None,
|
|
description="Lista de condições de filtro. Cada condição contém coluna, operador, valor, e operador lógico."
|
|
)
|
|
order_by: Optional[List[str]] = Field(
|
|
None, description="Lista de colunas para ordenação dos registros."
|
|
)
|
|
ascending: Optional[List[bool]] = Field(
|
|
None, description="Lista de direções da ordenação. True para ascendente, False para descendente. "
|
|
"Deve corresponder ao tamanho de 'order_by'."
|
|
)
|
|
relationships: Optional[List[str]] = Field(
|
|
None, description="Lista de nomes de relacionamentos para incluir na consulta."
|
|
)
|
|
|
|
|
|
def create_dynamic_router(
|
|
rota_prefix: str,
|
|
create,
|
|
request,
|
|
id_many,
|
|
id_one,
|
|
update,
|
|
updates,
|
|
db_model,
|
|
ativo: Optional = None,
|
|
rota_tags: Optional[List[Union[str, Enum]]] = None,
|
|
request_formatado: Optional = None,
|
|
permissao_total: Optional[int] = None,
|
|
permissao_endpoint: Optional[int] = None,
|
|
permissao_setor: Optional[int] = None,
|
|
permissao_especifica_inicial: Optional[int] = None,
|
|
ordenacao: Optional[str] = None,
|
|
formatter_keys: Optional[dict[str, str]] = None,
|
|
related_info_append: Optional[List[Dict[str, Any]]] = None, # Dados extras para append
|
|
related_info_add: Optional[List[Dict[str, Any]]] = None, # Dados extras para add
|
|
related_info_extra_columns: Optional[List[Dict[str, Any]]] = None # Dados extras para
|
|
):
|
|
# Verifica automaticamente se deve usar RelationalTableRepository
|
|
use_complex_repo = bool(related_info_append or related_info_add or related_info_extra_columns)
|
|
|
|
# Define o repositório com base na necessidade de relacionamentos complexos
|
|
if use_complex_repo:
|
|
repository_base = Annotated[
|
|
RelationalTableRepository[db_model],
|
|
Depends(get_repository(db_model, RelationalTableRepository)),
|
|
]
|
|
else:
|
|
repository_base = Annotated[
|
|
RepositoryBase[db_model],
|
|
Depends(get_repository(db_model, RepositoryBase)),
|
|
]
|
|
|
|
router = APIRouter(
|
|
prefix=rota_prefix,
|
|
tags=rota_tags,
|
|
responses={404: {"description": "Not found"}},
|
|
)
|
|
|
|
permissao_especifica_atual = permissao_especifica_inicial or 0
|
|
|
|
def get_permission_list():
|
|
"""Retorna a lista de permissões atualizada e incrementa a permissão específica."""
|
|
nonlocal permissao_especifica_atual
|
|
|
|
# Construir a lista de permissões baseada nos valores informados
|
|
permission_list = [
|
|
permissao_total,
|
|
permissao_endpoint,
|
|
permissao_setor,
|
|
permissao_especifica_atual if permissao_especifica_inicial is not None else None
|
|
]
|
|
# Incrementa a permissão específica somente se foi inicialmente definida
|
|
if permissao_especifica_inicial is not None:
|
|
permissao_especifica_atual += 1
|
|
|
|
# Filtrar permissões nulas
|
|
final_permissions = [perm for perm in permission_list if perm is not None]
|
|
|
|
return final_permissions
|
|
|
|
@router.post("/add_one", status_code=status.HTTP_201_CREATED, response_model=request,
|
|
# dependencies=[Depends(verify_permissions(get_permission_list()))]
|
|
)
|
|
async def create_one(data: create, repository: repository_base,
|
|
_user: RbacUser = Depends(verify_permissions(get_permission_list()))) -> db_model:
|
|
if use_complex_repo:
|
|
resultado = await repository.create(
|
|
data_one=data,
|
|
db_data=db_model,
|
|
related_info_append=related_info_append,
|
|
related_info_add=related_info_add,
|
|
related_info_extra_columns=related_info_extra_columns
|
|
)
|
|
else:
|
|
resultado = await repository.create(data_one=data)
|
|
|
|
return resultado
|
|
|
|
@router.post("/add_many", status_code=status.HTTP_201_CREATED, response_model=list[request])
|
|
async def create_many(datas: List[create], repository: repository_base,
|
|
_user: RbacUser = Depends(verify_permissions(get_permission_list()))) -> (
|
|
list[db_model] | bool):
|
|
if use_complex_repo:
|
|
resultado = await repository.create_many(
|
|
data=datas,
|
|
db_data=db_model,
|
|
return_models=True,
|
|
related_info_append=related_info_append,
|
|
related_info_add=related_info_add,
|
|
related_info_extra_columns=related_info_extra_columns
|
|
)
|
|
else:
|
|
resultado = await repository.create_many(data=datas, return_models=True)
|
|
return resultado
|
|
|
|
permission_list_iguais = get_permission_list() # Obtém a lista de permissões para as funções get_all e get_filter
|
|
|
|
if request_formatado and formatter_keys and "get_all" in formatter_keys:
|
|
formatter_function = formatters_map[formatter_keys["get_all"]]
|
|
|
|
@router.post("/get_all", status_code=status.HTTP_200_OK, response_model=list[request_formatado])
|
|
async def get_all(repository: repository_base,
|
|
_user: RbacUser = Depends(verify_permissions(permission_list_iguais))) \
|
|
-> Sequence[Row[Any] | RowMapping | Any]:
|
|
resultado = await repository.get_many_by_ids(coluna=COLUNA, order_by=ordenacao)
|
|
formatado = formatter_function(resultado)
|
|
return formatado
|
|
|
|
@router.post("/get_filter", status_code=status.HTTP_200_OK, response_model=list[request_formatado])
|
|
async def get_all(repository: repository_base, filtro: Optional[FilterRequest] = None,
|
|
_user: RbacUser = Depends(verify_permissions(permission_list_iguais))) -> (
|
|
Sequence)[Row[Any] | RowMapping | Any]:
|
|
|
|
if not filtro:
|
|
resultado = await repository.get_many_by_ids(coluna=COLUNA, order_by=ordenacao)
|
|
formatado = formatter_function(resultado)
|
|
# Chamando a função do repositório com os filtros e ordenação
|
|
else:
|
|
resultado = await repository.get_filter(
|
|
coluna=COLUNA,
|
|
filters=filtro.filters,
|
|
order_by=filtro.order_by,
|
|
ascending=filtro.ascending
|
|
)
|
|
formatado = formatter_function(resultado)
|
|
return formatado
|
|
|
|
else:
|
|
@router.post("/get_all", status_code=status.HTTP_200_OK, response_model=list[request])
|
|
async def get_all(repository: repository_base,
|
|
_user: RbacUser = Depends(verify_permissions(permission_list_iguais))) -> (
|
|
Sequence)[Row[Any] | RowMapping | Any]:
|
|
resultado = await repository.get_many_by_ids(coluna=COLUNA, order_by=ordenacao)
|
|
return resultado
|
|
|
|
@router.post("/get_filter", status_code=status.HTTP_200_OK, response_model=list[request])
|
|
async def get_all(repository: repository_base, filtro: Optional[FilterRequest] = None,
|
|
_user: RbacUser = Depends(verify_permissions(permission_list_iguais))) -> (
|
|
Sequence)[Row[Any] | RowMapping | Any]:
|
|
|
|
if not filtro:
|
|
resultado = await repository.get_many_by_ids(coluna=COLUNA, order_by=ordenacao)
|
|
# Chamando a função do repositório com os filtros e ordenação
|
|
else:
|
|
resultado = await repository.get_filter(
|
|
coluna=COLUNA,
|
|
filters=filtro.filters,
|
|
order_by=filtro.order_by,
|
|
ascending=filtro.ascending
|
|
)
|
|
return resultado
|
|
|
|
@router.post("/get_many", status_code=status.HTTP_200_OK, response_model=List[request])
|
|
async def get_many(data: id_many, repository: repository_base,
|
|
_user: RbacUser = Depends(verify_permissions(get_permission_list()))) -> (
|
|
Sequence[Row[Any] | RowMapping | Any]):
|
|
resultado = await repository.get_many_by_ids(uuids=data.uuids, coluna=COLUNA, order_by=ordenacao)
|
|
return resultado
|
|
|
|
@router.post("/get_one", status_code=status.HTTP_200_OK, response_model=request)
|
|
async def get_one(data: id_one, repository: repository_base,
|
|
_user: RbacUser = Depends(verify_permissions(get_permission_list()))) -> db_model:
|
|
|
|
resultado = await repository.get_one_by_id(uuid=data.uuid, coluna=COLUNA)
|
|
return resultado
|
|
|
|
@router.put("/update_one", status_code=status.HTTP_201_CREATED, response_model=request)
|
|
async def update_one(data: update, repository: repository_base,
|
|
_user: RbacUser = Depends(verify_permissions(get_permission_list()))) -> db_model:
|
|
if use_complex_repo:
|
|
resultado = await repository.update_by_id(
|
|
update=data,
|
|
coluna=COLUNA,
|
|
db_data=db_model,
|
|
related_info_append=related_info_append,
|
|
related_info_add=related_info_add,
|
|
related_info_extra_columns=related_info_extra_columns
|
|
)
|
|
else:
|
|
resultado = await repository.update_by_id(update=data, coluna=COLUNA)
|
|
return resultado
|
|
|
|
if not use_complex_repo:
|
|
@router.put("/update_many", status_code=status.HTTP_201_CREATED, response_model=List[request])
|
|
async def update_many(data: List[updates], repository: repository_base,
|
|
_user: RbacUser = Depends(verify_permissions(get_permission_list()))) -> (
|
|
list[db_model] | db_model):
|
|
resultado = await repository.update_many_by_ids(updates=data, coluna=COLUNA, return_models=True)
|
|
return resultado
|
|
else:
|
|
# Não registramos a rota update_many, mas consumimos o efeito colateral da chamada
|
|
# de get_permission_list() para que o contador seja incrementado.
|
|
_loopPermissoes = get_permission_list()
|
|
|
|
if ativo is not None:
|
|
@router.put("/desativar", status_code=status.HTTP_201_CREATED, response_model=request)
|
|
async def desativar(data: ativo, repository: repository_base,
|
|
_user: RbacUser = Depends(verify_permissions(get_permission_list()))) -> db_model:
|
|
resultado = await repository.desativar_registro(update=data, coluna=COLUNA)
|
|
return resultado
|
|
|
|
@router.put("/ativar", status_code=status.HTTP_201_CREATED, response_model=request)
|
|
async def ativar(data: ativo, repository: repository_base,
|
|
_user: RbacUser = Depends(verify_permissions([permissao_total]))) -> db_model:
|
|
resultado = await repository.ativar_registro(update=data, coluna=COLUNA)
|
|
return resultado
|
|
else:
|
|
# Não registramos a rota desativar, mas consumimos o efeito colateral da chamada
|
|
# de get_permission_list() para que o contador seja incrementado.
|
|
_loopPermissoes = get_permission_list()
|
|
|
|
@router.delete("/delete_one", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_one(data: id_one, repository: repository_base,
|
|
_user: RbacUser = Depends(verify_permissions([permissao_total]))) -> Response:
|
|
await repository.remove_by_id(uuid=data.uuid, coluna=COLUNA)
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
|
|
|
@router.delete("/delete_many", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_many(data: id_many, repository: repository_base,
|
|
_user: RbacUser = Depends(verify_permissions([permissao_total]))) -> Response:
|
|
await repository.remove_many_by_ids(uuids=data.uuids, coluna=COLUNA)
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
|
|
|
return router
|