OriusTecnologia/saas_api
6
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[MVPTN-83] feat(CRUD): Criação de CRUD completo para a tabela t_censec_qualidade

Browse source
This commit is contained in:
Kenio 2025-09-19 11:17:46 -03:00
parent a4c6bf6241
commit f22eb5e6f1
23 changed files with 1366 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

380
Orius.postman_collection.json
View file

@ -3415,6 +3415,382 @@
}
]
},
{
"name": "CENSEC - Qualidade",
"item": [
{
"name": "GET",
"event": [
{
"listen": "test",
"script": {
"exec": [
"pm.test(\"Status code é 201 ou 200\", function () {\r",
" pm.expect(pm.response.code).to.be.oneOf([200, 201]);\r",
"});\r",
"\r",
"pm.test(\"Resposta contém ID ou confirmação\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData).to.have.any.keys(\"id\", \"message\", \"status\");\r",
"});\r",
"\r",
"pm.test(\"Mensagem indica sucesso\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData.message || \"\").to.match(/sucesso|criado|registrado/i);\r",
"});\r",
""
],
"type": "text/javascript",
"packages": {}
}
}
],
"request": {
"auth": {
"type": "bearer",
"bearer": [
{
"key": "token",
"value": "{{BearerToken}}",
"type": "string"
}
]
},
"method": "GET",
"header": [],
"url": {
"raw": "{{BaseUrlV1}}administrativo/t_censec_qualidade",
"host": [
"{{BaseUrlV1}}administrativo"
],
"path": [
"t_censec_qualidade"
]
}
},
"response": []
},
{
"name": "POST",
"event": [
{
"listen": "test",
"script": {
"exec": [
"const response = pm.response.json();\r",
"\r",
"// Salvando em variáveis da *collection*\r",
"pm.collectionVariables.set(\"censecQualidadeId\", response.data.censec_qualidade_id);\r",
"\r",
"pm.test(\"Status code é 201 ou 200\", function () {\r",
" pm.expect(pm.response.code).to.be.oneOf([200, 201]);\r",
"});\r",
"\r",
"pm.test(\"Resposta contém ID ou confirmação\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData).to.have.any.keys(\"id\", \"message\", \"status\");\r",
"});\r",
"\r",
"pm.test(\"Mensagem indica sucesso\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData.message || \"\").to.match(/sucesso|criado|registrado/i);\r",
"});\r",
""
],
"type": "text/javascript",
"packages": {}
}
},
{
"listen": "prerequest",
"script": {
"exec": [
""
],
"type": "text/javascript",
"packages": {}
}
}
],
"request": {
"auth": {
"type": "bearer",
"bearer": [
{
"key": "token",
"value": "{{BearerToken}}",
"type": "string"
}
]
},
"method": "POST",
"header": [],
"body": {
"mode": "raw",
"raw": "{\r\n \"descricao\": \"{{$randomFullName}}\",\r\n \"situacao\": \"{{status}}\",\r\n \"aceita_cnpj\": \"{{useEmail}}\"\r\n}",
"options": {
"raw": {
"language": "json"
}
}
},
"url": {
"raw": "{{BaseUrlV1}}administrativo/t_censec_qualidade",
"host": [
"{{BaseUrlV1}}administrativo"
],
"path": [
"t_censec_qualidade"
]
}
},
"response": []
},
{
"name": "GET DESCRIÇÃO",
"event": [
{
"listen": "test",
"script": {
"exec": [
"pm.test(\"Status code é 201 ou 200\", function () {\r",
" pm.expect(pm.response.code).to.be.oneOf([200, 201]);\r",
"});\r",
"\r",
"pm.test(\"Resposta contém ID ou confirmação\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData).to.have.any.keys(\"id\", \"message\", \"status\");\r",
"});\r",
"\r",
"pm.test(\"Mensagem indica sucesso\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData.message || \"\").to.match(/sucesso|criado|registrado/i);\r",
"});\r",
""
],
"type": "text/javascript",
"packages": {}
}
}
],
"protocolProfileBehavior": {
"disableBodyPruning": true
},
"request": {
"auth": {
"type": "bearer",
"bearer": [
{
"key": "token",
"value": "{{BearerToken}}",
"type": "string"
}
]
},
"method": "GET",
"header": [],
"body": {
"mode": "urlencoded",
"urlencoded": []
},
"url": {
"raw": "{{BaseUrlV1}}administrativo/t_censec_qualidade/descricao?descricao=RCTO - Testamentos",
"host": [
"{{BaseUrlV1}}administrativo"
],
"path": [
"t_censec_qualidade",
"descricao"
],
"query": [
{
"key": "descricao",
"value": "RCTO - Testamentos"
}
]
}
},
"response": []
},
{
"name": "GET ID",
"event": [
{
"listen": "test",
"script": {
"exec": [
"pm.test(\"Status code é 201 ou 200\", function () {\r",
" pm.expect(pm.response.code).to.be.oneOf([200, 201]);\r",
"});\r",
"\r",
"pm.test(\"Resposta contém ID ou confirmação\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData).to.have.any.keys(\"id\", \"message\", \"status\");\r",
"});\r",
"\r",
"pm.test(\"Mensagem indica sucesso\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData.message || \"\").to.match(/sucesso|criado|registrado/i);\r",
"});\r",
""
],
"type": "text/javascript",
"packages": {}
}
}
],
"request": {
"auth": {
"type": "bearer",
"bearer": [
{
"key": "token",
"value": "{{BearerToken}}",
"type": "string"
}
]
},
"method": "GET",
"header": [],
"url": {
"raw": "{{BaseUrlV1}}administrativo/t_censec_qualidade/{{censecQualidadeId}}",
"host": [
"{{BaseUrlV1}}administrativo"
],
"path": [
"t_censec_qualidade",
"{{censecQualidadeId}}"
]
}
},
"response": []
},
{
"name": "PUT",
"event": [
{
"listen": "test",
"script": {
"exec": [
"pm.test(\"Status code é 201 ou 200\", function () {\r",
" pm.expect(pm.response.code).to.be.oneOf([200, 201]);\r",
"});\r",
"\r",
"pm.test(\"Resposta contém ID ou confirmação\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData).to.have.any.keys(\"id\", \"message\", \"status\");\r",
"});\r",
"\r",
"pm.test(\"Mensagem indica sucesso\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData.message || \"\").to.match(/sucesso|criado|registrado/i);\r",
"});\r",
""
],
"type": "text/javascript",
"packages": {}
}
}
],
"request": {
"auth": {
"type": "bearer",
"bearer": [
{
"key": "token",
"value": "{{BearerToken}}",
"type": "string"
}
]
},
"method": "PUT",
"header": [],
"body": {
"mode": "raw",
"raw": "{\r\n \"descricao\": \"{{$randomFullName}}\",\r\n \"situacao\": \"{{status}}\",\r\n \"aceita_cnpj\": \"{{useEmail}}\"\r\n}",
"options": {
"raw": {
"language": "json"
}
}
},
"url": {
"raw": "{{BaseUrlV1}}administrativo/t_censec_qualidade/{{censecQualidadeId}}",
"host": [
"{{BaseUrlV1}}administrativo"
],
"path": [
"t_censec_qualidade",
"{{censecQualidadeId}}"
]
}
},
"response": []
},
{
"name": "DELETE",
"event": [
{
"listen": "test",
"script": {
"exec": [
"pm.test(\"Status code é 201 ou 200\", function () {\r",
" pm.expect(pm.response.code).to.be.oneOf([200, 201]);\r",
"});\r",
"\r",
"pm.test(\"Resposta contém ID ou confirmação\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData).to.have.any.keys(\"id\", \"message\", \"status\");\r",
"});\r",
"\r",
"pm.test(\"Mensagem indica sucesso\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData.message || \"\").to.match(/sucesso|criado|registrado/i);\r",
"});\r",
""
],
"type": "text/javascript",
"packages": {}
}
}
],
"request": {
"auth": {
"type": "bearer",
"bearer": [
{
"key": "token",
"value": "{{BearerToken}}",
"type": "string"
}
]
},
"method": "DELETE",
"header": [],
"body": {
"mode": "raw",
"raw": "",
"options": {
"raw": {
"language": "json"
}
}
},
"url": {
"raw": "{{BaseUrlV1}}administrativo/t_censec_qualidade/{{censecQualidadeId}}",
"host": [
"{{BaseUrlV1}}administrativo"
],
"path": [
"t_censec_qualidade",
"{{censecQualidadeId}}"
]
}
},
"response": []
}
]
},
{
"name": "CENSEC - Natureza Litigio",
"item": [
@ -4768,6 +5144,10 @@
},
{
"key": "censecNaturezaLitigioId",
"value": ""
},
{
"key": "censecQualidadeId",
"value": "",
"type": "default"
}

26
packages/v1/administrativo/actions/t_censec_qualidade/t_censec_qualidade_delete_action.py Normal file
View file

@ -0,0 +1,26 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import TCensecQualidadeIdSchema
from packages.v1.administrativo.repositories.t_censec_qualidade.t_censec_qualidade_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_censec_qualidade.
"""
def execute(self, censec_qualidade_schema: TCensecQualidadeIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
censec_qualidade_schema (TCensecQualidadeIdSchema): 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(censec_qualidade_schema)

29
packages/v1/administrativo/actions/t_censec_qualidade/t_censec_qualidade_get_by_descricao_action.py Normal file
View file

@ -0,0 +1,29 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import TCensecQualidadeDescricaoSchema
from packages.v1.administrativo.repositories.t_censec_qualidade.t_censec_qualidade_get_by_descricao_repository import GetByDescricaoRepository
class GetByDescricaoAction(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_censec_qualidade por descrição.
"""
def execute(self, censec_qualidade_schema: TCensecQualidadeDescricaoSchema):
"""
Executa a operação de busca no banco de dados.
Args:
censec_qualidade_schema (TCensecQualidadeDescricaoSchema): 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(censec_qualidade_schema)
# Retorno da informação
return response

24
packages/v1/administrativo/actions/t_censec_qualidade/t_censec_qualidade_index_action.py Normal file
View file

@ -0,0 +1,24 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.t_censec_qualidade.t_censec_qualidade_index_repository import IndexRepository
class IndexAction(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_censec_qualidade.
"""
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

28
packages/v1/administrativo/actions/t_censec_qualidade/t_censec_qualidade_save_action.py Normal file
View file

@ -0,0 +1,28 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import TCensecQualidadeSaveSchema
from packages.v1.administrativo.repositories.t_censec_qualidade.t_censec_qualidade_save_repository import SaveRepository
class SaveAction(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_censec_qualidade.
"""
def execute(self, censec_qualidade_schema: TCensecQualidadeSaveSchema):
"""
Executa a operação de salvamento.
Args:
censec_qualidade_schema (TCensecQualidadeSaveSchema): O esquema com os dados a serem salvos.
Returns:
O resultado da operação de salvamento.
"""
# Instanciamento do repositório
save_repository = SaveRepository()
# Execução do repositório
response = save_repository.execute(censec_qualidade_schema)
# Retorno da informação
return response

28
packages/v1/administrativo/actions/t_censec_qualidade/t_censec_qualidade_show_action.py Normal file
View file

@ -0,0 +1,28 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import TCensecQualidadeIdSchema
from packages.v1.administrativo.repositories.t_censec_qualidade.t_censec_qualidade_show_repository import ShowRepository
class ShowAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela t_censec_qualidade.
"""
def execute(self, censec_qualidade_schema: TCensecQualidadeIdSchema):
"""
Executa a operação de exibição.
Args:
censec_qualidade_schema (TCensecQualidadeIdSchema): 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(censec_qualidade_schema)
# Retorno da informação
return response

27
packages/v1/administrativo/actions/t_censec_qualidade/t_censec_qualidade_update_action.py Normal file
View file

@ -0,0 +1,27 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import TCensecQualidadeUpdateSchema
from packages.v1.administrativo.repositories.t_censec_qualidade.t_censec_qualidade_update_repository import UpdateRepository
class UpdateAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela t_censec_qualidade.
"""
def execute(self, censec_qualidade_id: int, censec_qualidade_schema: TCensecQualidadeUpdateSchema):
"""
Executa a operação de atualização.
Args:
censec_qualidade_id (int): O ID do registro a ser atualizado.
censec_qualidade_schema (TCensecQualidadeUpdateSchema): 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(censec_qualidade_id, censec_qualidade_schema)

113
packages/v1/administrativo/controllers/t_censec_qualidade_controller.py Normal file
View file

@ -0,0 +1,113 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import (
TCensecQualidadeSchema,
TCensecQualidadeSaveSchema,
TCensecQualidadeUpdateSchema,
TCensecQualidadeIdSchema,
TCensecQualidadeDescricaoSchema
)
class TCensecQualidadeController:
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_censec_qualidade")
pass
# Lista todos os registros de censec_qualidade
def index(self):
# Importação da classe desejada
indexService = self.dynamic_import.service("t_censec_qualidade_index_service", "IndexService")
# Instância da classe service
self.indexService = indexService()
# Lista todos os registros de censec_qualidade
return {
'message': 'Registros de censec_qualidade localizados com sucesso',
'data': self.indexService.execute()
}
# Busca um registro de censec_qualidade específico pelo ID
def show(self, censec_qualidade_schema: TCensecQualidadeIdSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('t_censec_qualidade_show_service', 'ShowService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de censec_qualidade desejado
return {
'message': 'Registro de censec_qualidade localizado com sucesso',
'data': self.show_service.execute(censec_qualidade_schema)
}
# Busca um registro de censec_qualidade pela descrição
def get_by_descricao(self, censec_qualidade_schema: TCensecQualidadeDescricaoSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('t_censec_qualidade_get_descricao_service', 'GetByDescricaoService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de censec_qualidade desejado
return {
'message': 'Registro de censec_qualidade localizado com sucesso',
'data': self.show_service.execute(censec_qualidade_schema, True)
}
# Cadastra um novo registro de censec_qualidade
def save(self, censec_qualidade_schema: TCensecQualidadeSaveSchema):
#Importação da classe desejada
save_service = self.dynamic_import.service('t_censec_qualidade_save_service', 'SaveService')
# Instância da classe desejada
self.save_service = save_service()
# Busca e retorna o registro de censec_qualidade desejado
return {
'message': 'Registro de censec_qualidade salvo com sucesso',
'data': self.save_service.execute(censec_qualidade_schema)
}
# Atualiza os dados de um registro de censec_qualidade
def update(self, censec_qualidade_id: int, censec_qualidade_schema: TCensecQualidadeUpdateSchema):
#Importação da classe desejada
update_service = self.dynamic_import.service('t_censec_qualidade_update_service', 'UpdateService')
# Instância da classe desejada
self.update_service = update_service()
# Busca e retorna o registro de censec_qualidade desejado
return {
'message': 'Registro de censec_qualidade atualizado com sucesso',
'data': self.update_service.execute(censec_qualidade_id, censec_qualidade_schema)
}
# Exclui um registro de censec_qualidade
def delete(self, censec_qualidade_schema: TCensecQualidadeIdSchema):
#Importação da classe desejada
delete_service = self.dynamic_import.service('t_censec_qualidade_delete_service', 'DeleteService')
# Instância da classe desejada
self.delete_service = delete_service()
# Busca e retorna o registro de censec_qualidade desejado
return {
'message': 'Registro de censec_qualidade removido com sucesso',
'data': self.delete_service.execute(censec_qualidade_schema)
}

109
packages/v1/administrativo/endpoints/t_censec_qualidade_endpoint.py Normal file
View file

@ -0,0 +1,109 @@
# 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_censec_qualidade_controller import TCensecQualidadeController
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import (
TCensecQualidadeSchema,
TCensecQualidadeSaveSchema,
TCensecQualidadeUpdateSchema,
TCensecQualidadeIdSchema,
TCensecQualidadeDescricaoSchema
)
# Inicializa o roteador para as rotas do tipo de reconhecimento
router = APIRouter()
# Instanciamento do controller desejado
t_censec_qualidade_controller = TCensecQualidadeController()
# Lista todos os registros de t_censec_qualidade
@router.get('/',
status_code=status.HTTP_200_OK,
summary='Lista todos os registros de t_censec_qualidade cadastrados',
response_description='Lista todos os registros de t_censec_qualidade cadastrados')
async def index(current_user: dict = Depends(get_current_user)):
# Busca todos os registros de t_censec_qualidade cadastrados
response = t_censec_qualidade_controller.index()
# Retorna os dados localizados
return response
# Localiza um registro de t_censec_qualidade pela descrição
@router.get('/descricao',
status_code=status.HTTP_200_OK,
summary='Busca um registro de t_censec_qualidade em específico pela descrição',
response_description='Busca um registro de t_censec_qualidade em específico')
async def get_by_descricao(descricao : str, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
censec_qualidade_schema = TCensecQualidadeSchema(descricao=descricao)
# Busca um registro de t_censec_qualidade específico pela descrição
response = t_censec_qualidade_controller.get_by_descricao(censec_qualidade_schema)
# Retorna os dados localizados
return response
# Localiza um registro de t_censec_qualidade pelo ID
@router.get('/{censec_qualidade_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de t_censec_qualidade em específico pelo ID',
response_description='Busca um registro de t_censec_qualidade em específico')
async def show(censec_qualidade_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
censec_qualidade_schema = TCensecQualidadeIdSchema(censec_qualidade_id=censec_qualidade_id)
# Busca um registro de t_censec_qualidade específico pelo ID
response = t_censec_qualidade_controller.show(censec_qualidade_schema)
# Retorna os dados localizados
return response
# Cadastro de registro de t_censec_qualidade
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um registro de t_censec_qualidade',
response_description='Cadastra um registro de t_censec_qualidade')
async def save(censec_qualidade_schema: TCensecQualidadeSaveSchema, current_user: dict = Depends(get_current_user)):
# Efetua o cadastro no banco de dados
response = t_censec_qualidade_controller.save(censec_qualidade_schema)
# Retorna os dados localizados
return response
# Atualiza os dados de um registro de t_censec_qualidade
@router.put('/{censec_qualidade_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza um registro de t_censec_qualidade',
response_description='Atualiza um registro de t_censec_qualidade')
async def update(censec_qualidade_id: int, censec_qualidade_schema: TCensecQualidadeUpdateSchema, current_user: dict = Depends(get_current_user)):
# Efetua a atualização dos dados
response = t_censec_qualidade_controller.update(censec_qualidade_id, censec_qualidade_schema)
# Retorna os dados localizados
return response
# Exclui um determinado registro de t_censec_qualidade
@router.delete('/{censec_qualidade_id}',
status_code=status.HTTP_200_OK,
summary='Remove um registro de t_censec_qualidade',
response_description='Remove um registro de t_censec_qualidade')
async def delete(censec_qualidade_id: int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
censec_qualidade_schema = TCensecQualidadeIdSchema(censec_qualidade_id=censec_qualidade_id)
# Efetua a exclusão do registro de t_censec_qualidade
response = t_censec_qualidade_controller.delete(censec_qualidade_schema)
# Retorna os dados localizados
return response

41
packages/v1/administrativo/repositories/t_censec_qualidade/t_censec_qualidade_delete_repository.py Normal file
View file

@ -0,0 +1,41 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import TCensecQualidadeIdSchema
from fastapi import HTTPException, status
class DeleteRepository(BaseRepository):
"""
Repositório para a operação de exclusão de um registro na tabela
t_censec_qualidade.
"""
def execute(self, censec_qualidade_schema: TCensecQualidadeIdSchema):
"""
Executa a consulta SQL para remover um registro pelo ID.
Args:
censec_qualidade_schema (TCensecQualidadeIdSchema): 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_CENSEC_QUALIDADE WHERE CENSEC_QUALIDADE_ID = :censec_qualidade_id """
# Preenchimento de parâmetros
params = {
"censec_qualidade_id": censec_qualidade_schema.censec_qualidade_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 CENSEC_QUALIDADE: {e}"
)

29
packages/v1/administrativo/repositories/t_censec_qualidade/t_censec_qualidade_get_by_descricao_repository.py Normal file
View file

@ -0,0 +1,29 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import TCensecQualidadeDescricaoSchema
class GetByDescricaoRepository(BaseRepository):
"""
Repositório para a operação de busca de um registro na tabela
t_censec_qualidade por descrição.
"""
def execute(self, censec_qualidade_schema: TCensecQualidadeDescricaoSchema):
"""
Executa a consulta SQL para buscar um registro pela descrição.
Args:
censec_qualidade_schema (TCensecQualidadeDescricaoSchema): 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_CENSEC_QUALIDADE WHERE DESCRICAO = :descricao """
# Preenchimento de parâmetros
params = {
'descricao': censec_qualidade_schema.descricao
}
# Execução do sql
return self.fetch_one(sql, params)

23
packages/v1/administrativo/repositories/t_censec_qualidade/t_censec_qualidade_index_repository.py Normal file
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_censec_qualidade.
"""
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_CENSEC_QUALIDADE """
# Execução do sql
response = self.fetch_all(sql)
# Retorna os dados localizados
return response

56
packages/v1/administrativo/repositories/t_censec_qualidade/t_censec_qualidade_save_repository.py Normal file
View file

@ -0,0 +1,56 @@
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import TCensecQualidadeSaveSchema
class SaveRepository(BaseRepository):
"""
Repositório para a operação de salvamento de um novo registro na tabela t_censec_qualidade.
"""
def execute(self, censec_qualidade_schema: TCensecQualidadeSaveSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
censec_qualidade_schema (TCensecQualidadeSaveSchema): 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_CENSEC_QUALIDADE(
CENSEC_QUALIDADE_ID,
DESCRICAO,
SITUACAO,
ACEITA_CNPJ
) VALUES (
:censec_qualidade_id,
:descricao,
:situacao,
:aceita_cnpj
) RETURNING *;"""
# Preenchimento de parâmetros
params = {
'censec_qualidade_id': censec_qualidade_schema.censec_qualidade_id,
'descricao': censec_qualidade_schema.descricao,
'situacao': censec_qualidade_schema.situacao,
'aceita_cnpj': censec_qualidade_schema.aceita_cnpj
}
# 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 CENSEC_QUALIDADE: {e}"
)

46
packages/v1/administrativo/repositories/t_censec_qualidade/t_censec_qualidade_show_repository.py Normal file
View file

@ -0,0 +1,46 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import TCensecQualidadeIdSchema
from fastapi import HTTPException, status
class ShowRepository(BaseRepository):
"""
Repositório para a operação de exibição de um registro na tabela t_censec_qualidade.
"""
def execute(self, censec_qualidade_schema: TCensecQualidadeIdSchema):
"""
Busca um registro específico de CENSEC_QUALIDADE pelo ID.
Args:
censec_qualidade_schema (TCensecQualidadeIdSchema): 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_CENSEC_QUALIDADE WHERE CENSEC_QUALIDADE_ID = :censec_qualidade_id"
# Preenchimento de parâmetros
params = {
'censec_qualidade_id': censec_qualidade_schema.censec_qualidade_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)}"
)

65
packages/v1/administrativo/repositories/t_censec_qualidade/t_censec_qualidade_update_repository.py Normal file
View file

@ -0,0 +1,65 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import TCensecQualidadeUpdateSchema
from fastapi import HTTPException, status
class UpdateRepository(BaseRepository):
"""
Repositório para a operação de atualização na tabela T_CENSEC_QUALIDADE.
"""
def execute(self, censec_qualidade_id : int, censec_qualidade_schema: TCensecQualidadeUpdateSchema):
"""
Executa a atualização de um registro na tabela.
Args:
censec_qualidade_id (int): O ID do registro a ser atualizado.
censec_qualidade_schema (TCensecQualidadeUpdateSchema): 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 censec_qualidade_schema.descricao is not None:
updates.append("DESCRICAO = :descricao")
params["descricao"] = censec_qualidade_schema.descricao
if censec_qualidade_schema.situacao is not None:
updates.append("SITUACAO = :situacao")
params["situacao"] = censec_qualidade_schema.situacao
if censec_qualidade_schema.aceita_cnpj is not None:
updates.append("ACEITA_CNPJ = :aceita_cnpj")
params["aceita_cnpj"] = censec_qualidade_schema.aceita_cnpj
if not updates:
return False
params["censec_qualidade_id"] = censec_qualidade_id
sql = f"UPDATE T_CENSEC_QUALIDADE SET {', '.join(updates)} WHERE censec_qualidade_id = :censec_qualidade_id RETURNING *;"
# Executa a query
result = self.run_and_return(sql, params)
if not result.censec_qualidade_id:
# Informa que não existe o registro a ser modificado
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail='Nenhum CENSEC_QUALIDADE 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 CENSEC_QUALIDADE: {e}"
)

