[MVPTN-58] feat(CrudTipoReconhecimento): Criação do CRUD para a tabela T_TB_RECONHECIMENTOTIPO

This commit is contained in:
Kenio 2025-09-10 15:21:09 -03:00
parent f7275c438f
commit 498e86d16a
24 changed files with 1038 additions and 16 deletions

View file

@ -0,0 +1,26 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoIdSchema
from packages.v1.administrativo.repositories.t_tb_reconhecimentotipo.t_tb_reconhecimentotipo_delete_repository import DeleteRepository
class DeleteAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela t_tb_reconhecimentotipo.
"""
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
reconhecimentotipo_schema (T_TbReconhecimentotipoIdSchema): O esquema com o ID a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# Instanciamento do repositório
delete_repository = DeleteRepository()
# Execução do repositório
return delete_repository.execute(reconhecimentotipo_schema)

View file

@ -0,0 +1,28 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoDescricaoSchema
from packages.v1.administrativo.repositories.t_tb_reconhecimentotipo.t_tb_reconhecimentotipo_get_by_descricao_repository import GetByDescricaoRepository
class GetByDescricaoService(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela t_tb_reconhecimentotipo por descrição.
"""
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoDescricaoSchema):
"""
Executa a operação de busca no banco de dados.
Args:
reconhecimentotipo_schema (T_TbReconhecimentotipoDescricaoSchema): O esquema com a descrição a ser buscada.
Returns:
O registro encontrado ou None.
"""
# Instanciamento do repositório
show_repository = GetByDescricaoRepository()
# Execução do repositório
response = show_repository.execute(reconhecimentotipo_schema)
# Retorno da informação
return response

View file

@ -0,0 +1,24 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.t_tb_reconhecimentotipo.t_tb_reconhecimentotipo_index_repository import IndexRepository
class IndexService(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de todos os registros na tabela t_tb_reconhecimentotipo.
"""
def execute(self):
"""
Executa a operação de listagem no banco de dados.
Returns:
A lista de todos os registros.
"""
# Instanciamento do repositório
index_repository = IndexRepository()
# Execução do repositório
response = index_repository.execute()
# Retorno da informação
return response

View file

@ -0,0 +1,28 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoSaveSchema
from packages.v1.administrativo.repositories.t_tb_reconhecimentotipo.t_tb_reconhecimentotipo_save_repository import SaveRepository
class SaveService(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvar um novo registro na tabela t_tb_reconhecimentotipo.
"""
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoSaveSchema):
"""
Executa a operação de salvamento.
Args:
reconhecimentotipo_schema (T_TbReconhecimentotipoSaveSchema): O esquema com os dados a serem salvos.
Returns:
O resultado da operação de salvamento.
"""
# Instânciamento do repositório
save_repository = SaveRepository()
# Execução do repositório
response = save_repository.execute(reconhecimentotipo_schema)
# Retorno da informação
return response

View file

