feat(CRUD): Adiciona os endpoints para vinculadar qualidades ao ato

This commit is contained in:
Keven Willian Pereira de Souza 2025-10-13 15:24:01 -03:00
parent 88fd24c6b8
commit 642f2ed826
37 changed files with 1878 additions and 0 deletions

View file

@ -0,0 +1,36 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_censec_qualidadeato.t_censec_qualidadeato_delete_repository import TCensecQualidadeAtoDeleteRepository
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoIdSchema
class TCensecQualidadeAtoDeleteAction(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_QUALIDADEATO.
"""
def execute(self, t_censec_qualidadeato_id_schema: TCensecQualidadeAtoIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
t_censec_qualidadeato_id_schema (TCensecQualidadeAtoIdSchema):
O esquema contendo o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
t_censec_qualidadeato_delete_repository = TCensecQualidadeAtoDeleteRepository()
# ----------------------------------------------------
# Execução da exclusão
# ----------------------------------------------------
response = t_censec_qualidadeato_delete_repository.execute(t_censec_qualidadeato_id_schema)
# ----------------------------------------------------
# Retorno do resultado
# ----------------------------------------------------
return response

View file

@ -0,0 +1,32 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_censec_qualidadeato.t_censec_qualidadeato_index_repository import TCensecQualidadeAtoIndexRepository
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoIndexSchema
class TCensecQualidadeAtoIndexAction(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_QUALIDADEATO.
"""
def execute(self, censec_qualidade_ato_index_schema: TCensecQualidadeAtoIndexSchema):
"""
Executa a operação de listagem no banco de dados.
Returns:
A lista de registros encontrados.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
t_censec_qualidadeato_index_repository = TCensecQualidadeAtoIndexRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = t_censec_qualidadeato_index_repository.execute(censec_qualidade_ato_index_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -0,0 +1,36 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_censec_qualidadeato.t_censec_qualidadeato_save_repository import TCensecQualidadeAtoSaveRepository
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoSaveSchema
class TCensecQualidadeAtoSaveAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvamento de um novo registro na tabela T_CENSEC_QUALIDADEATO.
"""
def execute(self, t_censec_qualidadeato_save_schema: TCensecQualidadeAtoSaveSchema):
"""
Executa a operação de salvamento.
Args:
t_censec_qualidadeato_save_schema (TCensecQualidadeAtoSaveSchema):
O esquema com os dados a serem persistidos.
Returns:
O resultado da operação de salvamento.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
t_censec_qualidadeato_save_repository = TCensecQualidadeAtoSaveRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = t_censec_qualidadeato_save_repository.execute(t_censec_qualidadeato_save_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -0,0 +1,36 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_censec_qualidadeato.t_censec_qualidadeato_show_repository import TCensecQualidadeAtoShowRepository
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoIdSchema
class TCensecQualidadeAtoShowAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela T_CENSEC_QUALIDADEATO.
"""
def execute(self, t_censec_qualidadeato_id_schema: TCensecQualidadeAtoIdSchema):
"""
Executa a operação de exibição.
Args:
t_censec_qualidadeato_id_schema (TCensecQualidadeAtoIdSchema):
O esquema com o ID do registro a ser exibido.
Returns:
O resultado da operação de exibição.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
t_censec_qualidadeato_show_repository = TCensecQualidadeAtoShowRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = t_censec_qualidadeato_show_repository.execute(t_censec_qualidadeato_id_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -0,0 +1,36 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_censec_qualidadeato.t_censec_qualidadeato_update_repository import TCensecQualidadeAtoUpdateRepository
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoUpdateSchema
class TCensecQualidadeAtoUpdateAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela T_CENSEC_QUALIDADEATO.
"""
def execute(self, t_censec_qualidadeato_update_schema: TCensecQualidadeAtoUpdateSchema):
"""
Executa a operação de atualização.
Args:
t_censec_qualidadeato_update_schema (TCensecQualidadeAtoUpdateSchema):
O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# ----------------------------------------------------
# Instanciamento do repositório de atualização
# ----------------------------------------------------
t_censec_qualidadeato_update_repository = TCensecQualidadeAtoUpdateRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = t_censec_qualidadeato_update_repository.execute(t_censec_qualidadeato_update_schema)
# ----------------------------------------------------
# Retorno do resultado
# ----------------------------------------------------
return response

View file

@ -0,0 +1,33 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_censec_tipoato.t_censec_tipoato_delete_repository import TCensecTipoAtoDeleteRepository
from api.packages.v1.administrativo.schemas.t_censec_tipoato_schema import TCensecTipoAtoIdSchema
class TCensecTipoAtoDeleteAction(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_TIPOATO.
"""
def execute(self, t_censec_tipoato_id_schema: TCensecTipoAtoIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
t_censec_tipoato_id_schema (TCensecTipoAtoIdSchema):
O esquema contendo o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
t_censec_tipoato_delete_repository = TCensecTipoAtoDeleteRepository()
# ----------------------------------------------------
# Execução da exclusão
# ----------------------------------------------------
response = t_censec_tipoato_delete_repository.execute(t_censec_tipoato_id_schema)
return response

View file

@ -0,0 +1,35 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_censec_tipoato.t_censec_tipoato_index_repository import TCensecTipoAtoIndexRepository
class TCensecTipoAtoIndexAction(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_TIPOATO.
"""
def execute(self):
"""
Executa a operação de listagem no banco de dados.
Args:
t_censec_tipoato_index_schema (TCensecTipoAtoIndexSchema):
Esquema contendo parâmetros opcionais de filtro.
Returns:
A lista de registros encontrados.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
t_censec_tipoato_index_repository = TCensecTipoAtoIndexRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = t_censec_tipoato_index_repository.execute()
# ----------------------------------------------------
# Retorno da informação
# --------------------------------
return response

View file

@ -0,0 +1,36 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_censec_tipoato.t_censec_tipoato_save_repository import TCensecTipoAtoSaveRepository
from api.packages.v1.administrativo.schemas.t_censec_tipoato_schema import TCensecTipoAtoSaveSchema
class TCensecTipoAtoSaveAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvamento de um novo registro na tabela T_CENSEC_TIPOATO.
"""
def execute(self, t_censec_tipoato_save_schema: TCensecTipoAtoSaveSchema):
"""
Executa a operação de salvamento.
Args:
t_censec_tipoato_schema (TCensecTipoAtoSchema):
O esquema com os dados a serem persistidos.
Returns:
O resultado da operação de salvamento.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
t_censec_tipoato_save_repository = TCensecTipoAtoSaveRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = t_censec_tipoato_save_repository.execute(t_censec_tipoato_save_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -0,0 +1,36 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_censec_tipoato.t_censec_tipoato_show_repository import TCensecTipoAtoShowRepository
from api.packages.v1.administrativo.schemas.t_censec_tipoato_schema import TCensecTipoAtoIdSchema
class TCensecTipoAtoShowAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela T_CENSEC_TIPOATO.
"""
def execute(self, t_censec_tipoato_id_schema: TCensecTipoAtoIdSchema):
"""
Executa a operação de exibição.
Args:
t_censec_tipoato_id_schema (TCensecTipoAtoIdSchema):
O esquema com o ID do registro a ser exibido.
Returns:
O resultado da operação de exibição.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
t_censec_tipoato_show_repository = TCensecTipoAtoShowRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = t_censec_tipoato_show_repository.execute(t_censec_tipoato_id_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return response

View file

@ -0,0 +1,36 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_censec_tipoato.t_censec_tipoato_update_repository import TCensecTipoAtoUpdateRepository
from api.packages.v1.administrativo.schemas.t_censec_tipoato_schema import TCensecTipoAtoUpdateSchema
class TCensecTipoAtoUpdateAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela T_CENSEC_TIPOATO.
"""
def execute(self, t_censec_tipoato_update_schema: TCensecTipoAtoUpdateSchema):
"""
Executa a operação de atualização.
Args:
t_censec_tipoato_update_schema (TCensecTipoAtoUpdateSchema):
O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# ----------------------------------------------------
# Instanciamento do repositório de atualização
# ----------------------------------------------------
t_censec_tipoato_update_repository = TCensecTipoAtoUpdateRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = t_censec_tipoato_update_repository.execute(t_censec_tipoato_update_schema)
# ----------------------------------------------------
# Retorno do resultado
# ----------------------------------------------------
return response

View file

@ -0,0 +1,102 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import (
TCensecQualidadeAtoIndexSchema,
TCensecQualidadeAtoSaveSchema,
TCensecQualidadeAtoUpdateSchema,
TCensecQualidadeAtoIdSchema
)
class TCensecQualidadeAtoController:
"""
Controller responsável por orquestrar as operações CRUD da tabela T_CENSEC_QUALIDADEATO,
utilizando carregamento dinâmico de serviços via DynamicImport.
"""
def __init__(self):
# ----------------------------------------------------
# Inicialização do DynamicImport
# ----------------------------------------------------
self.dynamic_import = DynamicImport()
self.dynamic_import.set_package("administrativo")
self.dynamic_import.set_table("t_censec_qualidadeato")
# ----------------------------------------------------
# Lista todos os registros de T_CENSEC_QUALIDADEATO
# ----------------------------------------------------
def index(self, censec_qualidade_ato_index_schema: TCensecQualidadeAtoIndexSchema):
# Importação da classe desejada
index_service = self.dynamic_import.service("t_censec_qualidadeato_index_service", "TCensecQualidadeAtoIndexService")
# Instância da classe service
self.index_service = index_service()
# Execução da listagem
return {
"message": "Registros de T_CENSEC_QUALIDADEATO localizados com sucesso.",
"data": self.index_service.execute(censec_qualidade_ato_index_schema),
}
# ----------------------------------------------------
# Busca um registro específico de T_CENSEC_QUALIDADEATO pelo ID
# ----------------------------------------------------
def show(self, t_censec_qualidadeato_id_schema: TCensecQualidadeAtoIdSchema):
# Importação da classe desejada
show_service = self.dynamic_import.service("t_censec_qualidadeato_show_service", "TCensecQualidadeAtoShowService")
# Instância da classe service
self.show_service = show_service()
# Execução da busca
return {
"message": "Registro de T_CENSEC_QUALIDADEATO localizado com sucesso.",
"data": self.show_service.execute(t_censec_qualidadeato_id_schema),
}
# ----------------------------------------------------
# Cadastra um novo registro em T_CENSEC_QUALIDADEATO
# ----------------------------------------------------
def save(self, t_censec_qualidadeato_save_schema: TCensecQualidadeAtoSaveSchema):
# Importação da classe desejada
save_service = self.dynamic_import.service("t_censec_qualidadeato_save_service", "TCensecQualidadeAtoSaveService")
# Instância da classe service
self.save_service = save_service()
# Execução do salvamento
return {
"message": "Registro de T_CENSEC_QUALIDADEATO salvo com sucesso.",
"data": self.save_service.execute(t_censec_qualidadeato_save_schema),
}
# ----------------------------------------------------
# Atualiza um registro existente de T_CENSEC_QUALIDADEATO
# ----------------------------------------------------
def update(self, t_censec_qualidadeato_update_schema: TCensecQualidadeAtoUpdateSchema):
# Importação da classe desejada
update_service = self.dynamic_import.service("t_censec_qualidadeato_update_service", "TCensecQualidadeAtoUpdateService")
# Instância da classe service
self.update_service = update_service()
# Execução da atualização
return {
"message": "Registro de T_CENSEC_QUALIDADEATO atualizado com sucesso.",
"data": self.update_service.execute(t_censec_qualidadeato_update_schema),
}
# ----------------------------------------------------
# Exclui um registro de T_CENSEC_QUALIDADEATO
# ----------------------------------------------------
def delete(self, t_censec_qualidadeato_id_schema: TCensecQualidadeAtoIdSchema):
# Importação da classe desejada
delete_service = self.dynamic_import.service("t_censec_qualidadeato_delete_service", "TCensecQualidadeAtoDeleteService")
# Instância da classe service
self.delete_service = delete_service()
# Execução da exclusão
return {
"message": "Registro de T_CENSEC_QUALIDADEATO removido com sucesso.",
"data": self.delete_service.execute(t_censec_qualidadeato_id_schema),
}

View file

@ -0,0 +1,101 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.t_censec_tipoato_schema import (
TCensecTipoAtoSaveSchema,
TCensecTipoAtoUpdateSchema,
TCensecTipoAtoIdSchema
)
class TCensecTipoAtoController:
"""
Controller responsável por orquestrar as operações CRUD da tabela T_CENSEC_TIPOATO,
utilizando carregamento dinâmico de serviços via DynamicImport.
"""
def __init__(self):
# ----------------------------------------------------
# Inicialização do DynamicImport
# ----------------------------------------------------
self.dynamic_import = DynamicImport()
self.dynamic_import.set_package("administrativo")
self.dynamic_import.set_table("t_censec_tipoato")
# ----------------------------------------------------
# Lista todos os registros de T_CENSEC_TIPOATO
# ----------------------------------------------------
def index(self):
# Importação da classe desejada
index_service = self.dynamic_import.service("t_censec_tipoato_index_service", "TCensecTipoAtoIndexService")
# Instância da classe service
self.index_service = index_service()
# Execução da listagem
return {
"message": "Registros de T_CENSEC_TIPOATO localizados com sucesso.",
"data": self.index_service.execute(),
}
# ----------------------------------------------------
# Busca um registro específico de T_CENSEC_TIPOATO pelo ID
# ----------------------------------------------------
def show(self, t_censec_tipoato_id_schema: TCensecTipoAtoIdSchema):
# Importação da classe desejada
show_service = self.dynamic_import.service("t_censec_tipoato_show_service", "TCensecTipoAtoShowService")
# Instância da classe service
self.show_service = show_service()
# Execução da busca
return {
"message": "Registro de T_CENSEC_TIPOATO localizado com sucesso.",
"data": self.show_service.execute(t_censec_tipoato_id_schema),
}
# ----------------------------------------------------
# Cadastra um novo registro em T_CENSEC_TIPOATO
# ----------------------------------------------------
def save(self, t_censec_tipoato_save_schema: TCensecTipoAtoSaveSchema):
# Importação da classe desejada
save_service = self.dynamic_import.service("t_censec_tipoato_save_service", "TCensecTipoAtoSaveService")
# Instância da classe service
self.save_service = save_service()
# Execução do salvamento
return {
"message": "Registro de T_CENSEC_TIPOATO salvo com sucesso.",
"data": self.save_service.execute(t_censec_tipoato_save_schema),
}
# ----------------------------------------------------
# Atualiza um registro existente de T_CENSEC_TIPOATO
# ----------------------------------------------------
def update(self, t_censec_tipoato_update_schema: TCensecTipoAtoUpdateSchema):
# Importação da classe desejada
update_service = self.dynamic_import.service("t_censec_tipoato_update_service", "TCensecTipoAtoUpdateService")
# Instância da classe service
self.update_service = update_service()
# Execução da atualização
return {
"message": "Registro de T_CENSEC_TIPOATO atualizado com sucesso.",
"data": self.update_service.execute(t_censec_tipoato_update_schema),
}
# ----------------------------------------------------
# Exclui um registro de T_CENSEC_TIPOATO
# ----------------------------------------------------
def delete(self, t_censec_tipoato_id_schema: TCensecTipoAtoIdSchema):
# Importação da classe desejada
delete_service = self.dynamic_import.service("t_censec_tipoato_delete_service", "TCensecTipoAtoDeleteService")
# Instância da classe service
self.delete_service = delete_service()
# Execução da exclusão
return {
"message": "Registro de T_CENSEC_TIPOATO removido com sucesso.",
"data": self.delete_service.execute(t_censec_tipoato_id_schema),
}

View file

@ -0,0 +1,98 @@
# Importação de bibliotecas
from fastapi import APIRouter, Depends, status
from actions.jwt.get_current_user import get_current_user
from packages.v1.administrativo.controllers.t_censec_qualidadeato_controller import TCensecQualidadeAtoController
from packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import (
TCensecQualidadeAtoIndexSchema,
TCensecQualidadeAtoSaveSchema,
TCensecQualidadeAtoUpdateSchema,
TCensecQualidadeAtoIdSchema
)
# ----------------------------------------------------
# Inicializa o roteador para as rotas da tabela T_CENSEC_QUALIDADEATO
# ----------------------------------------------------
router = APIRouter()
# Instanciamento do controller
t_censec_qualidadeato_controller = TCensecQualidadeAtoController()
# ----------------------------------------------------
# Lista todos os registros de T_CENSEC_QUALIDADEATO
# ----------------------------------------------------
@router.get('/tipo-ato/{censec_tipo_ato_id}',
status_code=status.HTTP_200_OK,
summary='Lista todos os registros de T_CENSEC_QUALIDADEATO cadastrados',
response_description='Lista todos os registros de T_CENSEC_QUALIDADEATO cadastrados')
async def index(censec_tipo_ato_id: int, current_user: dict = Depends(get_current_user)):
"""
Retorna todos os registros da tabela T_CENSEC_QUALIDADEATO.
"""
t_censec_qualidadeato_index_schema = TCensecQualidadeAtoIndexSchema(censec_tipoato_id=censec_tipo_ato_id)
response = t_censec_qualidadeato_controller.index(t_censec_qualidadeato_index_schema)
return response
# ----------------------------------------------------
# Busca um registro específico de T_CENSEC_QUALIDADEATO pelo ID
# ----------------------------------------------------
@router.get('/{censec_qualidadeato_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de T_CENSEC_QUALIDADEATO pelo ID',
response_description='Busca um registro de T_CENSEC_QUALIDADEATO em específico')
async def show(censec_qualidadeato_id: int, current_user: dict = Depends(get_current_user)):
"""
Retorna um registro específico de T_CENSEC_QUALIDADEATO com base no ID informado.
"""
t_censec_qualidadeato_id_schema = TCensecQualidadeAtoIdSchema(censec_qualidadeato_id=censec_qualidadeato_id)
response = t_censec_qualidadeato_controller.show(t_censec_qualidadeato_id_schema)
return response
# ----------------------------------------------------
# Cadastra um novo registro em T_CENSEC_QUALIDADEATO
# ----------------------------------------------------
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um novo registro em T_CENSEC_QUALIDADEATO',
response_description='Cadastra um novo registro em T_CENSEC_QUALIDADEATO')
async def save(t_censec_qualidadeato_schema: TCensecQualidadeAtoSaveSchema, current_user: dict = Depends(get_current_user)):
"""
Cria um novo registro na tabela T_CENSEC_QUALIDADEATO.
"""
response = t_censec_qualidadeato_controller.save(t_censec_qualidadeato_schema)
return response
# ----------------------------------------------------
# Atualiza um registro existente de T_CENSEC_QUALIDADEATO
# ----------------------------------------------------
@router.put('/{censec_qualidadeato_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza um registro existente em T_CENSEC_QUALIDADEATO',
response_description='Atualiza um registro existente em T_CENSEC_QUALIDADEATO')
async def update(censec_qualidadeato_id: int, t_censec_qualidadeato_update_schema: TCensecQualidadeAtoUpdateSchema, current_user: dict = Depends(get_current_user)):
"""
Atualiza um registro existente de T_CENSEC_QUALIDADEATO com base no ID informado.
"""
t_censec_qualidadeato_update_schema.censec_qualidadeato_id = censec_qualidadeato_id
response = t_censec_qualidadeato_controller.update(t_censec_qualidadeato_update_schema)
return response
# ----------------------------------------------------
# Exclui um registro de T_CENSEC_QUALIDADEATO
# ----------------------------------------------------
@router.delete('/{censec_qualidadeato_id}',
status_code=status.HTTP_200_OK,
summary='Remove um registro de T_CENSEC_QUALIDADEATO',
response_description='Remove um registro de T_CENSEC_QUALIDADEATO')
async def delete(censec_qualidadeato_id: int, current_user: dict = Depends(get_current_user)):
"""
Remove um registro específico de T_CENSEC_QUALIDADEATO com base no ID informado.
"""
t_censec_qualidadeato_id_schema = TCensecQualidadeAtoIdSchema(censec_qualidadeato_id=censec_qualidadeato_id)
response = t_censec_qualidadeato_controller.delete(t_censec_qualidadeato_id_schema)
return response

View file

@ -0,0 +1,96 @@
# Importação de bibliotecas
from fastapi import APIRouter, Depends, status
from actions.jwt.get_current_user import get_current_user
from packages.v1.administrativo.controllers.t_censec_tipoato_controller import TCensecTipoAtoController
from packages.v1.administrativo.schemas.t_censec_tipoato_schema import (
TCensecTipoAtoIndexSchema,
TCensecTipoAtoSaveSchema,
TCensecTipoAtoUpdateSchema,
TCensecTipoAtoIdSchema
)
# ----------------------------------------------------
# Inicializa o roteador para as rotas da tabela T_CENSEC_TIPOATO
# ----------------------------------------------------
router = APIRouter()
# Instanciamento do controller
t_censec_tipoato_controller = TCensecTipoAtoController()
# ----------------------------------------------------
# Lista todos os registros de T_CENSEC_TIPOATO
# ----------------------------------------------------
@router.get('/',
status_code=status.HTTP_200_OK,
summary='Lista todos os registros de T_CENSEC_TIPOATO cadastrados',
response_description='Lista todos os registros de T_CENSEC_TIPOATO cadastrados')
async def index(current_user: dict = Depends(get_current_user)):
"""
Retorna todos os registros da tabela T_CENSEC_TIPOATO.
"""
response = t_censec_tipoato_controller.index()
return response
# ----------------------------------------------------
# Busca um registro específico de T_CENSEC_TIPOATO pelo ID
# ----------------------------------------------------
@router.get('/{censec_tipoato_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de T_CENSEC_TIPOATO pelo ID',
response_description='Busca um registro de T_CENSEC_TIPOATO em específico')
async def show(censec_tipoato_id: int, current_user: dict = Depends(get_current_user)):
"""
Retorna um registro específico de T_CENSEC_TIPOATO com base no ID informado.
"""
t_censec_tipoato_id_schema = TCensecTipoAtoIdSchema(censec_tipoato_id=censec_tipoato_id)
response = t_censec_tipoato_controller.show(t_censec_tipoato_id_schema)
return response
# ----------------------------------------------------
# Cadastra um novo registro em T_CENSEC_TIPOATO
# ----------------------------------------------------
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um novo registro em T_CENSEC_TIPOATO',
response_description='Cadastra um novo registro em T_CENSEC_TIPOATO')
async def save(t_censec_tipoato_schema: TCensecTipoAtoSaveSchema, current_user: dict = Depends(get_current_user)):
"""
Cria um novo registro na tabela T_CENSEC_TIPOATO.
"""
response = t_censec_tipoato_controller.save(t_censec_tipoato_schema)
return response
# ----------------------------------------------------
# Atualiza um registro existente de T_CENSEC_TIPOATO
# ----------------------------------------------------
@router.put('/{censec_tipoato_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza um registro existente em T_CENSEC_TIPOATO',
response_description='Atualiza um registro existente em T_CENSEC_TIPOATO')
async def update(censec_tipoato_id: int, t_censec_tipoato_update_schema: TCensecTipoAtoUpdateSchema, current_user: dict = Depends(get_current_user)):
"""
Atualiza um registro existente de T_CENSEC_TIPOATO com base no ID informado.
"""
t_censec_tipoato_update_schema.censec_tipoato_id = censec_tipoato_id
response = t_censec_tipoato_controller.update(t_censec_tipoato_update_schema)
return response
# ----------------------------------------------------
# Exclui um registro de T_CENSEC_TIPOATO
# ----------------------------------------------------
@router.delete('/{censec_tipoato_id}',
status_code=status.HTTP_200_OK,
summary='Remove um registro de T_CENSEC_TIPOATO',
response_description='Remove um registro de T_CENSEC_TIPOATO')
async def delete(censec_tipoato_id: int, current_user: dict = Depends(get_current_user)):
"""
Remove um registro específico de T_CENSEC_TIPOATO com base no ID informado.
"""
t_censec_tipoato_id_schema = TCensecTipoAtoIdSchema(censec_tipoato_id=censec_tipoato_id)
response = t_censec_tipoato_controller.delete(t_censec_tipoato_id_schema)
return response

View file

@ -0,0 +1,57 @@
from abstracts.repository import BaseRepository
from fastapi import HTTPException, status
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoIdSchema
class TCensecQualidadeAtoDeleteRepository(BaseRepository):
"""
Repositório responsável pela exclusão de registros na tabela
T_CENSEC_QUALIDADEATO.
"""
def execute(self, t_censec_qualidadeato_id_schema: TCensecQualidadeAtoIdSchema):
"""
Executa a exclusão de um registro específico da tabela T_CENSEC_QUALIDADEATO
com base no ID informado.
Args:
t_censec_qualidadeato_id_schema (TCensecQualidadeAtoIdSchema):
Esquema contendo o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
try:
# ----------------------------------------------------
# Montagem do SQL
# ----------------------------------------------------
sql = """
DELETE FROM T_CENSEC_QUALIDADEATO CQA
WHERE CQA.CENSEC_QUALIDADEATO_ID = :censec_qualidadeato_id
"""
# ----------------------------------------------------
# Preenchimento dos parâmetros
# ----------------------------------------------------
params = {
"censec_qualidadeato_id": t_censec_qualidadeato_id_schema.censec_qualidadeato_id
}
# ----------------------------------------------------
# Execução da instrução SQL
# ----------------------------------------------------
response = self.run(sql, params)
# ----------------------------------------------------
# Retorno do resultado
# ----------------------------------------------------
return response
except Exception as e:
# ----------------------------------------------------
# Tratamento de erro e exceção HTTP padronizada
# ----------------------------------------------------
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao excluir registro de T_CENSEC_QUALIDADEATO: {e}"
)

View file

@ -0,0 +1,42 @@
from abstracts.repository import BaseRepository
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoIndexSchema
class TCensecQualidadeAtoIndexRepository(BaseRepository):
"""
Repositório responsável pela operação de listagem de todos os registros
na tabela T_CENSEC_QUALIDADEATO.
"""
def execute(self, censec_qualidade_ato_index_schema: TCensecQualidadeAtoIndexSchema):
"""
Executa a consulta SQL para buscar todos os registros da tabela
T_CENSEC_QUALIDADEATO.
Returns:
Uma lista de dicionários contendo os dados dos registros.
"""
# ----------------------------------------------------
# Montagem do SQL
# ----------------------------------------------------
sql = """
SELECT *
FROM T_CENSEC_QUALIDADEATO WHERE censec_tipoato_id = :censec_tipoato_id
"""
# ----------------------------------------------------
# Montagem do parâmetros
# ----------------------------------------------------
params = {
"censec_tipoato_id": censec_qualidade_ato_index_schema.censec_tipoato_id
}
# ----------------------------------------------------
# Execução do SQL
# ----------------------------------------------------
response = self.fetch_all(sql, params)
# ----------------------------------------------------
# Retorno dos dados localizados
# ----------------------------------------------------
return response

View file

@ -0,0 +1,67 @@
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoSaveSchema
class TCensecQualidadeAtoSaveRepository(BaseRepository):
"""
Repositório responsável pela operação de salvamento de um novo registro
na tabela T_CENSEC_QUALIDADEATO.
"""
def execute(self, t_censec_qualidadeato_save_schema: TCensecQualidadeAtoSaveSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
t_censec_qualidadeato_save_schema (TCensecQualidadeAtoSaveSchema):
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_QUALIDADEATO (
CENSEC_QUALIDADEATO_ID,
CENSEC_TIPOATO_ID,
CENSEC_QUALIDADE_ID,
QTD_MINIMA
) VALUES (
:censec_qualidadeato_id,
:censec_tipoato_id,
:censec_qualidade_id,
:qtd_minima
)
RETURNING *;
"""
# ----------------------------------------------------
# Preenchimento dos parâmetros
# ----------------------------------------------------
params = {
"censec_qualidadeato_id": t_censec_qualidadeato_save_schema.censec_qualidadeato_id,
"censec_tipoato_id": t_censec_qualidadeato_save_schema.censec_tipoato_id,
"censec_qualidade_id": t_censec_qualidadeato_save_schema.censec_qualidade_id,
"qtd_minima": t_censec_qualidadeato_save_schema.qtd_minima,
}
# ----------------------------------------------------
# Execução do SQL e retorno do registro
# ----------------------------------------------------
return self.run_and_return(sql, params)
except Exception as e:
# ----------------------------------------------------
# Tratamento de erros e lançamento de exceção HTTP
# ----------------------------------------------------
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao salvar registro em T_CENSEC_QUALIDADEATO: {e}"
)

View file

@ -0,0 +1,68 @@
from abstracts.repository import BaseRepository
from fastapi import HTTPException, status
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoIdSchema
class TCensecQualidadeAtoShowRepository(BaseRepository):
"""
Repositório responsável pela operação de exibição de um registro
na tabela T_CENSEC_QUALIDADEATO.
"""
def execute(self, t_censec_qualidadeato_id_schema: TCensecQualidadeAtoIdSchema):
"""
Busca um registro específico de T_CENSEC_QUALIDADEATO pelo ID.
Args:
t_censec_qualidadeato_id_schema (TCensecQualidadeAtoIdSchema):
Esquema contendo o ID do registro a ser buscado.
Returns:
O registro encontrado ou levanta exceção HTTP 404 se não existir.
Raises:
HTTPException: Caso ocorra um erro na execução da query.
"""
try:
# ----------------------------------------------------
# Montagem do SQL
# ----------------------------------------------------
sql = """
SELECT *
FROM T_CENSEC_QUALIDADEATO CQA
WHERE CQA.CENSEC_QUALIDADEATO_ID = :censec_qualidadeato_id
"""
# ----------------------------------------------------
# Preenchimento de parâmetros
# ----------------------------------------------------
params = {
"censec_qualidadeato_id": t_censec_qualidadeato_id_schema.censec_qualidadeato_id
}
# ----------------------------------------------------
# Execução do SQL
# ----------------------------------------------------
result = self.fetch_one(sql, params)
# ----------------------------------------------------
# Validação de retorno
# ----------------------------------------------------
if not result:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Registro de T_CENSEC_QUALIDADEATO não encontrado."
)
return result
except HTTPException:
# Repassa exceções HTTP explícitas (como 404)
raise
except Exception as e:
# Captura falhas inesperadas de execução
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao buscar registro em T_CENSEC_QUALIDADEATO: {e}"
)