110
packages/v1/administrativo/schemas/t_censec_qualidade_schema.py Normal file
View file

@ -0,0 +1,110 @@
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.)
from actions.validations.text import Text
# ----------------------------------------------------
# Schema base
# ----------------------------------------------------
class TCensecQualidadeSchema(BaseModel):
censec_qualidade_id: Optional[int] = None
descricao: Optional[str] = None
situacao: Optional[str] = None
aceita_cnpj: Optional[str] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar um CENSEC_QUALIDADE especifico pelo ID (GET)
# ----------------------------------------------------
class TCensecQualidadeIdSchema(BaseModel):
censec_qualidade_id: int
# ----------------------------------------------------
# Schema para localizar um CENSEC_QUALIDADE especifico pela descrição (GET)
# ----------------------------------------------------
class TCensecQualidadeDescricaoSchema(BaseModel):
descricao: str
# ----------------------------------------------------
# Schema para criação de novo CENSEC_QUALIDADE (POST)
# ----------------------------------------------------
class TCensecQualidadeSaveSchema(BaseModel):
censec_qualidade_id: Optional[int] = None
descricao: str
situacao: str
aceita_cnpj: str
# Sanitiza os inputs enviados
@field_validator('descricao', 'situacao', 'aceita_cnpj')
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 not self.aceita_cnpj or len(self.aceita_cnpj.strip()) == 0:
errors.append({'input': 'aceita_cnpj', 'message': 'O campo aceita_cnpj é obrigatório.'})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self
# ----------------------------------------------------
# Schema para atualizar CENSEC_QUALIDADE (PUT)
# ----------------------------------------------------
class TCensecQualidadeUpdateSchema(BaseModel):
descricao: Optional[str] = None
situacao: Optional[str] = None
aceita_cnpj: Optional[str] = None
# Sanitiza os inputs enviados
@field_validator('descricao', 'situacao', 'aceita_cnpj')
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 self.descricao is not None and len(self.descricao.strip()) == 0:
errors.append({'input': 'descricao', 'message': 'A descrição é obrigatória.'})
if self.situacao is not None and len(self.situacao.strip()) == 0:
errors.append({'input': 'situacao', 'message': 'A situação é obrigatória.'})
if self.aceita_cnpj is not None and len(self.aceita_cnpj.strip()) == 0:
errors.append({'input': 'aceita_cnpj', 'message': 'O campo aceita_cnpj é obrigatório.'})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self