@ -0,0 +1,28 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoIdSchema
from packages.v1.administrativo.repositories.t_tb_reconhecimentotipo.t_tb_reconhecimentotipo_show_repository import ShowRepository
class ShowService(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela t_tb_reconhecimentotipo.
"""
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoIdSchema):
"""
Executa a operação de exibição.
Args:
reconhecimentotipo_schema (T_TbReconhecimentotipoIdSchema): O esquema com o ID do registro a ser exibido.
Returns:
O resultado da operação de exibição.
"""
# Instânciamento do repositório
show_repository = ShowRepository()
# Execução do repositório
response = show_repository.execute(reconhecimentotipo_schema)
# Retorno da informação
return response

View file

@ -0,0 +1,25 @@
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoUpdateSchema
from packages.v1.administrativo.repositories.t_tb_reconhecimentotipo.t_tb_reconhecimentotipo_update_repository import UpdateRepository
class TTbReconhecimentotipoUpdateService:
"""
Service responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela t_tb_reconhecimentotipo.
"""
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoUpdateSchema):
"""
Executa a operação de atualização.
Args:
reconhecimentotipo_schema (T_TbReconhecimentotipoUpdateSchema): O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# Instância o repositório de atualização
update_repository = UpdateRepository()
# Chama o método de execução do repositório para realizar a atualização
return update_repository.execute(reconhecimentotipo_schema)

View file

@ -0,0 +1,112 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import (
TTbReconhecimentotipoSchema,
TTbReconhecimentotipoSaveSchema,
TTbReconhecimentotipoUpdateSchema,
TTbReconhecimentotipoIdSchema
)
class TTbReconhecimentotipoController:
def __init__(self):
# Action responsável por carregar as services de acordo com o estado
self.dynamic_import = DynamicImport()
# Define o pacote que deve ser carregado
self.dynamic_import.set_package("administrativo")
# Define a tabela que o pacote pertence
self.dynamic_import.set_table("t_tb_reconhecimentotipo")
pass
# Lista todos os tipos de reconhecimento
def index(self):
# Importação da classe desejada
indexService = self.dynamic_import.service("t_tb_reconhecimentotipo_index_service", "IndexService")
# Instância da classe service
self.indexService = indexService()
# Lista todos os tipos de reconhecimento
return {
'message': 'Tipos de reconhecimento localizados com sucesso',
'data': self.indexService.execute()
}
# Busca um tipo de reconhecimento específico pelo ID
def show(self, reconhecimentotipo_schema : TTbReconhecimentotipoIdSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('t_tb_reconhecimentotipo_show_service', 'ShowService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o tipo de reconhecimento desejado
return {
'message' : 'Tipo de reconhecimento localizado com sucesso',
'data': self.show_service.execute(reconhecimentotipo_schema)
}
# Busca um tipo de reconhecimento pela descrição
def get_by_descricao(self, descricao : str):
#Importação da classe desejada
show_service = self.dynamic_import.service('t_tb_reconhecimentotipo_get_by_descricao_service', 'GetByDescricaoService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o tipo de reconhecimento desejado
return {
'message' : 'Tipo de reconhecimento localizado com sucesso',
'data': self.show_service.execute(descricao)
}
# Cadastra um novo tipo de reconhecimento
def save(self, reconhecimentotipo_schema : TTbReconhecimentotipoSaveSchema):
#Importação da classe desejada
save_service = self.dynamic_import.service('t_tb_reconhecimentotipo_save_service', 'TTbReconhecimentotipoSaveService')
# Instância da classe desejada
self.save_service = save_service()
# Busca e retorna o tipo de reconhecimento desejado
return {
'message' : 'Tipo de reconhecimento salvo com sucesso',
'data': self.save_service.execute(reconhecimentotipo_schema)
}
# Atualiza os dados de um tipo de reconhecimento
def update(self, reconhecimentotipo_schema : TTbReconhecimentotipoUpdateSchema):
#Importação da classe desejada
update_service = self.dynamic_import.service('t_tb_reconhecimentotipo_update_service', 'TTbReconhecimentotipoUpdateService')
# Instância da classe desejada
self.update_service = update_service()
# Busca e retorna o tipo de reconhecimento desejado
return {
'message' : 'Tipo de reconhecimento atualizado com sucesso',
'data': self.update_service.execute(reconhecimentotipo_schema)
}
# Exclui um tipo de reconhecimento
def delete(self, reconhecimentotipo_schema : TTbReconhecimentotipoIdSchema):
#Importação da classe desejada
delete_service = self.dynamic_import.service('t_tb_reconhecimentotipo_delete_service', 'DeleteService')
# Instância da classe desejada
self.delete_service = delete_service()
# Busca e retorna o tipo de reconhecimento desejado
return {
'message' : 'Tipo de reconhecimento removido com sucesso',
'data': self.delete_service.execute(reconhecimentotipo_schema)
}

View file

@ -0,0 +1,105 @@
# Importação de bibliotecas
from typing import Optional
from fastapi import APIRouter, Body, Depends, status
from actions.jwt.get_current_user import get_current_user
from packages.v1.administrativo.controllers.t_tb_reconhecimentotipo_controller import TTbReconhecimentotipoController
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import (
TTbReconhecimentotipoSchema,
TTbReconhecimentotipoSaveSchema,
TTbReconhecimentotipoUpdateSchema,
TTbReconhecimentotipoIdSchema
)
# Inicializa o roteador para as rotas do tipo de reconhecimento
router = APIRouter()
# Instânciamento do controller desejado
t_tb_reconhecimentotipo_controller = TTbReconhecimentotipoController()
# Lista todos os tipos de reconhecimento
@router.get('/',
status_code=status.HTTP_200_OK,
summary='Lista todos os tipos de reconhecimento cadastrados',
response_description='Lista todos os tipos de reconhecimento cadastrados')
async def index(current_user: dict = Depends(get_current_user)):
# Busca todos os tipos de reconhecimento cadastrados
response = t_tb_reconhecimentotipo_controller.index()
# Retorna os dados localizados
return response
# Localiza um tipo de reconhecimento pela descrição
@router.get('/descricao',
status_code=status.HTTP_200_OK,
summary='Busca um registro em específico pela descrição',
response_description='Busca um registro em específico')
async def get_by_descricao(descricao : str, current_user: dict = Depends(get_current_user)):
# Busca um tipo de reconhecimento específico pela descrição
response = t_tb_reconhecimentotipo_controller.get_by_descricao(descricao)
# Retorna os dados localizados
return response
# Localiza um tipo de reconhecimento pelo ID
@router.get('/{tb_reconhecimentotipo_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro em específico pelo ID do tipo de reconhecimento',
response_description='Busca um registro em específico')
async def show(tb_reconhecimentotipo_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
reconhecimentotipo_schema = TTbReconhecimentotipoIdSchema(tb_reconhecimentotipo_id=tb_reconhecimentotipo_id)
# Busca um tipo de reconhecimento específico pelo ID
response = t_tb_reconhecimentotipo_controller.show(reconhecimentotipo_schema)
# Retorna os dados localizados
return response
# Cadastro de tipo de reconhecimento
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um tipo de reconhecimento',
response_description='Cadastra um tipo de reconhecimento')
async def save(reconhecimentotipo_schema : TTbReconhecimentotipoSaveSchema, current_user: dict = Depends(get_current_user)):
# Efetua o cadastro no banco de dados
response = t_tb_reconhecimentotipo_controller.save(reconhecimentotipo_schema)
# Retorna os dados localizados
return response
# Atualiza os dados de um tipo de reconhecimento
@router.put('/',
status_code=status.HTTP_200_OK,
summary='Atualiza um tipo de reconhecimento',
response_description='Atualiza um tipo de reconhecimento')
async def update(reconhecimentotipo_schema : TTbReconhecimentotipoUpdateSchema, current_user: dict = Depends(get_current_user)):
# Efetua a atualização dos dados
response = t_tb_reconhecimentotipo_controller.update(reconhecimentotipo_schema)
# Retorna os dados localizados
return response
# Exclui um determinado tipo de reconhecimento
@router.delete('/{tb_reconhecimentotipo_id}',
status_code=status.HTTP_200_OK,
summary='Remove um tipo de reconhecimento',
response_description='Remove um tipo de reconhecimento')
async def delete(tb_reconhecimentotipo_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
reconhecimentotipo_schema = TTbReconhecimentotipoIdSchema(tb_reconhecimentotipo_id=tb_reconhecimentotipo_id)
# Efetua a exclusão do tipo de reconhecimento
response = t_tb_reconhecimentotipo_controller.delete(reconhecimentotipo_schema)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,41 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoIdSchema
from fastapi import HTTPException, status
class DeleteRepository(BaseRepository):
"""
Repositório para a operação de exclusão de um registro na tabela
t_tb_reconhecimentotipo.
"""
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoIdSchema):
"""
Executa a consulta SQL para remover um registro pelo ID.
Args:
reconhecimentotipo_schema (T_TbReconhecimentotipoIdSchema): O esquema com o ID a ser removido.
Returns:
O resultado da operação de exclusão.
"""
try:
# Montagem do sql
sql = """ DELETE FROM T_TB_RECONHECIMENTOTIPO WHERE TB_RECONHECIMENTOTIPO_ID = :tb_reconhecimentotipo_id """
# Preenchimento de parâmetros
params = {
"tb_reconhecimentotipo_id": reconhecimentotipo_schema.tb_reconhecimentotipo_id
}
# Execução do sql
response = self.run(sql, params)
# Retorna o resultado
return response
except Exception as e:
# Informa que houve uma falha na exclusão
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao excluir tipo de reconhecimento: {e}"
)

View file

@ -0,0 +1,29 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoDescricaoSchema
class GetByDescricaoRepository(BaseRepository):
"""
Repositório para a operação de busca de um registro na tabela
t_tb_reconhecimentotipo por descrição.
"""
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoDescricaoSchema):
"""
Executa a consulta SQL para buscar um registro pela descrição.
Args:
reconhecimentotipo_schema (T_TbReconhecimentotipoDescricaoSchema): O esquema com a descrição a ser buscada.
Returns:
Um dicionário contendo os dados do registro ou None se não for encontrado.
"""
# Montagem do SQL
sql = """ SELECT * FROM T_TB_RECONHECIMENTOTIPO WHERE DESCRICAO = :descricao """
# Preenchimento de parâmetros
params = {
'descricao': reconhecimentotipo_schema.descricao
}
# Execução do sql
return self.fetch_one(sql, params)

View file

@ -0,0 +1,23 @@
from abstracts.repository import BaseRepository
class IndexRepository(BaseRepository):
"""
Repositório para a operação de listagem de todos os registros
na tabela t_tb_reconhecimentotipo.
"""
def execute(self):
"""
Executa a consulta SQL para buscar todos os registros.
Returns:
Uma lista de dicionários contendo os dados dos registros.
"""
# Montagem do SQL
sql = """ SELECT * FROM T_TB_RECONHECIMENTOTIPO """
# Execução do sql
response = self.fetch_all(sql)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,53 @@
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoSaveSchema
class SaveRepository(BaseRepository):
"""
Repositório para a operação de salvamento de um novo registro na tabela t_tb_reconhecimentotipo.
"""
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoSaveSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
reconhecimentotipo_schema (T_TbReconhecimentotipoSaveSchema): O esquema com os dados a serem salvos.
Returns:
O registro recém-criado.
Raises:
HTTPException: Caso ocorra um erro na execução da query.
"""
try:
# Montagem do SQL
sql = """ INSERT INTO T_TB_RECONHECIMENTOTIPO(
TB_RECONHECIMENTOTIPO_ID,
DESCRICAO,
SITUACAO
) VALUES (
:tb_reconhecimentotipo_id,
:descricao,
:situacao
) RETURNING *;"""
# Preenchimento de parâmetros
params = {
'tb_reconhecimentotipo_id': reconhecimentotipo_schema.tb_reconhecimentotipo_id,
'descricao': reconhecimentotipo_schema.descricao,
'situacao': reconhecimentotipo_schema.situacao
}
# Execução do sql
return self.run_and_return(sql, params)
except Exception as e:
# Informa que houve uma falha no salvamento do registro
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao salvar registro: {e}"
)

View file

@ -0,0 +1,46 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoIdSchema
from fastapi import HTTPException, status
class ShowRepository(BaseRepository):
"""
Repositório para a operação de exibição de um registro na tabela t_tb_reconhecimentotipo.
"""
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoIdSchema):
"""
Busca um tipo de reconhecimento específico pelo ID.
Args:
reconhecimentotipo_schema (T_TbReconhecimentotipoIdSchema): O esquema que contém o ID do registro.
Returns:
O registro encontrado ou None se não existir.
Raises:
HTTPException: Caso ocorra um erro na execução da query.
"""
try:
# Montagem do SQL
sql = "SELECT * FROM T_TB_RECONHECIMENTOTIPO WHERE TB_RECONHECIMENTOTIPO_ID = :tb_reconhecimentotipo_id"
# Preenchimento de parâmetros
params = {
'tb_reconhecimentotipo_id': reconhecimentotipo_schema.tb_reconhecimentotipo_id
}
# Execução do SQL
result = self.fetch_one(sql, params)
if not result:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Registro não encontrado"
)
return result
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Erro ao buscar registro: {str(e)}"
)

View file

@ -0,0 +1,60 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoUpdateSchema
from fastapi import HTTPException, status
class UpdateRepository(BaseRepository):
"""
Repositório para a operação de atualização na tabela T_TB_RECONHECIMENTOTIPO.
"""
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoUpdateSchema):
"""
Executa a atualização de um registro na tabela.
Args:
reconhecimentotipo_schema (T_TbReconhecimentotipoUpdateSchema): O esquema com os dados a serem atualizados.
Returns:
O registro atualizado.
Raises:
HTTPException: Se o registro não for encontrado ou ocorrer um erro na atualização.
"""
try:
updates = []
params = {}
if reconhecimentotipo_schema.descricao is not None:
updates.append("DESCRICAO = :descricao")
params["descricao"] = reconhecimentotipo_schema.descricao
if reconhecimentotipo_schema.situacao is not None:
updates.append("SITUACAO = :situacao")
params["situacao"] = reconhecimentotipo_schema.situacao
if not updates:
return False
params["tb_reconhecimentotipo_id"] = reconhecimentotipo_schema.tb_reconhecimentotipo_id
sql = f"UPDATE T_TB_RECONHECIMENTOTIPO SET {', '.join(updates)} WHERE tb_reconhecimentotipo_id = :tb_reconhecimentotipo_id RETURNING *;"
# Executa a query
result = self.run_and_return(sql, params)
if not result.tb_reconhecimentotipo_id:
# Informa que não existe o registro a ser modificado
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail='Nenhum tipo de reconhecimento localizado para esta solicitação'
)
# Se houver um resultado, a atualização foi bem-sucedida
if result:
return result
except Exception as e:
# Informa que houve uma falha na atualização
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao atualizar o tipo de reconhecimento: {e}"
)

View file

@ -1,16 +0,0 @@
from datetime import date
from typing import Optional
from pydantic import BaseModel
class TTBReconhecimentoTipoSchema(BaseModel):
"""
Schema para a tabela T_TB_RECONHECIMENTOTIPO.
"""
tb_reconhecimentotipo_id: Optional[int] = None
descricao: Optional[str] = None
situacao: Optional[str] = None
class Config:
from_attributes = True

View file

@ -0,0 +1,105 @@
from pydantic import BaseModel, field_validator, model_validator
from fastapi import HTTPException, status
from typing import Optional
# Funções para sanitização de entradas (evitar XSS, SQLi etc.)
# Supondo que a classe Text esteja disponível e com o método sanitize_input
try:
from actions.validations.text import Text
except ImportError:
class Text:
@staticmethod
def sanitize_input(value: str) -> str:
return value.strip()
# ----------------------------------------------------
# Schema base
# ----------------------------------------------------
class TTbReconhecimentotipoSchema(BaseModel):
tb_reconhecimentotipo_id: Optional[int] = None
descricao: Optional[str] = None
situacao: Optional[str] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar um tipo especifico pelo ID (GET)
# ----------------------------------------------------
class TTbReconhecimentotipoIdSchema(BaseModel):
tb_reconhecimentotipo_id: int
# ----------------------------------------------------
# Schema para criação de novo tipo (POST)
# ----------------------------------------------------
class TTbReconhecimentotipoSaveSchema(BaseModel):
tb_reconhecimentotipo_id: Optional[int] = None
descricao: str
situacao: str
# Sanitiza os inputs enviados
@field_validator('descricao', 'situacao')
def sanitize_fields(cls, v):
if v:
return Text.sanitize_input(v)
return v
# Verifica se os campos obrigatórios foram enviados
@model_validator(mode='after')
def validate_all_fields(self):
errors = []
if not self.descricao or len(self.descricao.strip()) == 0:
errors.append({'input': 'descricao', 'message': 'A descrição é obrigatória.'})
if not self.situacao or len(self.situacao.strip()) == 0:
errors.append({'input': 'situacao', 'message': 'A situação é obrigatória.'})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self
# ----------------------------------------------------
# Schema para atualizar tipo (PUT)
# ----------------------------------------------------
class TTbReconhecimentotipoUpdateSchema(BaseModel):
tb_reconhecimentotipo_id: int
descricao: Optional[str] = None
situacao: Optional[str] = None
# Sanitiza os inputs enviados
@field_validator('descricao', 'situacao')
def sanitize_fields(cls, v):
if v:
return Text.sanitize_input(v)
return v
# Verifica se os campos obrigatórios foram enviados
@model_validator(mode='after')
def validate_all_fields(self):
errors = []
if not self.tb_reconhecimentotipo_id or self.tb_reconhecimentotipo_id == 0:
errors.append({'input': 'tb_reconhecimentotipo_id', 'message': 'O ID do tipo é obrigatório.'})
if not self.descricao or len(self.descricao.strip()) == 0:
errors.append({'input': 'descricao', 'message': 'A descrição é obrigatória.'})
if not self.situacao or len(self.situacao.strip()) == 0:
errors.append({'input': 'situacao', 'message': 'A situação é obrigatória.'})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self

View file

@ -0,0 +1,27 @@
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoIdSchema
from packages.v1.administrativo.actions.t_tb_reconhecimentotipo.t_tb_reconhecimentotipo_delete_action import DeleteAction
class T_TbReconhecimentotipoDeleteService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela t_tb_reconhecimentotipo.
"""
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoIdSchema):
"""
Executa a operação de exclusão do registro no banco de dados.
Args:
reconhecimentotipo_schema (T_TbReconhecimentotipoIdSchema): O esquema com o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# Instanciamento da ação
delete_action = DeleteAction()
# Executa a ação em questão
data = delete_action.execute(reconhecimentotipo_schema)
# Retorno da informação
return data

View file

@ -0,0 +1,38 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoDescricaoSchema
from packages.v1.administrativo.actions.t_tb_reconhecimentotipo.t_tb_reconhecimentotipo_get_by_descricao_action import GetByDescricaoAction
class T_TbReconhecimentotipoGetByDescricaoService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela t_tb_reconhecimentotipo pela sua descrição.
"""
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoDescricaoSchema, messageValidate: bool):
"""
Executa a operação de busca no banco de dados.
Args:
reconhecimentotipo_schema (T_TbReconhecimentotipoDescricaoSchema): O esquema com a descrição a ser buscada.
messageValidate (bool): Se True, lança uma exceção HTTP caso o registro não seja encontrado.
Returns:
O registro encontrado ou None.
"""
# Instanciamento da ação
show_action = GetByDescricaoAction()
# Executa a ação em questão
data = show_action.execute(reconhecimentotipo_schema)
if messageValidate:
if not data:
# Retorna uma exceção
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail='Não foi possível localizar o registro de tipo de reconhecimento'
)
# Retorno da informação
return data