View file

@ -0,0 +1,60 @@
from abstracts.repository import BaseRepository
from api.actions.data.prepare_update_data import prepare_update_data
from fastapi import HTTPException, status
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoUpdateSchema
class TCensecQualidadeAtoUpdateRepository(BaseRepository):
"""
Repositório responsável pela operação de atualização de registros
na tabela T_CENSEC_QUALIDADEATO.
"""
def execute(self, t_censec_qualidadeato_update_schema: TCensecQualidadeAtoUpdateSchema):
"""
Atualiza um registro existente na tabela T_CENSEC_QUALIDADEATO.
Args:
t_censec_qualidadeato_update_schema (TCensecQualidadeAtoUpdateSchema):
Esquema contendo os dados a serem atualizados.
Returns:
O registro atualizado (via RETURNING *).
Raises:
HTTPException: Caso ocorra um erro na execução do SQL.
"""
try:
# ----------------------------------------------------
# Prepara parâmetros e colunas de atualização dinâmicas
# ----------------------------------------------------
params, update_columns = prepare_update_data(
t_censec_qualidadeato_update_schema,
exclude_fields=["censec_qualidadeato_id"],
id_field="censec_qualidadeato_id"
)
# ----------------------------------------------------
# Montagem do SQL dinâmico
# ----------------------------------------------------
sql = f"""
UPDATE T_CENSEC_QUALIDADEATO
SET {update_columns}
WHERE CENSEC_QUALIDADEATO_ID = :censec_qualidadeato_id
RETURNING *;
"""
# ----------------------------------------------------
# Execução e retorno do registro atualizado
# ----------------------------------------------------
response = self.run_and_return(sql, params)
return response
except Exception as e:
# ----------------------------------------------------
# Tratamento de exceção e retorno HTTP padronizado
# ----------------------------------------------------
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao atualizar registro em T_CENSEC_QUALIDADEATO: {str(e)}"
)