27
packages/v1/administrativo/services/t_censec_qualidade/go/t_censec_qualidade_delete_service.py Normal file
View file

@ -0,0 +1,27 @@
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import TCensecQualidadeIdSchema
from packages.v1.administrativo.actions.t_censec_qualidade.t_censec_qualidade_delete_action import DeleteAction
class DeleteService:
"""
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_censec_qualidade.
"""
def execute(self, censec_qualidade_schema: TCensecQualidadeIdSchema):
"""
Executa a operação de exclusão do registro no banco de dados.
Args:
censec_qualidade_schema (TCensecQualidadeIdSchema): 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(censec_qualidade_schema)
# Retorno da informação
return data

38
packages/v1/administrativo/services/t_censec_qualidade/go/t_censec_qualidade_get_descricao_service.py Normal file
View file

@ -0,0 +1,38 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import TCensecQualidadeDescricaoSchema
from packages.v1.administrativo.actions.t_censec_qualidade.t_censec_qualidade_get_by_descricao_action import GetByDescricaoAction
class GetByDescricaoService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela t_censec_qualidade pela sua descrição.
"""
def execute(self, censec_qualidade_schema: TCensecQualidadeDescricaoSchema, messageValidate: bool):
"""
Executa a operação de busca no banco de dados.
Args:
censec_qualidade_schema (TCensecQualidadeDescricaoSchema): 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(censec_qualidade_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 CENSEC_QUALIDADE'
)
# Retorno da informação
return data

32
packages/v1/administrativo/services/t_censec_qualidade/go/t_censec_qualidade_index_service.py Normal file
View file

@ -0,0 +1,32 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.actions.t_censec_qualidade.t_censec_qualidade_index_action import IndexAction
class IndexService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de registros na tabela t_censec_qualidade.
"""
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 CENSEC_QUALIDADE'
)
# Retorna as informações localizadas
return data