View file

@ -0,0 +1,32 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.actions.t_tb_reconhecimentotipo.t_tb_reconhecimentotipo_index_action import IndexAction
class T_TbReconhecimentotipoIndexService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de registros na tabela t_tb_reconhecimentotipo.
"""
def execute(self):
"""
Executa a operação de busca de todos os registros no banco de dados.
Returns:
A lista de registros encontrados.
"""
# Instanciamento da ação
index_action = IndexAction()
# Executa a busca de todas as ações
data = index_action.execute()
# Verifica se foram localizados registros
if not data:
# Retorna uma exceção
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail='Não foi possível localizar os registros de tipo de reconhecimento'
)
# Retorna as informações localizadas
return data

View file

@ -0,0 +1,70 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.sequencia.schemas.g_sequencia import GSequenciaSchema
from packages.v1.sequencia.services.g_sequencia.generate_service import GenerateService
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoSaveSchema, T_TbReconhecimentotipoDescricaoSchema
from packages.v1.administrativo.actions.t_tb_reconhecimentotipo.t_tb_reconhecimentotipo_save_action import SaveAction
from fastapi import HTTPException, status
class T_TbReconhecimentotipoSaveService:
def __init__(self):
# Ação responsável por carregar as services de acordo com o estado
self.dynamic_import = DynamicImport()
# Define o pacote que deve ser carregado
self.dynamic_import.set_package("administrativo")
# Define a tabela que o pacote pertence
self.dynamic_import.set_table("t_tb_reconhecimentotipo")
pass
# Cadastra o novo tipo de reconhecimento
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoSaveSchema):
# Armazena possíveis erros
errors = []
# Verifica se a descrição já está sendo utilizada
# Importação de service
descricao_service = self.dynamic_import.service("t_tb_reconhecimentotipo_get_by_descricao_service", "GetDescricaoService")
# Instanciamento da service
self.descricao_service = descricao_service()
# Verifica se a descrição já está sendo utilizada
self.response = self.descricao_service.execute(T_TbReconhecimentotipoDescricaoSchema(descricao=reconhecimentotipo_schema.descricao), False)
# Se houver retorno significa que a descrição já está sendo utilizada
if self.response:
errors.append({'input': 'descricao', 'message': 'a descrição informada já está sendo utilizada.'})
# Se houver erros, lança a exceção
if errors:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=errors
)
# Verifica se precisa gerar o ID de sequência
if not reconhecimentotipo_schema.tb_reconhecimentotipo_id:
# Crio um objeto de sequencia
sequencia_schema = GSequenciaSchema()
# Define os dados para atualizar a sequencia
sequencia_schema.tabela = 'T_TB_RECONHECIMENTOTIPO'
# Busco a sequência atualizada
generate = GenerateService()
# Busco a sequência atualizada
sequencia = generate.execute(sequencia_schema)
# Atualiza os dados da chave primária
reconhecimentotipo_schema.tb_reconhecimentotipo_id = sequencia.sequencia
# Instanciamento de ações
save_action = SaveAction()
# Retorna o resultado da operação
return save_action.execute(reconhecimentotipo_schema)

View file

@ -0,0 +1,35 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoIdSchema
from packages.v1.administrativo.actions.t_tb_reconhecimentotipo.t_tb_reconhecimentotipo_show_action import ShowAction
class T_TbReconhecimentotipoShowService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela t_tb_reconhecimentotipo.
"""
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoIdSchema):
"""
Executa a operação de busca no banco de dados.
Args:
reconhecimentotipo_schema (T_TbReconhecimentotipoIdSchema): O esquema com o ID a ser buscado.
Returns:
O resultado da busca.
"""
# Instanciamento da ação
show_action = ShowAction()
# Executa a ação em questão
data = show_action.execute(reconhecimentotipo_schema)
if not data:
# Retorna uma exceção
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail='Não foi possível localizar o registro de tipo de reconhecimento'
)
# Retorno da informação
return data