View file

@ -0,0 +1,46 @@
from abstracts.repository import BaseRepository
from fastapi import HTTPException, status
from api.packages.v1.administrativo.schemas.t_censec_tipoato_schema import TCensecTipoAtoIdSchema
class TCensecTipoAtoDeleteRepository(BaseRepository):
"""
Repositório responsável pela exclusão de registros na tabela
T_CENSEC_TIPOATO.
"""
def execute(self, t_censec_tipoato_id_schema: TCensecTipoAtoIdSchema):
"""
Executa a exclusão de um registro específico da tabela T_CENSEC_TIPOATO
com base no ID informado.
Args:
t_censec_tipoato_id_schema (TCensecTipoAtoIdSchema): Esquema contendo o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
try:
# Montagem do SQL
sql = """
DELETE FROM T_CENSEC_TIPOATO CTA
WHERE CTA.CENSEC_TIPOATO_ID = :censec_tipoato_id
"""
# Preenchimento dos parâmetros
params = {
"censec_tipoato_id": t_censec_tipoato_id_schema.censec_tipoato_id
}
# Execução da instrução SQL
response = self.run(sql, params)
# Retorna o resultado da exclusão
return response
except Exception as e:
# Lança exceção HTTP em caso de erro
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao excluir registro de T_CENSEC_TIPOATO: {e}"
)

View file

@ -0,0 +1,23 @@
from abstracts.repository import BaseRepository
class TCensecTipoAtoIndexRepository(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_TIPOATO """
# Execução do sql
response = self.fetch_all(sql)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,75 @@
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
from api.packages.v1.administrativo.schemas.t_censec_tipoato_schema import TCensecTipoAtoSaveSchema
class TCensecTipoAtoSaveRepository(BaseRepository):
"""
Repositório responsável pela operação de salvamento de um novo registro
na tabela T_CENSEC_TIPOATO.
"""
def execute(self, t_censec_tipoato_save_schema: TCensecTipoAtoSaveSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
t_censec_tipoato_save_schema (TCensecTipoAtoSchema): 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_TIPOATO (
CENSEC_TIPOATO_ID,
CENSEC_ID,
DESCRICAO,
SITUACAO,
TIPO_SEPARACAO,
TIPO_REVOGACAO,
CODIGO
) VALUES (
:censec_tipoato_id,
:censec_id,
:descricao,
:situacao,
:tipo_separacao,
:tipo_revogacao,
:codigo
)
RETURNING *;
"""
# ----------------------------------------------------
# Preenchimento dos parâmetros
# ----------------------------------------------------
params = {
"censec_tipoato_id": t_censec_tipoato_save_schema.censec_tipoato_id,
"censec_id": t_censec_tipoato_save_schema.censec_id,
"descricao": t_censec_tipoato_save_schema.descricao,
"situacao": t_censec_tipoato_save_schema.situacao,
"tipo_separacao": t_censec_tipoato_save_schema.tipo_separacao,
"tipo_revogacao": t_censec_tipoato_save_schema.tipo_revogacao,
"codigo": t_censec_tipoato_save_schema.codigo,
}
# ----------------------------------------------------
# Execução do SQL e retorno do registro
# ----------------------------------------------------
return self.run_and_return(sql, params)
except Exception as e:
# ----------------------------------------------------
# Tratamento de erros e lançamento de exceção HTTP
# ----------------------------------------------------
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao salvar registro em T_CENSEC_TIPOATO: {e}"
)

View file

@ -0,0 +1,69 @@
from abstracts.repository import BaseRepository
from fastapi import HTTPException, status
from api.packages.v1.administrativo.schemas.t_censec_tipoato_schema import TCensecTipoAtoIdSchema
class TCensecTipoAtoShowRepository(BaseRepository):
"""
Repositório responsável pela operação de exibição de um registro
na tabela T_CENSEC_TIPOATO.
"""
def execute(self, t_censec_tipoato_id_schema: TCensecTipoAtoIdSchema):
"""
Busca um registro específico de T_CENSEC_TIPOATO pelo ID.
Args:
t_censec_tipoato_id_schema (TCensecTipoAtoIdSchema):
Esquema contendo o ID do registro a ser buscado.
Returns:
O registro encontrado ou levanta exceção HTTP 404 se não existir.
Raises:
HTTPException: Caso ocorra um erro na execução da query.
"""
try:
# ----------------------------------------------------
# Montagem do SQL
# ----------------------------------------------------
sql = """
SELECT *
FROM T_CENSEC_TIPOATO CTA
WHERE CTA.CENSEC_TIPOATO_ID = :censec_tipoato_id
"""
# ----------------------------------------------------
# Preenchimento de parâmetros
# ----------------------------------------------------
params = {
"censec_tipoato_id": t_censec_tipoato_id_schema.censec_tipoato_id
}
# ----------------------------------------------------
# Execução do SQL
# ----------------------------------------------------
result = self.fetch_one(sql, params)
# ----------------------------------------------------
# Validação de retorno
# ----------------------------------------------------
if not result:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Registro de T_CENSEC_TIPOATO não encontrado."
)
return result
except HTTPException:
# Repassa exceções HTTP explícitas (como 404)
raise
except Exception as e:
# Captura falhas inesperadas de execução
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao buscar registro em T_CENSEC_TIPOATO: {e}"
)

View file

@ -0,0 +1,61 @@
from abstracts.repository import BaseRepository
from api.actions.data.prepare_update_data import prepare_update_data
from fastapi import HTTPException, status
from api.packages.v1.administrativo.schemas.t_censec_tipoato_schema import TCensecTipoAtoUpdateSchema
class TCensecTipoAtoUpdateRepository(BaseRepository):
"""
Repositório responsável pela operação de atualização de registros
na tabela T_CENSEC_TIPOATO.
"""
def execute(self, t_censec_tipoato_update_schema: TCensecTipoAtoUpdateSchema):
"""
Atualiza um registro existente na tabela T_CENSEC_TIPOATO.
Args:
t_censec_tipoato_update_schema (TCensecTipoAtoUpdateSchema):
Esquema contendo os dados a serem atualizados.
Returns:
O registro atualizado (via RETURNING *).
Raises:
HTTPException: Caso ocorra um erro na execução do SQL.
"""
try:
# ----------------------------------------------------
# Prepara parâmetros e colunas de atualização dinâmicas
# ----------------------------------------------------
params, update_columns = prepare_update_data(
t_censec_tipoato_update_schema,
exclude_fields=["censec_tipoato_id"],
id_field="censec_tipoato_id"
)
# ----------------------------------------------------
# Montagem do SQL dinâmico
# ----------------------------------------------------
sql = f"""
UPDATE T_CENSEC_TIPOATO
SET {update_columns}
WHERE CENSEC_TIPOATO_ID = :censec_tipoato_id
RETURNING *;
"""
# ----------------------------------------------------
# Execução e retorno do registro atualizado
# ----------------------------------------------------
response = self.run_and_return(sql, params)
return response
except Exception as e:
# ----------------------------------------------------
# Tratamento de exceção e retorno HTTP padronizado
# ----------------------------------------------------
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao atualizar registro em T_CENSEC_TIPOATO: {str(e)}"
)

View file

@ -0,0 +1,64 @@
from pydantic import BaseModel
from typing import Optional
# ----------------------------------------------------
# Schema base - representa a tabela T_CENSEC_QUALIDADEATO
# ----------------------------------------------------
class TCensecQualidadeAtoSchema(BaseModel):
censec_qualidadeato_id: Optional[float] = None
censec_tipoato_id: Optional[float] = None
censec_qualidade_id: Optional[float] = None
qtd_minima: Optional[float] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema de indexação/listagem (ex: GET /index)
# ----------------------------------------------------
class TCensecQualidadeAtoIndexSchema(BaseModel):
censec_qualidadeato_id: Optional[float] = None
censec_tipoato_id: Optional[float] = None
censec_qualidade_id: Optional[float] = None
qtd_minima: Optional[float] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar um registro pelo ID (GET /{id})
# ----------------------------------------------------
class TCensecQualidadeAtoIdSchema(BaseModel):
censec_qualidadeato_id: float
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para criação (POST)
# ----------------------------------------------------
class TCensecQualidadeAtoSaveSchema(BaseModel):
censec_qualidadeato_id: Optional[float] = None
censec_tipoato_id: Optional[float] = None
censec_qualidade_id: Optional[float] = None
qtd_minima: Optional[float] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para atualização (PUT)
# ----------------------------------------------------
class TCensecQualidadeAtoUpdateSchema(BaseModel):
censec_qualidadeato_id: Optional[float] = None
censec_tipoato_id: Optional[float] = None
censec_qualidade_id: Optional[float] = None
qtd_minima: Optional[float] = None
class Config:
from_attributes = True

View file

@ -0,0 +1,71 @@
from pydantic import BaseModel
from typing import Optional
# ----------------------------------------------------
# Schema base - representa a tabela T_CENSEC_TIPOATO
# ----------------------------------------------------
class TCensecTipoAtoSchema(BaseModel):
censec_tipoato_id: Optional[float] = None
censec_id: Optional[float] = None
descricao: Optional[str] = None
situacao: Optional[str] = None
tipo_separacao: Optional[str] = None
tipo_revogacao: Optional[str] = None
codigo: Optional[float] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema de indexação/listagem (ex: GET /index)
# ----------------------------------------------------
class TCensecTipoAtoIndexSchema(BaseModel):
censec_tipoato_id: float
descricao: Optional[str] = None
situacao: Optional[str] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar um registro pelo ID (GET /{id})
# ----------------------------------------------------
class TCensecTipoAtoIdSchema(BaseModel):
censec_tipoato_id: float
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para criação (POST)
# ----------------------------------------------------
class TCensecTipoAtoSaveSchema(BaseModel):
censec_tipoato_id: Optional[float] = None
censec_id: Optional[float] = None
descricao: Optional[str] = None
situacao: Optional[str] = None
tipo_separacao: Optional[str] = None
tipo_revogacao: Optional[str] = None
codigo: Optional[float] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para atualização (PUT)
# ----------------------------------------------------
class TCensecTipoAtoUpdateSchema(BaseModel):
censec_tipoato_id: Optional[float] = None
censec_id: Optional[float] = None
descricao: Optional[str] = None
situacao: Optional[str] = None
tipo_separacao: Optional[str] = None
tipo_revogacao: Optional[str] = None
codigo: Optional[float] = None
class Config:
from_attributes = True

View file

@ -0,0 +1,35 @@
from api.packages.v1.administrativo.actions.t_censec_qualidadeato.t_censec_qualidadeato_delete_action import TCensecQualidadeAtoDeleteAction
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoIdSchema
class TCensecQualidadeAtoDeleteService:
"""
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_ATO.
"""
def execute(self, t_censec_qualidade_ato_id_schema: TCensecQualidadeAtoIdSchema):
"""
Executa a operação de exclusão do registro no banco de dados.
Args:
t_censec_qualidade_ato_id_schema (TCensecQualidadeAtoIdSchema):
Esquema contendo o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# ----------------------------------------------------
# Instanciamento da ação responsável pela exclusão
# ----------------------------------------------------
t_censec_qualidade_ato_delete_action = TCensecQualidadeAtoDeleteAction()
# ----------------------------------------------------
# Execução da ação e obtenção do resultado
# ----------------------------------------------------
response = t_censec_qualidade_ato_delete_action.execute(t_censec_qualidade_ato_id_schema)
# ----------------------------------------------------
# Retorno do resultado da operação
# ----------------------------------------------------
return response

View file

@ -0,0 +1,42 @@
from api.packages.v1.administrativo.actions.t_censec_qualidadeato.t_censec_qualidadeato_index_action import TCensecQualidadeAtoIndexAction
from fastapi import HTTPException, status
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoIndexSchema
class TCensecQualidadeAtoIndexService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de registros na tabela T_CENSEC_QUALIDADEATO.
"""
def execute(self, censec_qualidade_ato_index_schema: TCensecQualidadeAtoIndexSchema):
"""
Executa a operação de busca de todos os registros no banco de dados.
Returns:
A lista de registros encontrados.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
t_censec_qualidadeato_index_action = TCensecQualidadeAtoIndexAction()
# ----------------------------------------------------
# Execução da ação
# ----------------------------------------------------
data = t_censec_qualidadeato_index_action.execute(censec_qualidade_ato_index_schema)
# ----------------------------------------------------
# Verificação de retorno
# ----------------------------------------------------
if not data:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Não foi possível localizar registros de T_CENSEC_QUALIDADEATO."
)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return data