70
packages/v1/administrativo/services/t_censec_qualidade/go/t_censec_qualidade_save_service.py Normal file
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_censec_qualidade_schema import TCensecQualidadeSaveSchema, TCensecQualidadeDescricaoSchema
from packages.v1.administrativo.actions.t_censec_qualidade.t_censec_qualidade_save_action import SaveAction
from fastapi import HTTPException, status
class SaveService:
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_censec_qualidade")
pass
# Cadastra o novo CENSEC_QUALIDADE
def execute(self, censec_qualidade_schema: TCensecQualidadeSaveSchema):
# 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_censec_qualidade_get_descricao_service", "GetByDescricaoService")
# Instanciamento da service
self.descricao_service = descricao_service()
# Verifica se a descrição já está sendo utilizada
self.response = self.descricao_service.execute(TCensecQualidadeDescricaoSchema(descricao=censec_qualidade_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 censec_qualidade_schema.censec_qualidade_id:
# Crio um objeto de sequencia
sequencia_schema = GSequenciaSchema()
# Define os dados para atualizar a sequencia
sequencia_schema.tabela = 'T_CENSEC_QUALIDADE'
# Busco a sequência atualizada
generate = GenerateService()
# Busco a sequência atualizada
sequencia = generate.execute(sequencia_schema)
# Atualiza os dados da chave primária
censec_qualidade_schema.censec_qualidade_id = sequencia.sequencia
# Instanciamento de ações
save_action = SaveAction()
# Retorna o resultado da operação
return save_action.execute(censec_qualidade_schema)

35
packages/v1/administrativo/services/t_censec_qualidade/go/t_censec_qualidade_show_service.py Normal file
View file

@ -0,0 +1,35 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import TCensecQualidadeIdSchema
from packages.v1.administrativo.actions.t_censec_qualidade.t_censec_qualidade_show_action import ShowAction
class ShowService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela t_censec_qualidade.
"""
def execute(self, censec_qualidade_schema: TCensecQualidadeIdSchema):
"""
Executa a operação de busca no banco de dados.
Args:
censec_qualidade_schema (TCensecQualidadeIdSchema): 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(censec_qualidade_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 CENSEC_QUALIDADE'
)
# Retorno da informação
return data

23
packages/v1/administrativo/services/t_censec_qualidade/go/t_censec_qualidade_update_service.py Normal file
View file

@ -0,0 +1,23 @@
from packages.v1.administrativo.schemas.t_censec_qualidade_schema import TCensecQualidadeUpdateSchema
from packages.v1.administrativo.actions.t_censec_qualidade.t_censec_qualidade_update_action import UpdateAction
class UpdateService:
"""
Serviço para a operação de atualização de um registro na tabela
t_censec_qualidade.
"""
def execute(self, censec_qualidade_id: int, censec_qualidade_schema: TCensecQualidadeUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
censec_qualidade_schema (TCensecQualidadeUpdateSchema): 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(censec_qualidade_id, censec_qualidade_schema)

7
packages/v1/api.py
View file

@ -13,6 +13,7 @@ from packages.v1.administrativo.endpoints import g_tb_regimecomunhao_endpoint
from packages.v1.administrativo.endpoints import g_tb_regimebens_endpoint
from packages.v1.administrativo.endpoints import t_censec_endpoint
from packages.v1.administrativo.endpoints import t_censec_naturezalitigio_endpoint
from packages.v1.administrativo.endpoints import t_censec_qualidade_endpoint
# Cria uma instância do APIRouter que vai agregar todas as rotas da API
api_router = APIRouter()
@ -73,3 +74,9 @@ api_router.include_router(
api_router.include_router(
t_censec_naturezalitigio_endpoint.router, prefix="/administrativo/t_censec_naturezalitigio", tags=["CENSEC - Natureza Litígio"]
)
# Inclui as rotas de t_censec_qualidade
api_router.include_router(
t_censec_qualidade_endpoint.router, prefix="/administrativo/t_censec_qualidade", tags=["CENSEC - Qualidade"]
)