View file

@ -0,0 +1,23 @@
from packages.v1.administrativo.schemas.t_tb_reconhecimentotipo_schema import TTbReconhecimentotipoUpdateSchema
from packages.v1.administrativo.actions.t_tb_reconhecimentotipo.t_tb_reconhecimentotipo_update_action import UpdateAction
class T_TbReconhecimentotipoUpdateService:
"""
Serviço para a operação de atualização de um registro na tabela
t_tb_reconhecimentotipo.
"""
def execute(self, reconhecimentotipo_schema: TTbReconhecimentotipoUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
reconhecimentotipo_schema (T_TbReconhecimentotipoUpdateSchema): O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# Instanciamento de ações
update_action = UpdateAction()
# Retorna o resultado da operação
return update_action.execute(reconhecimentotipo_schema)

View file

@ -4,6 +4,7 @@ from fastapi import APIRouter
# Importa os módulos de rotas específicos
from packages.v1.administrativo.endpoints import c_caixa_item
from packages.v1.administrativo.endpoints import g_usuario
from packages.v1.administrativo.endpoints import t_tb_reconhecimentotipo_endpoint
# Cria uma instância do APIRouter que vai agregar todas as rotas da API
api_router = APIRouter()
@ -17,3 +18,8 @@ api_router.include_router(
api_router.include_router(
g_usuario.router, prefix="/administrativo/usuarios", tags=["Usuário"]
)
# Inclui as rotas de caixa
api_router.include_router(
t_tb_reconhecimentotipo_endpoint.router, prefix="/administrativo/t_tb_reconhecimentotipo", tags=["Tipos de reconhecimento"]
)

View file

@ -27929,5 +27929,79 @@
"accept-encoding": "gzip, deflate, br, zstd",
"cookie": "RememberAccess=S; sidebar_state=false; UserEmail=az77Wx6pC3ofs2cnusAFMr5Z0EVE%2BqQLmgs4BnkDej7TV7l%2Fs5uvHMFcG%2FS0B6ChDSOTJiRh6vu4hJgS4p3oNpnG2F3Z7829pcfLIFp7w%2BnfBGcvgbT4bjR2AeuN0dbbTNkDTxiVDVSEomj5I19Grg%3D%3D; UserPassword=ISpqgl%2FwbnR49GPRbd8maNZ%2BSZ2zz4eiUWF80t9MjEfVt0LxZEw%2FD3D%2BL8ke8eV%2BZn3mWiIYbNwGtEJ5rrMwouuTl3rQOOtfS29xodxJQgNsIEarjv9fcB8pwuVfHkTY; access_token=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ0eXBlIjoiYWNjZXNzLXRva2VuIiwiZXhwIjoxNzU2NDkyOTcyLCJpYXQiOjE3NTU3OTgxNzIsImRhdGEiOiJ7J3VzdWFyaW9faWQnOiAxMjM0NjcsICdsb2dpbic6ICdrZXZlbicsICdub21lJzogJ2tldmVuJ30ifQ.J34F8X5Vuil1c1AoSFdEzuYN7kg7n4BHpwcBn4q9uG0"
}
},
{
"method": "GET",
"url": "http://localhost:8000/docs",
"headers": {
"host": "localhost:8000",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:142.0) Gecko/20100101 Firefox/142.0",
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"accept-language": "pt-BR,pt;q=0.8,en-US;q=0.5,en;q=0.3",
"accept-encoding": "gzip, deflate, br, zstd",
"connection": "keep-alive",
"cookie": "UserEmail=X%2FOrAPDmZqxFGRaPqXvDtH1coSgFFmzv5cQHKCoO9t3k58jBwT5xcGMds49E0wrRm4cnAG2WdBFPL8l2IfXSBD%2BEey4%2FIEDPizQdOe1l%2FLFmTCxPF08IA1mqHyVl9gmY5eqrYLUXPpYwIdHQNf7vQA%3D%3D; UserPassword=EMXw00pesMlUHxM3y2tHO7gZLfHAmo1aeBGJxhnRao0KrWwtbHjbh1ywpeWUPEa%2BDJkyHOqjjL0KVRJTpaD0lcWL%2B8II6JiHJBx%2BDCyjO8lTgr5SU7VxThgHxTUKHIto; RememberAccess=S; next-auth.session-token=eyJhbGciOiJkaXIiLCJlbmMiOiJBMjU2R0NNIn0..IHS7Sgxy9Iv8_VnG.CFBNk9kYT82XneOjTLpJEu9fP2WLr_ApxKMo4CfwDIHo7EKzvqoe5icPHm5F0nCkwUct4LPkj5a5bYfPQ6RGVCoSzynQAEE6IdtAgs2UmVLxEOCf8kDyP20LswCg32M2eWQDCeYSAq_Pq8CehZEgW0CAeupJJzyy7HtmEKTOQztb1IogJP9e0-H90Gt6yYpZnGETK2_nbVJdiX7nVReTPKj1tMexshdDbX_T3xb4BFDMXaEoRTczZiWvoGe_Din6Z9mrZpYjBnOVrOehWQ9TJJ9M8orVZEtpPfM0e9iRIfrKUG1-3RWZMVlDP3X6MrzCmpu0mN7XN_qX3Eqdk_skdknHASPDhA.ahjUD942GZ36VL0v_KgRzw; acceptCookies=true; next-auth.csrf-token=f0e3369e77c3cc1f591e0d86bfc9c712239b5fb8e66eaa694cc1e58651471a39%7C290985f013fcc50f194726dcac2ea9d920bfc856addde580ffd04f962f4cc243; next-auth.callback-url=http%3A%2F%2Flocalhost%3A3000",
"upgrade-insecure-requests": "1",
"sec-fetch-dest": "document",
"sec-fetch-mode": "navigate",
"sec-fetch-site": "none",
"sec-fetch-user": "?1",
"priority": "u=0, i"
}
},
{
"method": "GET",
"url": "http://localhost:8000/openapi.json",
"headers": {
"host": "localhost:8000",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:142.0) Gecko/20100101 Firefox/142.0",
"accept": "application/json,*/*",
"accept-language": "pt-BR,pt;q=0.8,en-US;q=0.5,en;q=0.3",
"accept-encoding": "gzip, deflate, br, zstd",
"referer": "http://localhost:8000/docs",
"connection": "keep-alive",
"cookie": "UserEmail=X%2FOrAPDmZqxFGRaPqXvDtH1coSgFFmzv5cQHKCoO9t3k58jBwT5xcGMds49E0wrRm4cnAG2WdBFPL8l2IfXSBD%2BEey4%2FIEDPizQdOe1l%2FLFmTCxPF08IA1mqHyVl9gmY5eqrYLUXPpYwIdHQNf7vQA%3D%3D; UserPassword=EMXw00pesMlUHxM3y2tHO7gZLfHAmo1aeBGJxhnRao0KrWwtbHjbh1ywpeWUPEa%2BDJkyHOqjjL0KVRJTpaD0lcWL%2B8II6JiHJBx%2BDCyjO8lTgr5SU7VxThgHxTUKHIto; RememberAccess=S; next-auth.session-token=eyJhbGciOiJkaXIiLCJlbmMiOiJBMjU2R0NNIn0..IHS7Sgxy9Iv8_VnG.CFBNk9kYT82XneOjTLpJEu9fP2WLr_ApxKMo4CfwDIHo7EKzvqoe5icPHm5F0nCkwUct4LPkj5a5bYfPQ6RGVCoSzynQAEE6IdtAgs2UmVLxEOCf8kDyP20LswCg32M2eWQDCeYSAq_Pq8CehZEgW0CAeupJJzyy7HtmEKTOQztb1IogJP9e0-H90Gt6yYpZnGETK2_nbVJdiX7nVReTPKj1tMexshdDbX_T3xb4BFDMXaEoRTczZiWvoGe_Din6Z9mrZpYjBnOVrOehWQ9TJJ9M8orVZEtpPfM0e9iRIfrKUG1-3RWZMVlDP3X6MrzCmpu0mN7XN_qX3Eqdk_skdknHASPDhA.ahjUD942GZ36VL0v_KgRzw; acceptCookies=true; next-auth.csrf-token=f0e3369e77c3cc1f591e0d86bfc9c712239b5fb8e66eaa694cc1e58651471a39%7C290985f013fcc50f194726dcac2ea9d920bfc856addde580ffd04f962f4cc243; next-auth.callback-url=http%3A%2F%2Flocalhost%3A3000",
"sec-fetch-dest": "empty",
"sec-fetch-mode": "cors",
"sec-fetch-site": "same-origin",
"priority": "u=4"
}
},
{
"method": "GET",
"url": "http://localhost:8000/docs",
"headers": {
"host": "localhost:8000",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:142.0) Gecko/20100101 Firefox/142.0",
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"accept-language": "pt-BR,pt;q=0.8,en-US;q=0.5,en;q=0.3",
"accept-encoding": "gzip, deflate, br, zstd",
"connection": "keep-alive",
"cookie": "UserEmail=X%2FOrAPDmZqxFGRaPqXvDtH1coSgFFmzv5cQHKCoO9t3k58jBwT5xcGMds49E0wrRm4cnAG2WdBFPL8l2IfXSBD%2BEey4%2FIEDPizQdOe1l%2FLFmTCxPF08IA1mqHyVl9gmY5eqrYLUXPpYwIdHQNf7vQA%3D%3D; UserPassword=EMXw00pesMlUHxM3y2tHO7gZLfHAmo1aeBGJxhnRao0KrWwtbHjbh1ywpeWUPEa%2BDJkyHOqjjL0KVRJTpaD0lcWL%2B8II6JiHJBx%2BDCyjO8lTgr5SU7VxThgHxTUKHIto; RememberAccess=S; next-auth.session-token=eyJhbGciOiJkaXIiLCJlbmMiOiJBMjU2R0NNIn0..IHS7Sgxy9Iv8_VnG.CFBNk9kYT82XneOjTLpJEu9fP2WLr_ApxKMo4CfwDIHo7EKzvqoe5icPHm5F0nCkwUct4LPkj5a5bYfPQ6RGVCoSzynQAEE6IdtAgs2UmVLxEOCf8kDyP20LswCg32M2eWQDCeYSAq_Pq8CehZEgW0CAeupJJzyy7HtmEKTOQztb1IogJP9e0-H90Gt6yYpZnGETK2_nbVJdiX7nVReTPKj1tMexshdDbX_T3xb4BFDMXaEoRTczZiWvoGe_Din6Z9mrZpYjBnOVrOehWQ9TJJ9M8orVZEtpPfM0e9iRIfrKUG1-3RWZMVlDP3X6MrzCmpu0mN7XN_qX3Eqdk_skdknHASPDhA.ahjUD942GZ36VL0v_KgRzw; acceptCookies=true; next-auth.csrf-token=f0e3369e77c3cc1f591e0d86bfc9c712239b5fb8e66eaa694cc1e58651471a39%7C290985f013fcc50f194726dcac2ea9d920bfc856addde580ffd04f962f4cc243; next-auth.callback-url=http%3A%2F%2Flocalhost%3A3000",
"upgrade-insecure-requests": "1",
"sec-fetch-dest": "document",
"sec-fetch-mode": "navigate",
"sec-fetch-site": "none",
"sec-fetch-user": "?1",
"priority": "u=0, i"
}
},
{
"method": "GET",
"url": "http://localhost:8000/openapi.json",
"headers": {
"host": "localhost:8000",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:142.0) Gecko/20100101 Firefox/142.0",
"accept": "application/json,*/*",
"accept-language": "pt-BR,pt;q=0.8,en-US;q=0.5,en;q=0.3",
"accept-encoding": "gzip, deflate, br, zstd",
"referer": "http://localhost:8000/docs",
"connection": "keep-alive",
"cookie": "UserEmail=X%2FOrAPDmZqxFGRaPqXvDtH1coSgFFmzv5cQHKCoO9t3k58jBwT5xcGMds49E0wrRm4cnAG2WdBFPL8l2IfXSBD%2BEey4%2FIEDPizQdOe1l%2FLFmTCxPF08IA1mqHyVl9gmY5eqrYLUXPpYwIdHQNf7vQA%3D%3D; UserPassword=EMXw00pesMlUHxM3y2tHO7gZLfHAmo1aeBGJxhnRao0KrWwtbHjbh1ywpeWUPEa%2BDJkyHOqjjL0KVRJTpaD0lcWL%2B8II6JiHJBx%2BDCyjO8lTgr5SU7VxThgHxTUKHIto; RememberAccess=S; next-auth.session-token=eyJhbGciOiJkaXIiLCJlbmMiOiJBMjU2R0NNIn0..IHS7Sgxy9Iv8_VnG.CFBNk9kYT82XneOjTLpJEu9fP2WLr_ApxKMo4CfwDIHo7EKzvqoe5icPHm5F0nCkwUct4LPkj5a5bYfPQ6RGVCoSzynQAEE6IdtAgs2UmVLxEOCf8kDyP20LswCg32M2eWQDCeYSAq_Pq8CehZEgW0CAeupJJzyy7HtmEKTOQztb1IogJP9e0-H90Gt6yYpZnGETK2_nbVJdiX7nVReTPKj1tMexshdDbX_T3xb4BFDMXaEoRTczZiWvoGe_Din6Z9mrZpYjBnOVrOehWQ9TJJ9M8orVZEtpPfM0e9iRIfrKUG1-3RWZMVlDP3X6MrzCmpu0mN7XN_qX3Eqdk_skdknHASPDhA.ahjUD942GZ36VL0v_KgRzw; acceptCookies=true; next-auth.csrf-token=f0e3369e77c3cc1f591e0d86bfc9c712239b5fb8e66eaa694cc1e58651471a39%7C290985f013fcc50f194726dcac2ea9d920bfc856addde580ffd04f962f4cc243; next-auth.callback-url=http%3A%2F%2Flocalhost%3A3000",
"sec-fetch-dest": "empty",
"sec-fetch-mode": "cors",
"sec-fetch-site": "same-origin",
"priority": "u=4"
}
}
]