View file

@ -0,0 +1,54 @@
from api.actions.dynamic_import.dynamic_import import DynamicImport
from api.packages.v1.administrativo.actions.t_censec_qualidadeato.t_censec_qualidadeato_save_action import TCensecQualidadeAtoSaveAction
from api.packages.v1.sequencia.schemas.g_sequencia import GSequenciaSchema
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoSaveSchema
from api.packages.v1.sequencia.services.g_sequencia.generate_service import GenerateService
class TCensecQualidadeAtoSaveService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de criação de registros na tabela T_CENSEC_QUALIDADEATO.
"""
def __init__(self):
# ----------------------------------------------------
# Inicializa o DynamicImport para permitir injeção dinâmica
# de pacotes, conforme o padrão do sistema.
# ----------------------------------------------------
self.dynamic_import = DynamicImport()
self.dynamic_import.set_package("administrativo")
self.dynamic_import.set_table("t_censec_qualidadeato")
def execute(self, t_censec_qualidadeato_save_schema: TCensecQualidadeAtoSaveSchema):
"""
Executa a operação de salvamento do registro no banco de dados.
Args:
t_censec_qualidadeato_save_schema (TCensecQualidadeAtoSaveSchema):
O esquema com os dados a serem salvos.
Returns:
O registro recém-criado.
"""
# ----------------------------------------------------
# Geração automática de ID (sequência)
# ----------------------------------------------------
if not t_censec_qualidadeato_save_schema.censec_qualidadeato_id:
# Cria o schema de sequência
sequencia_schema = GSequenciaSchema()
sequencia_schema.tabela = "T_CENSEC_QUALIDADEATO"
# Gera a sequência atualizada
generate = GenerateService()
sequencia = generate.execute(sequencia_schema)
# Atualiza o ID no schema
t_censec_qualidadeato_save_schema.censec_qualidadeato_id = sequencia.sequencia
# ----------------------------------------------------
# Instanciamento e execução da Action de salvamento
# ----------------------------------------------------
t_censec_qualidadeato_save_action = TCensecQualidadeAtoSaveAction()
return t_censec_qualidadeato_save_action.execute(t_censec_qualidadeato_save_schema)

View file

@ -0,0 +1,45 @@
from api.packages.v1.administrativo.actions.t_censec_qualidadeato.t_censec_qualidadeato_show_action import TCensecQualidadeAtoShowAction
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoIdSchema
from fastapi import HTTPException, status
class TCensecQualidadeAtoShowService:
"""
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_QUALIDADEATO.
"""
def execute(self, t_censec_qualidadeato_id_schema: TCensecQualidadeAtoIdSchema):
"""
Executa a operação de busca no banco de dados.
Args:
t_censec_qualidadeato_id_schema (TCensecQualidadeAtoIdSchema):
O esquema com o ID do registro a ser buscado.
Returns:
O resultado da busca.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
t_censec_qualidadeato_show_action = TCensecQualidadeAtoShowAction()
# ----------------------------------------------------
# Execução da ação
# ----------------------------------------------------
data = t_censec_qualidadeato_show_action.execute(t_censec_qualidadeato_id_schema)
# ----------------------------------------------------
# Verificação de resultado
# ----------------------------------------------------
if not data:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Não foi possível localizar o registro de T_CENSEC_QUALIDADEATO."
)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return data

View file

@ -0,0 +1,30 @@
from api.packages.v1.administrativo.actions.t_censec_qualidadeato.t_censec_qualidadeato_update_action import TCensecQualidadeAtoUpdateAction
from api.packages.v1.administrativo.schemas.t_censec_qualidadeato_schema import TCensecQualidadeAtoUpdateSchema
class TCensecQualidadeAtoUpdateService:
"""
Serviço responsável pela operação de atualização de um registro
na tabela T_CENSEC_QUALIDADEATO.
"""
def execute(self, t_censec_qualidadeato_update_schema: TCensecQualidadeAtoUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
t_censec_qualidadeato_update_schema (TCensecQualidadeAtoUpdateSchema):
O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
t_censec_qualidadeato_update_action = TCensecQualidadeAtoUpdateAction()
# ----------------------------------------------------
# Execução da ação e retorno do resultado
# ----------------------------------------------------
return t_censec_qualidadeato_update_action.execute(t_censec_qualidadeato_update_schema)

View file

@ -0,0 +1,35 @@
from api.packages.v1.administrativo.actions.t_censec_tipoato.t_censec_tipoato_delete_action import TCensecTipoAtoDeleteAction
from api.packages.v1.administrativo.schemas.t_censec_tipoato_schema import TCensecTipoAtoIdSchema
class TCensecTipoAtoDeleteService:
"""
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_TIPOATO.
"""
def execute(self, t_censec_tipoato_id_schema: TCensecTipoAtoIdSchema):
"""
Executa a operação de exclusão do registro no banco de dados.
Args:
t_censec_tipoato_id_schema (TCensecTipoAtoIdSchema):
O esquema com o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
t_censec_tipoato_delete_action = TCensecTipoAtoDeleteAction()
# ----------------------------------------------------
# Execução da ação
# ----------------------------------------------------
data = t_censec_tipoato_delete_action.execute(t_censec_tipoato_id_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return data

View file

@ -0,0 +1,44 @@
from api.packages.v1.administrativo.actions.t_censec_tipoato.t_censec_tipoato_index_action import TCensecTipoAtoIndexAction
from fastapi import HTTPException, status
class TCensecTipoAtoIndexService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de registros na tabela T_CENSEC_TIPOATO.
"""
def execute(self):
"""
Executa a operação de busca de todos os registros no banco de dados.
Args:
t_censec_tipoato_index_schema (TCensecTipoAtoIndexSchema):
Esquema que pode conter filtros ou parâmetros de busca.
Returns:
A lista de registros encontrados.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
t_censec_tipoato_index_action = TCensecTipoAtoIndexAction()
# ----------------------------------------------------
# Execução da ação
# ----------------------------------------------------
data = t_censec_tipoato_index_action.execute()
# ----------------------------------------------------
# Verificação de retorno
# ----------------------------------------------------
if not data:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Não foi possível localizar registros de T_CENSEC_TIPOATO."
)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return data

View file

@ -0,0 +1,54 @@
from api.actions.dynamic_import.dynamic_import import DynamicImport
from api.packages.v1.administrativo.actions.t_censec_tipoato.t_censec_tipoato_save_action import TCensecTipoAtoSaveAction
from api.packages.v1.sequencia.schemas.g_sequencia import GSequenciaSchema
from api.packages.v1.administrativo.schemas.t_censec_tipoato_schema import TCensecTipoAtoSaveSchema
from api.packages.v1.sequencia.services.g_sequencia.generate_service import GenerateService
class TCensecTipoAtoSaveService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de criação de registros na tabela T_CENSEC_TIPOATO.
"""
def __init__(self):
# ----------------------------------------------------
# Inicializa o DynamicImport para permitir injeção dinâmica
# de pacotes, conforme padrão do sistema.
# ----------------------------------------------------
self.dynamic_import = DynamicImport()
self.dynamic_import.set_package("administrativo")
self.dynamic_import.set_table("t_censec_tipoato")
def execute(self, t_censec_tipoato_save_schema: TCensecTipoAtoSaveSchema):
"""
Executa a operação de salvamento do registro no banco de dados.
Args:
t_censec_tipoato_save_schema (TCensecTipoAtoSaveSchema):
O esquema com os dados a serem salvos.
Returns:
O registro recém-criado.
"""
# ----------------------------------------------------
# Geração automática de ID (sequência)
# ----------------------------------------------------
if not t_censec_tipoato_save_schema.censec_tipoato_id:
# Cria o schema de sequência
sequencia_schema = GSequenciaSchema()
sequencia_schema.tabela = "T_CENSEC_TIPOATO"
# Gera a sequência atualizada
generate = GenerateService()
sequencia = generate.execute(sequencia_schema)
# Atualiza o ID no schema
t_censec_tipoato_save_schema.censec_tipoato_id = sequencia.sequencia
# ----------------------------------------------------
# Instanciamento e execução da Action de salvamento
# ----------------------------------------------------
t_censec_tipoato_save_action = TCensecTipoAtoSaveAction()
return t_censec_tipoato_save_action.execute(t_censec_tipoato_save_schema)

View file

@ -0,0 +1,45 @@
from api.packages.v1.administrativo.actions.t_censec_tipoato.t_censec_tipoato_show_action import TCensecTipoAtoShowAction
from api.packages.v1.administrativo.schemas.t_censec_tipoato_schema import TCensecTipoAtoIdSchema
from fastapi import HTTPException, status
class TCensecTipoAtoShowService:
"""
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_TIPOATO.
"""
def execute(self, t_censec_tipoato_id_schema: TCensecTipoAtoIdSchema):
"""
Executa a operação de busca no banco de dados.
Args:
t_censec_tipoato_id_schema (TCensecTipoAtoIdSchema):
O esquema com o ID do registro a ser buscado.
Returns:
O resultado da busca.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
t_censec_tipoato_show_action = TCensecTipoAtoShowAction()
# ----------------------------------------------------
# Execução da ação
# ----------------------------------------------------
data = t_censec_tipoato_show_action.execute(t_censec_tipoato_id_schema)
# ----------------------------------------------------
# Verificação de resultado
# ----------------------------------------------------
if not data:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Não foi possível localizar o registro de T_CENSEC_TIPOATO."
)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return data

View file

@ -0,0 +1,30 @@
from api.packages.v1.administrativo.actions.t_censec_tipoato.t_censec_tipoato_update_action import TCensecTipoAtoUpdateAction
from api.packages.v1.administrativo.schemas.t_censec_tipoato_schema import TCensecTipoAtoUpdateSchema
class TCensecTipoAtoUpdateService:
"""
Serviço responsável pela operação de atualização de um registro
na tabela T_CENSEC_TIPOATO.
"""
def execute(self, t_censec_tipoato_update_schema: TCensecTipoAtoUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
t_censec_tipoato_update_schema (TCensecTipoAtoUpdateSchema):
O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
t_censec_tipoato_update_action = TCensecTipoAtoUpdateAction()
# ----------------------------------------------------
# Execução da ação e retorno do resultado
# ----------------------------------------------------
return t_censec_tipoato_update_action.execute(t_censec_tipoato_update_schema)

View file

@ -2,6 +2,8 @@
from fastapi import APIRouter from fastapi import APIRouter
# Importa os módulos de rotas específicos # Importa os módulos de rotas específicos
from packages.v1.administrativo.endpoints import t_censec_tipoato_endpoint
from packages.v1.administrativo.endpoints import t_censec_qualidadeato_endpoint
from packages.v1.administrativo.endpoints import g_tb_txmodelogrupo_endpoint from packages.v1.administrativo.endpoints import g_tb_txmodelogrupo_endpoint
from packages.v1.administrativo.endpoints import g_tb_profissao_endpoint from packages.v1.administrativo.endpoints import g_tb_profissao_endpoint
from packages.v1.administrativo.endpoints import c_caixa_item_endpoint from packages.v1.administrativo.endpoints import c_caixa_item_endpoint
@ -177,3 +179,13 @@ api_router.include_router(
api_router.include_router( api_router.include_router(
t_pessoa_representante_endpoint.router, prefix="/administrativo/t_pessoa_representante", tags=["Representante de pessoas jurídicas"] t_pessoa_representante_endpoint.router, prefix="/administrativo/t_pessoa_representante", tags=["Representante de pessoas jurídicas"]
) )
# Inclui as rotas de t_imovel_unidade
api_router.include_router(
t_censec_tipoato_endpoint.router, prefix="/administrativo/t_censec_tipoato", tags=["CENSEC Tipo atos"]
)
# Inclui as rotas de t_imovel_unidade
api_router.include_router(
t_censec_qualidadeato_endpoint.router, prefix="/administrativo/t_censec_qualidadeato", tags=["CENSEC Qualidade Ato"]
)