[MVPTN-32] feat(CRUD): Cria cruds para manipular GEmolumento e GSeloGrupo

This commit is contained in:
Keven Willian Pereira de Souza 2025-10-24 10:52:55 -03:00
parent 252427ae0e
commit 59f462c7bd
37 changed files with 1821 additions and 2 deletions

View file

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

View file

@ -0,0 +1,35 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.g_emolumento.g_emolumento_index_repository import GEmolumentoIndexRepository
class GEmolumentoIndexAction(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 G_NATUREZA_TITULO.
"""
def execute(self):
"""
Executa a operação de listagem no banco de dados.
Args:
g_emolumento_index_schema (GEmolumentoIndexSchema):
Esquema contendo parâmetros opcionais de filtro.
Returns:
A lista de registros encontrados.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_index_repository = GEmolumentoIndexRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_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.g_emolumento.g_emolumento_save_repository import GEmolumentoSaveRepository
from api.packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoSaveSchema
class GEmolumentoSaveAction(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 G_EMOLUMENTO.
"""
def execute(self, g_emolumento_save_schema: GEmolumentoSaveSchema):
"""
Executa a operação de salvamento.
Args:
g_emolumento_schema (GEmolumentoSchema):
O esquema com os dados a serem persistidos.
Returns:
O resultado da operação de salvamento.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_save_repository = GEmolumentoSaveRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_save_repository.execute(g_emolumento_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.g_emolumento.g_emolumento_show_repository import GEmolumentoShowRepository
from api.packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoIdSchema
class GEmolumentoShowAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(self, g_emolumento_id_schema: GEmolumentoIdSchema):
"""
Executa a operação de exibição.
Args:
g_emolumento_id_schema (GEmolumentoIdSchema):
O esquema com o ID do registro a ser exibido.
Returns:
O resultado da operação de exibição.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_show_repository = GEmolumentoShowRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_show_repository.execute(g_emolumento_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.g_emolumento.g_emolumento_update_repository import GEmolumentoUpdateRepository
from api.packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoUpdateSchema
class GEmolumentoUpdateAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(self, g_emolumento_update_schema: GEmolumentoUpdateSchema):
"""
Executa a operação de atualização.
Args:
g_emolumento_update_schema (GEmolumentoUpdateSchema):
O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# ----------------------------------------------------
# Instanciamento do repositório de atualização
# ----------------------------------------------------
g_emolumento_update_repository = GEmolumentoUpdateRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_update_repository.execute(g_emolumento_update_schema)
# ----------------------------------------------------
# Retorno do resultado
# ----------------------------------------------------
return response

View file

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

View file

@ -0,0 +1,35 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.g_selo_grupo.g_selo_grupo_index_repository import GSeloGrupoIndexRepository
class GSeloGrupoIndexAction(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 G_NATUREZA_TITULO.
"""
def execute(self):
"""
Executa a operação de listagem no banco de dados.
Args:
g_selo_grupo_index_schema (GSeloGrupoIndexSchema):
Esquema contendo parâmetros opcionais de filtro.
Returns:
A lista de registros encontrados.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_selo_grupo_index_repository = GSeloGrupoIndexRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_selo_grupo_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.g_selo_grupo.g_selo_grupo_save_repository import GSeloGrupoSaveRepository
from api.packages.v1.administrativo.schemas.g_selo_grupo_schema import GSeloGrupoSaveSchema
class GSeloGrupoSaveAction(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 G_GRAMATICA.
"""
def execute(self, g_selo_grupo_save_schema: GSeloGrupoSaveSchema):
"""
Executa a operação de salvamento.
Args:
g_selo_grupo_schema (GSeloGrupoSchema):
O esquema com os dados a serem persistidos.
Returns:
O resultado da operação de salvamento.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_selo_grupo_save_repository = GSeloGrupoSaveRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_selo_grupo_save_repository.execute(g_selo_grupo_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.g_selo_grupo.g_selo_grupo_show_repository import GSeloGrupoShowRepository
from api.packages.v1.administrativo.schemas.g_selo_grupo_schema import GSeloGrupoIdSchema
class GSeloGrupoShowAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(self, g_selo_grupo_id_schema: GSeloGrupoIdSchema):
"""
Executa a operação de exibição.
Args:
g_selo_grupo_id_schema (GSeloGrupoIdSchema):
O esquema com o ID do registro a ser exibido.
Returns:
O resultado da operação de exibição.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_selo_grupo_show_repository = GSeloGrupoShowRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_selo_grupo_show_repository.execute(g_selo_grupo_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.g_selo_grupo.g_selo_grupo_update_repository import GSeloGrupoUpdateRepository
from api.packages.v1.administrativo.schemas.g_selo_grupo_schema import GSeloGrupoUpdateSchema
class GSeloGrupoUpdateAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela G_NATUREZA_TITULO.
"""
def execute(self, g_selo_grupo_update_schema: GSeloGrupoUpdateSchema):
"""
Executa a operação de atualização.
Args:
g_selo_grupo_update_schema (GSeloGrupoUpdateSchema):
O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# ----------------------------------------------------
# Instanciamento do repositório de atualização
# ----------------------------------------------------
g_selo_grupo_update_repository = GSeloGrupoUpdateRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_selo_grupo_update_repository.execute(g_selo_grupo_update_schema)
# ----------------------------------------------------
# Retorno do resultado
# ----------------------------------------------------
return response

View file

@ -0,0 +1,101 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.g_emolumento_schema import (
GEmolumentoSaveSchema,
GEmolumentoUpdateSchema,
GEmolumentoIdSchema
)
class GEmolumentoController:
"""
Controller responsável por orquestrar as operações CRUD da tabela G_EMOLUMENTO,
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("g_emolumento")
# ----------------------------------------------------
# Lista todos os registros de G_EMOLUMENTO
# ----------------------------------------------------
def index(self):
# Importação da classe desejada
index_service = self.dynamic_import.service("g_emolumento_index_service", "GEmolumentoIndexService")
# Instância da classe service
self.index_service = index_service()
# Execução da listagem
return {
"message": "Registros de G_EMOLUMENTO localizados com sucesso.",
"data": self.index_service.execute(),
}
# ----------------------------------------------------
# Busca um registro específico de G_EMOLUMENTO pelo ID
# ----------------------------------------------------
def show(self, g_emolumento_id_schema: GEmolumentoIdSchema):
# Importação da classe desejada
show_service = self.dynamic_import.service("g_emolumento_show_service", "GEmolumentoShowService")
# Instância da classe service
self.show_service = show_service()
# Execução da busca
return {
"message": "Registro de G_EMOLUMENTO localizado com sucesso.",
"data": self.show_service.execute(g_emolumento_id_schema),
}
# ----------------------------------------------------
# Cadastra um novo registro em G_EMOLUMENTO
# ----------------------------------------------------
def save(self, g_emolumento_save_schema: GEmolumentoSaveSchema):
# Importação da classe desejada
save_service = self.dynamic_import.service("g_emolumento_save_service", "GEmolumentoSaveService")
# Instância da classe service
self.save_service = save_service()
# Execução do salvamento
return {
"message": "Registro de G_EMOLUMENTO salvo com sucesso.",
"data": self.save_service.execute(g_emolumento_save_schema),
}
# ----------------------------------------------------
# Atualiza um registro existente de G_EMOLUMENTO
# ----------------------------------------------------
def update(self, g_emolumento_update_schema: GEmolumentoUpdateSchema):
# Importação da classe desejada
update_service = self.dynamic_import.service("g_emolumento_update_service", "GEmolumentoUpdateService")
# Instância da classe service
self.update_service = update_service()
# Execução da atualização
return {
"message": "Registro de G_EMOLUMENTO atualizado com sucesso.",
"data": self.update_service.execute(g_emolumento_update_schema),
}
# ----------------------------------------------------
# Exclui um registro de G_EMOLUMENTO
# ----------------------------------------------------
def delete(self, g_emolumento_id_schema: GEmolumentoIdSchema):
# Importação da classe desejada
delete_service = self.dynamic_import.service("g_emolumento_delete_service", "GEmolumentoDeleteService")
# Instância da classe service
self.delete_service = delete_service()
# Execução da exclusão
return {
"message": "Registro de G_EMOLUMENTO removido com sucesso.",
"data": self.delete_service.execute(g_emolumento_id_schema),
}

View file

@ -0,0 +1,101 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.g_selo_grupo_schema import (
GSeloGrupoSaveSchema,
GSeloGrupoUpdateSchema,
GSeloGrupoIdSchema
)
class GSeloGrupoController:
"""
Controller responsável por orquestrar as operações CRUD da tabela G_SELO_GRUPO,
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("g_selo_grupo")
# ----------------------------------------------------
# Lista todos os registros de G_SELO_GRUPO
# ----------------------------------------------------
def index(self):
# Importação da classe desejada
index_service = self.dynamic_import.service("g_selo_grupo_index_service", "GSeloGrupoIndexService")
# Instância da classe service
self.index_service = index_service()
# Execução da listagem
return {
"message": "Registros de G_SELO_GRUPO localizados com sucesso.",
"data": self.index_service.execute(),
}
# ----------------------------------------------------
# Busca um registro específico de G_SELO_GRUPO pelo ID
# ----------------------------------------------------
def show(self, g_selo_grupo_id_schema: GSeloGrupoIdSchema):
# Importação da classe desejada
show_service = self.dynamic_import.service("g_selo_grupo_show_service", "GSeloGrupoShowService")
# Instância da classe service
self.show_service = show_service()
# Execução da busca
return {
"message": "Registro de G_SELO_GRUPO localizado com sucesso.",
"data": self.show_service.execute(g_selo_grupo_id_schema),
}
# ----------------------------------------------------
# Cadastra um novo registro em G_SELO_GRUPO
# ----------------------------------------------------
def save(self, g_selo_grupo_save_schema: GSeloGrupoSaveSchema):
# Importação da classe desejada
save_service = self.dynamic_import.service("g_selo_grupo_save_service", "GSeloGrupoSaveService")
# Instância da classe service
self.save_service = save_service()
# Execução do salvamento
return {
"message": "Registro de G_SELO_GRUPO salvo com sucesso.",
"data": self.save_service.execute(g_selo_grupo_save_schema),
}
# ----------------------------------------------------
# Atualiza um registro existente de G_SELO_GRUPO
# ----------------------------------------------------
def update(self, g_selo_grupo_update_schema: GSeloGrupoUpdateSchema):
# Importação da classe desejada
update_service = self.dynamic_import.service("g_selo_grupo_update_service", "GSeloGrupoUpdateService")
# Instância da classe service
self.update_service = update_service()
# Execução da atualização
return {
"message": "Registro de G_SELO_GRUPO atualizado com sucesso.",
"data": self.update_service.execute(g_selo_grupo_update_schema),
}
# ----------------------------------------------------
# Exclui um registro de G_SELO_GRUPO
# ----------------------------------------------------
def delete(self, g_selo_grupo_id_schema: GSeloGrupoIdSchema):
# Importação da classe desejada
delete_service = self.dynamic_import.service("g_selo_grupo_delete_service", "GSeloGrupoDeleteService")
# Instância da classe service
self.delete_service = delete_service()
# Execução da exclusão
return {
"message": "Registro de G_SELO_GRUPO removido com sucesso.",
"data": self.delete_service.execute(g_selo_grupo_id_schema),
}

View file

@ -0,0 +1,94 @@
# 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.g_emolumento_controller import GEmolumentoController
from packages.v1.administrativo.schemas.g_emolumento_schema import (
GEmolumentoSaveSchema,
GEmolumentoUpdateSchema,
GEmolumentoIdSchema
)
# ----------------------------------------------------
# Inicializa o roteador para as rotas da tabela G_EMOLUMENTO
# ----------------------------------------------------
router = APIRouter()
# Instanciamento do controller
g_emolumento_controller = GEmolumentoController()
# ----------------------------------------------------
# Lista todos os registros de G_EMOLUMENTO
# ----------------------------------------------------
@router.get('/',
status_code=status.HTTP_200_OK,
summary='Lista todos os registros de G_EMOLUMENTO cadastrados',
response_description='Lista todos os registros de G_EMOLUMENTO cadastrados')
async def index(current_user: dict = Depends(get_current_user)):
"""
Retorna todos os registros da tabela G_EMOLUMENTO.
"""
response = g_emolumento_controller.index()
return response
# ----------------------------------------------------
# Busca um registro específico de G_EMOLUMENTO pelo ID
# ----------------------------------------------------
@router.get('/{emolumento_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de G_EMOLUMENTO pelo ID',
response_description='Busca um registro de G_EMOLUMENTO em específico')
async def show(emolumento_id: int, current_user: dict = Depends(get_current_user)):
"""
Retorna um registro específico de G_EMOLUMENTO com base no ID informado.
"""
g_emolumento_id_schema = GEmolumentoIdSchema(emolumento_id=emolumento_id)
response = g_emolumento_controller.show(g_emolumento_id_schema)
return response
# ----------------------------------------------------
# Cadastra um novo registro em G_EMOLUMENTO
# ----------------------------------------------------
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um novo registro em G_EMOLUMENTO',
response_description='Cadastra um novo registro em G_EMOLUMENTO')
async def save(g_emolumento_schema: GEmolumentoSaveSchema, current_user: dict = Depends(get_current_user)):
"""
Cria um novo registro na tabela G_EMOLUMENTO.
"""
response = g_emolumento_controller.save(g_emolumento_schema)
return response
# ----------------------------------------------------
# Atualiza um registro existente de G_EMOLUMENTO
# ----------------------------------------------------
@router.put('/{emolumento_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza um registro existente em G_EMOLUMENTO',
response_description='Atualiza um registro existente em G_EMOLUMENTO')
async def update(emolumento_id: int, g_emolumento_update_schema: GEmolumentoUpdateSchema, current_user: dict = Depends(get_current_user)):
"""
Atualiza um registro existente de G_EMOLUMENTO com base no ID informado.
"""
g_emolumento_update_schema.emolumento_id = emolumento_id
response = g_emolumento_controller.update(g_emolumento_update_schema)
return response
# ----------------------------------------------------
# Exclui um registro de G_EMOLUMENTO
# ----------------------------------------------------
@router.delete('/{emolumento_id}',
status_code=status.HTTP_200_OK,
summary='Remove um registro de G_EMOLUMENTO',
response_description='Remove um registro de G_EMOLUMENTO')
async def delete(emolumento_id: int, current_user: dict = Depends(get_current_user)):
"""
Remove um registro específico de G_EMOLUMENTO com base no ID informado.
"""
g_emolumento_id_schema = GEmolumentoIdSchema(emolumento_id=emolumento_id)
response = g_emolumento_controller.delete(g_emolumento_id_schema)
return response

View file

@ -0,0 +1,94 @@
# 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.g_selo_grupo_controller import GSeloGrupoController
from packages.v1.administrativo.schemas.g_selo_grupo_schema import (
GSeloGrupoSaveSchema,
GSeloGrupoUpdateSchema,
GSeloGrupoIdSchema
)
# ----------------------------------------------------
# Inicializa o roteador para as rotas da tabela G_SELO_GRUPO
# ----------------------------------------------------
router = APIRouter()
# Instanciamento do controller
g_selo_grupo_controller = GSeloGrupoController()
# ----------------------------------------------------
# Lista todos os registros de G_SELO_GRUPO
# ----------------------------------------------------
@router.get('/',
status_code=status.HTTP_200_OK,
summary='Lista todos os registros de G_SELO_GRUPO cadastrados',
response_description='Lista todos os registros de G_SELO_GRUPO cadastrados')
async def index(current_user: dict = Depends(get_current_user)):
"""
Retorna todos os registros da tabela G_SELO_GRUPO.
"""
response = g_selo_grupo_controller.index()
return response
# ----------------------------------------------------
# Busca um registro específico de G_SELO_GRUPO pelo ID
# ----------------------------------------------------
@router.get('/{selo_grupo_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de G_SELO_GRUPO pelo ID',
response_description='Busca um registro de G_SELO_GRUPO em específico')
async def show(selo_grupo_id: int, current_user: dict = Depends(get_current_user)):
"""
Retorna um registro específico de G_SELO_GRUPO com base no ID informado.
"""
g_selo_grupo_id_schema = GSeloGrupoIdSchema(selo_grupo_id=selo_grupo_id)
response = g_selo_grupo_controller.show(g_selo_grupo_id_schema)
return response
# ----------------------------------------------------
# Cadastra um novo registro em G_SELO_GRUPO
# ----------------------------------------------------
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um novo registro em G_SELO_GRUPO',
response_description='Cadastra um novo registro em G_SELO_GRUPO')
async def save(g_selo_grupo_schema: GSeloGrupoSaveSchema, current_user: dict = Depends(get_current_user)):
"""
Cria um novo registro na tabela G_SELO_GRUPO.
"""
response = g_selo_grupo_controller.save(g_selo_grupo_schema)
return response
# ----------------------------------------------------
# Atualiza um registro existente de G_SELO_GRUPO
# ----------------------------------------------------
@router.put('/{selo_grupo_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza um registro existente em G_SELO_GRUPO',
response_description='Atualiza um registro existente em G_SELO_GRUPO')
async def update(selo_grupo_id: int, g_selo_grupo_update_schema: GSeloGrupoUpdateSchema, current_user: dict = Depends(get_current_user)):
"""
Atualiza um registro existente de G_SELO_GRUPO com base no ID informado.
"""
g_selo_grupo_update_schema.selo_grupo_id = selo_grupo_id
response = g_selo_grupo_controller.update(g_selo_grupo_update_schema)
return response
# ----------------------------------------------------
# Exclui um registro de G_SELO_GRUPO
# ----------------------------------------------------
@router.delete('/{selo_grupo_id}',
status_code=status.HTTP_200_OK,
summary='Remove um registro de G_SELO_GRUPO',
response_description='Remove um registro de G_SELO_GRUPO')
async def delete(selo_grupo_id: int, current_user: dict = Depends(get_current_user)):
"""
Remove um registro específico de G_SELO_GRUPO com base no ID informado.
"""
g_selo_grupo_id_schema = GSeloGrupoIdSchema(selo_grupo_id=selo_grupo_id)
response = g_selo_grupo_controller.delete(g_selo_grupo_id_schema)
return response

View file

@ -0,0 +1,45 @@
from abstracts.repository import BaseRepository
from fastapi import HTTPException, status
from api.packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoIdSchema
class GEmolumentoDeleteRepository(BaseRepository):
"""
Repositório responsável pela exclusão de registros na tabela
G_EMOLUMENTO.
"""
def execute(self, g_emolumento_id_schema: GEmolumentoIdSchema):
"""
Executa a exclusão de um registro específico da tabela G_EMOLUMENTO
com base no ID informado.
Args:
g_emolumento_id_schema (GEmolumentoIdSchema): 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 G_EMOLUMENTO GG
WHERE GG.EMOLUMENTO_ID = :emolumento_id
"""
# Preenchimento dos parâmetros
params = {
"emolumento_id": g_emolumento_id_schema.emolumento_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 G_EMOLUMENTO: {e}"
)

View file

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

View file

@ -0,0 +1,50 @@
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
from api.actions.data.generate_insert_sql import generate_insert_sql
from api.packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoSaveSchema
class GEmolumentoSaveRepository(BaseRepository):
"""
Repositório responsável pela operação de salvamento de um novo registro
na tabela G_EMOLUMENTO.
"""
def execute(self, g_emolumento_save_schema: GEmolumentoSaveSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
g_emolumento_save_schema (GEmolumentoSchema): 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:
# ----------------------------------------------------
# Preenchimento dos parâmetros
# ----------------------------------------------------
params = g_emolumento_save_schema.model_dump(exclude_unset=True)
# ----------------------------------------------------
# Montagem do SQL dinâmico
# ----------------------------------------------------
sql = generate_insert_sql('G_EMOLUMENTO', params)
# ----------------------------------------------------
# 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 G_EMOLUMENTO: {e}"
)

View file

@ -0,0 +1,67 @@
from abstracts.repository import BaseRepository
from fastapi import HTTPException, status
from api.packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoIdSchema
class GEmolumentoShowRepository(BaseRepository):
"""
Repositório responsável pela operação de exibição de um registro
na tabela G_EMOLUMENTO.
"""
def execute(self, g_emolumento_id_schema: GEmolumentoIdSchema):
"""
Busca um registro específico de G_EMOLUMENTO pelo ID.
Args:
g_emolumento_id_schema (GEmolumentoIdSchema):
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 G_EMOLUMENTO GG
WHERE GG.EMOLUMENTO_ID = :emolumento_id
"""
# ----------------------------------------------------
# Preenchimento de parâmetros
# ----------------------------------------------------
params = g_emolumento_id_schema.model_dump(exclude_unset=True)
# ----------------------------------------------------
# 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 G_EMOLUMENTO 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 G_EMOLUMENTO: {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.g_emolumento_schema import GEmolumentoUpdateSchema
class GEmolumentoUpdateRepository(BaseRepository):
"""
Repositório responsável pela operação de atualização de registros
na tabela G_EMOLUMENTO.
"""
def execute(self, g_emolumento_update_schema: GEmolumentoUpdateSchema):
"""
Atualiza um registro existente na tabela G_EMOLUMENTO.
Args:
g_emolumento_update_schema (GEmolumentoUpdateSchema):
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(
g_emolumento_update_schema,
exclude_fields=["emolumento_id"],
id_field="emolumento_id"
)
# ----------------------------------------------------
# Montagem do SQL dinâmico
# ----------------------------------------------------
sql = f"""
UPDATE G_EMOLUMENTO
SET {update_columns}
WHERE EMOLUMENTO_ID = :emolumento_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 G_EMOLUMENTO: {str(e)}"
)

View file

@ -0,0 +1,45 @@
from abstracts.repository import BaseRepository
from fastapi import HTTPException, status
from api.packages.v1.administrativo.schemas.g_selo_grupo_schema import GSeloGrupoIdSchema
class GSeloGrupoDeleteRepository(BaseRepository):
"""
Repositório responsável pela exclusão de registros na tabela
G_SELO_GRUPO.
"""
def execute(self, g_selo_grupo_id_schema: GSeloGrupoIdSchema):
"""
Executa a exclusão de um registro específico da tabela G_SELO_GRUPO
com base no ID informado.
Args:
g_selo_grupo_id_schema (GSeloGrupoIdSchema): 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 G_SELO_GRUPO GG
WHERE GG.SELO_GRUPO_ID = :selo_grupo_id
"""
# Preenchimento dos parâmetros
params = {
"selo_grupo_id": g_selo_grupo_id_schema.selo_grupo_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 G_SELO_GRUPO: {e}"
)

View file

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

View file

@ -0,0 +1,50 @@
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
from api.actions.data.generate_insert_sql import generate_insert_sql
from api.packages.v1.administrativo.schemas.g_selo_grupo_schema import GSeloGrupoSaveSchema
class GSeloGrupoSaveRepository(BaseRepository):
"""
Repositório responsável pela operação de salvamento de um novo registro
na tabela G_SELO_GRUPO.
"""
def execute(self, g_selo_grupo_save_schema: GSeloGrupoSaveSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
g_selo_grupo_save_schema (GSeloGrupoSchema): 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:
# ----------------------------------------------------
# Preenchimento dos parâmetros
# ----------------------------------------------------
params = g_selo_grupo_save_schema.model_dump(exclude_unset=True)
# ----------------------------------------------------
# Montagem do SQL dinâmico
# ----------------------------------------------------
sql = generate_insert_sql('G_SELO_GRUPO', params)
# ----------------------------------------------------
# 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 G_SELO_GRUPO: {e}"
)

View file

@ -0,0 +1,67 @@
from abstracts.repository import BaseRepository
from fastapi import HTTPException, status
from api.packages.v1.administrativo.schemas.g_selo_grupo_schema import GSeloGrupoIdSchema
class GSeloGrupoShowRepository(BaseRepository):
"""
Repositório responsável pela operação de exibição de um registro
na tabela G_SELO_GRUPO.
"""
def execute(self, g_selo_grupo_id_schema: GSeloGrupoIdSchema):
"""
Busca um registro específico de G_SELO_GRUPO pelo ID.
Args:
g_selo_grupo_id_schema (GSeloGrupoIdSchema):
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 G_SELO_GRUPO GG
WHERE GG.SELO_GRUPO_ID = :selo_grupo_id
"""
# ----------------------------------------------------
# Preenchimento de parâmetros
# ----------------------------------------------------
params = g_selo_grupo_id_schema.model_dump(exclude_unset=True)
# ----------------------------------------------------
# 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 G_SELO_GRUPO 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 G_SELO_GRUPO: {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.g_selo_grupo_schema import GSeloGrupoUpdateSchema
class GSeloGrupoUpdateRepository(BaseRepository):
"""
Repositório responsável pela operação de atualização de registros
na tabela G_SELO_GRUPO.
"""
def execute(self, g_selo_grupo_update_schema: GSeloGrupoUpdateSchema):
"""
Atualiza um registro existente na tabela G_SELO_GRUPO.
Args:
g_selo_grupo_update_schema (GSeloGrupoUpdateSchema):
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(
g_selo_grupo_update_schema,
exclude_fields=["selo_grupo_id"],
id_field="selo_grupo_id"
)
# ----------------------------------------------------
# Montagem do SQL dinâmico
# ----------------------------------------------------
sql = f"""
UPDATE G_SELO_GRUPO
SET {update_columns}
WHERE SELO_GRUPO_ID = :selo_grupo_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 G_SELO_GRUPO: {str(e)}"
)

View file

@ -0,0 +1,89 @@
from pydantic import BaseModel
from typing import Optional
# ----------------------------------------------------
# Schema base - representa a tabela G_EMOLUMENTO
# ----------------------------------------------------
class GEmolumentoSchema(BaseModel):
emolumento_id: Optional[float] = None
descricao: Optional[str] = None
tipo: Optional[str] = None
sistema_id: Optional[float] = None
selo_grupo_id: Optional[float] = None
reg_averb: Optional[str] = None
pre_definido: Optional[str] = None
situacao: Optional[str] = None
situacao_ri: Optional[str] = None
com_reducao: Optional[str] = None
motivo_reducao: Optional[str] = None
valor_maximo_certidao: Optional[float] = None
tipo_objetivo: Optional[str] = None
modelo_tag: Optional[str] = None
codigo_nota_id: Optional[float] = None
convenio_codhab: Optional[str] = None
item_df: Optional[str] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar um registro pelo ID (GET /{id})
# ----------------------------------------------------
class GEmolumentoIdSchema(BaseModel):
emolumento_id: float
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para criação (POST)
# ----------------------------------------------------
class GEmolumentoSaveSchema(BaseModel):
emolumento_id: Optional[float] = None
descricao: Optional[str] = None
tipo: Optional[str] = None
sistema_id: Optional[float] = None
selo_grupo_id: Optional[float] = None
reg_averb: Optional[str] = None
pre_definido: Optional[str] = None
situacao: Optional[str] = None
situacao_ri: Optional[str] = None
com_reducao: Optional[str] = None
motivo_reducao: Optional[str] = None
valor_maximo_certidao: Optional[float] = None
tipo_objetivo: Optional[str] = None
modelo_tag: Optional[str] = None
codigo_nota_id: Optional[float] = None
convenio_codhab: Optional[str] = None
item_df: Optional[str] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para atualização (PUT)
# ----------------------------------------------------
class GEmolumentoUpdateSchema(BaseModel):
emolumento_id: Optional[float] = None
descricao: Optional[str] = None
tipo: Optional[str] = None
sistema_id: Optional[float] = None
selo_grupo_id: Optional[float] = None
reg_averb: Optional[str] = None
pre_definido: Optional[str] = None
situacao: Optional[str] = None
situacao_ri: Optional[str] = None
com_reducao: Optional[str] = None
motivo_reducao: Optional[str] = None
valor_maximo_certidao: Optional[float] = None
tipo_objetivo: Optional[str] = None
modelo_tag: Optional[str] = None
codigo_nota_id: Optional[float] = None
convenio_codhab: Optional[str] = None
item_df: Optional[str] = None
class Config:
from_attributes = True

View file

@ -0,0 +1,84 @@
from pydantic import BaseModel
from typing import Optional
# ----------------------------------------------------
# Schema base - representa a tabela G_SELO_GRUPO
# ----------------------------------------------------
class GSeloGrupoSchema(BaseModel):
selo_grupo_id: Optional[float] = None
descricao: Optional[str] = None
numero: Optional[float] = None
situacao: Optional[str] = None
controle_automatico: Optional[str] = None
sistema_id: Optional[float] = None
valor: Optional[float] = None
tipo_cartorio: Optional[str] = None
descricao_completa: Optional[str] = None
agrupador: Optional[str] = None
um_por_protocolo: Optional[str] = None
numero_principal_ini: Optional[float] = None
numero_principal_fim: Optional[float] = None
selo_grupo_id_principal: Optional[float] = None
envio_automatico: Optional[str] = None
selo_grupo_id_agrupador: Optional[float] = None
codigo_conta: Optional[float] = None
id_tipo_ato_antigo: Optional[float] = None
grupos_principal: Optional[str] = None
sigla: Optional[str] = None
tipo_selo: Optional[str] = None
natureza: Optional[float] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar um registro pelo ID (GET /{id})
# ----------------------------------------------------
class GSeloGrupoIdSchema(BaseModel):
selo_grupo_id: float
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para criação (POST)
# ----------------------------------------------------
class GSeloGrupoSaveSchema(BaseModel):
selo_grupo_id: Optional[float] = None
descricao: Optional[str] = None
numero: Optional[float] = None
situacao: Optional[str] = None
controle_automatico: Optional[str] = None
sistema_id: Optional[float] = None
valor: Optional[float] = None
tipo_cartorio: Optional[str] = None
descricao_completa: Optional[str] = None
agrupador: Optional[str] = None
um_por_protocolo: Optional[str] = None
numero_principal_ini: Optional[float] = None
numero_principal_fim: Optional[float] = None
selo_grupo_id_principal: Optional[float] = None
envio_automatico: Optional[str] = None
selo_grupo_id_agrupador: Optional[float] = None
codigo_conta: Optional[float] = None
id_tipo_ato_antigo: Optional[float] = None
grupos_principal: Optional[str] = None
sigla: Optional[str] = None
tipo_selo: Optional[str] = None
natureza: Optional[float] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para atualização (PUT)
# ----------------------------------------------------
class GSeloGrupoUpdateSchema(GSeloGrupoSaveSchema):
selo_grupo_id: Optional[float] = None
class Config:
from_attributes = True

View file

@ -0,0 +1,35 @@
from api.packages.v1.administrativo.actions.g_emolumento.g_emolumento_delete_action import GEmolumentoDeleteAction
from api.packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoIdSchema
class GEmolumentoDeleteService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela G_EMOLUMENTO.
"""
def execute(self, g_emolumento_id_schema: GEmolumentoIdSchema):
"""
Executa a operação de exclusão do registro no banco de dados.
Args:
g_emolumento_id_schema (GEmolumentoIdSchema):
O esquema com o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
g_emolumento_delete_action = GEmolumentoDeleteAction()
# ----------------------------------------------------
# Execução da ação
# ----------------------------------------------------
data = g_emolumento_delete_action.execute(g_emolumento_id_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return data

View file

@ -0,0 +1,44 @@
from api.packages.v1.administrativo.actions.g_emolumento.g_emolumento_index_action import GEmolumentoIndexAction
from fastapi import HTTPException, status
class GEmolumentoIndexService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de registros na tabela G_EMOLUMENTO.
"""
def execute(self):
"""
Executa a operação de busca de todos os registros no banco de dados.
Args:
g_emolumento_index_schema (GEmolumentoIndexSchema):
Esquema que pode conter filtros ou parâmetros de busca.
Returns:
A lista de registros encontrados.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
g_emolumento_index_action = GEmolumentoIndexAction()
# ----------------------------------------------------
# Execução da ação
# ----------------------------------------------------
data = g_emolumento_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 G_EMOLUMENTO."
)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return data

View file

@ -0,0 +1,44 @@
from api.packages.v1.administrativo.actions.g_emolumento.g_emolumento_save_action import GEmolumentoSaveAction
from api.packages.v1.sequencia.schemas.g_sequencia import GSequenciaSchema
from api.packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoSaveSchema
from api.packages.v1.sequencia.services.g_sequencia.generate_service import GenerateService
class GEmolumentoSaveService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de criação de registros na tabela G_EMOLUMENTO.
"""
def execute(self, g_emolumento_save_schema: GEmolumentoSaveSchema):
"""
Executa a operação de salvamento do registro no banco de dados.
Args:
g_emolumento_save_schema (GEmolumentoSaveSchema):
O esquema com os dados a serem salvos.
Returns:
O registro recém-criado.
"""
# ----------------------------------------------------
# Geração automática de ID (sequência)
# ----------------------------------------------------
if not g_emolumento_save_schema.emolumento_id:
# Cria o schema de sequência
sequencia_schema = GSequenciaSchema()
sequencia_schema.tabela = "G_EMOLUMENTO"
# Gera a sequência atualizada
generate = GenerateService()
sequencia = generate.execute(sequencia_schema)
# Atualiza o ID no schema
g_emolumento_save_schema.emolumento_id = sequencia.sequencia
# ----------------------------------------------------
# Instanciamento e execução da Action de salvamento
# ----------------------------------------------------
g_emolumento_save_action = GEmolumentoSaveAction()
return g_emolumento_save_action.execute(g_emolumento_save_schema)

View file

@ -0,0 +1,45 @@
from api.packages.v1.administrativo.actions.g_emolumento.g_emolumento_show_action import GEmolumentoShowAction
from api.packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoIdSchema
from fastapi import HTTPException, status
class GEmolumentoShowService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela G_EMOLUMENTO.
"""
def execute(self, g_emolumento_id_schema: GEmolumentoIdSchema):
"""
Executa a operação de busca no banco de dados.
Args:
g_emolumento_id_schema (GEmolumentoIdSchema):
O esquema com o ID do registro a ser buscado.
Returns:
O resultado da busca.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
g_emolumento_show_action = GEmolumentoShowAction()
# ----------------------------------------------------
# Execução da ação
# ----------------------------------------------------
data = g_emolumento_show_action.execute(g_emolumento_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 G_EMOLUMENTO."
)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return data

View file

@ -0,0 +1,30 @@
from api.packages.v1.administrativo.actions.g_emolumento.g_emolumento_update_action import GEmolumentoUpdateAction
from api.packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoUpdateSchema
class GEmolumentoUpdateService:
"""
Serviço responsável pela operação de atualização de um registro
na tabela G_EMOLUMENTO.
"""
def execute(self, g_emolumento_update_schema: GEmolumentoUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
g_emolumento_update_schema (GEmolumentoUpdateSchema):
O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
g_emolumento_update_action = GEmolumentoUpdateAction()
# ----------------------------------------------------
# Execução da ação e retorno do resultado
# ----------------------------------------------------
return g_emolumento_update_action.execute(g_emolumento_update_schema)

View file

@ -0,0 +1,35 @@
from api.packages.v1.administrativo.actions.g_selo_grupo.g_selo_grupo_delete_action import GSeloGrupoDeleteAction
from api.packages.v1.administrativo.schemas.g_selo_grupo_schema import GSeloGrupoIdSchema
class GSeloGrupoDeleteService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela G_SELO_GRUPO.
"""
def execute(self, g_selo_grupo_id_schema: GSeloGrupoIdSchema):
"""
Executa a operação de exclusão do registro no banco de dados.
Args:
g_selo_grupo_id_schema (GSeloGrupoIdSchema):
O esquema com o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
g_selo_grupo_delete_action = GSeloGrupoDeleteAction()
# ----------------------------------------------------
# Execução da ação
# ----------------------------------------------------
data = g_selo_grupo_delete_action.execute(g_selo_grupo_id_schema)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return data

View file

@ -0,0 +1,44 @@
from api.packages.v1.administrativo.actions.g_selo_grupo.g_selo_grupo_index_action import GSeloGrupoIndexAction
from fastapi import HTTPException, status
class GSeloGrupoIndexService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de registros na tabela G_SELO_GRUPO.
"""
def execute(self):
"""
Executa a operação de busca de todos os registros no banco de dados.
Args:
g_selo_grupo_index_schema (GSeloGrupoIndexSchema):
Esquema que pode conter filtros ou parâmetros de busca.
Returns:
A lista de registros encontrados.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
g_selo_grupo_index_action = GSeloGrupoIndexAction()
# ----------------------------------------------------
# Execução da ação
# ----------------------------------------------------
data = g_selo_grupo_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 G_SELO_GRUPO."
)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return data

View file

@ -0,0 +1,44 @@
from api.packages.v1.administrativo.actions.g_selo_grupo.g_selo_grupo_save_action import GSeloGrupoSaveAction
from api.packages.v1.sequencia.schemas.g_sequencia import GSequenciaSchema
from api.packages.v1.administrativo.schemas.g_selo_grupo_schema import GSeloGrupoSaveSchema
from api.packages.v1.sequencia.services.g_sequencia.generate_service import GenerateService
class GSeloGrupoSaveService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de criação de registros na tabela G_SELO_GRUPO.
"""
def execute(self, g_selo_grupo_save_schema: GSeloGrupoSaveSchema):
"""
Executa a operação de salvamento do registro no banco de dados.
Args:
g_selo_grupo_save_schema (GSeloGrupoSaveSchema):
O esquema com os dados a serem salvos.
Returns:
O registro recém-criado.
"""
# ----------------------------------------------------
# Geração automática de ID (sequência)
# ----------------------------------------------------
if not g_selo_grupo_save_schema.selo_grupo_id:
# Cria o schema de sequência
sequencia_schema = GSequenciaSchema()
sequencia_schema.tabela = "G_SELO_GRUPO"
# Gera a sequência atualizada
generate = GenerateService()
sequencia = generate.execute(sequencia_schema)
# Atualiza o ID no schema
g_selo_grupo_save_schema.selo_grupo_id = sequencia.sequencia
# ----------------------------------------------------
# Instanciamento e execução da Action de salvamento
# ----------------------------------------------------
g_selo_grupo_save_action = GSeloGrupoSaveAction()
return g_selo_grupo_save_action.execute(g_selo_grupo_save_schema)

View file

@ -0,0 +1,45 @@
from api.packages.v1.administrativo.actions.g_selo_grupo.g_selo_grupo_show_action import GSeloGrupoShowAction
from api.packages.v1.administrativo.schemas.g_selo_grupo_schema import GSeloGrupoIdSchema
from fastapi import HTTPException, status
class GSeloGrupoShowService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela G_SELO_GRUPO.
"""
def execute(self, g_selo_grupo_id_schema: GSeloGrupoIdSchema):
"""
Executa a operação de busca no banco de dados.
Args:
g_selo_grupo_id_schema (GSeloGrupoIdSchema):
O esquema com o ID do registro a ser buscado.
Returns:
O resultado da busca.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
g_selo_grupo_show_action = GSeloGrupoShowAction()
# ----------------------------------------------------
# Execução da ação
# ----------------------------------------------------
data = g_selo_grupo_show_action.execute(g_selo_grupo_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 G_SELO_GRUPO."
)
# ----------------------------------------------------
# Retorno da informação
# ----------------------------------------------------
return data

View file

@ -0,0 +1,30 @@
from api.packages.v1.administrativo.actions.g_selo_grupo.g_selo_grupo_update_action import GSeloGrupoUpdateAction
from api.packages.v1.administrativo.schemas.g_selo_grupo_schema import GSeloGrupoUpdateSchema
class GSeloGrupoUpdateService:
"""
Serviço responsável pela operação de atualização de um registro
na tabela G_SELO_GRUPO.
"""
def execute(self, g_selo_grupo_update_schema: GSeloGrupoUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
g_selo_grupo_update_schema (GSeloGrupoUpdateSchema):
O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# ----------------------------------------------------
# Instanciamento da ação
# ----------------------------------------------------
g_selo_grupo_update_action = GSeloGrupoUpdateAction()
# ----------------------------------------------------
# Execução da ação e retorno do resultado
# ----------------------------------------------------
return g_selo_grupo_update_action.execute(g_selo_grupo_update_schema)

View file

@ -2,7 +2,7 @@
from fastapi import APIRouter
# Importa os módulos de rotas específicos
from api.packages.v1.administrativo.endpoints import g_cartorio_endpoint, g_emolumento_periodo_endpoint, g_gramatica_endpoint, g_natureza_titulo_endpoint, t_ato_partetipo_endpoint, t_censec_tiponatureza_endpoint
from api.packages.v1.administrativo.endpoints import g_cartorio_endpoint, g_emolumento_endpoint, g_emolumento_periodo_endpoint, g_gramatica_endpoint, g_natureza_titulo_endpoint, g_selo_grupo_endpoint, t_ato_partetipo_endpoint, t_censec_tiponatureza_endpoint
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
@ -216,7 +216,17 @@ api_router.include_router(
g_cartorio_endpoint.router, prefix="/administrativo/g_cartorio", tags=["Dados do Cartório"]
)
# Inclui as rotas de g_cartorio
# Inclui as rotas de g_emolumento
api_router.include_router(
g_emolumento_endpoint.router, prefix="/administrativo/g_emolumento", tags=["Emolumentos"]
)
# Inclui as rotas de g_emolumento periodo
api_router.include_router(
g_emolumento_periodo_endpoint.router, prefix="/administrativo/g_emolumento_periodo", tags=["Periodos de Emolumentos"]
)
# Inclui as rotas de g_emolumento periodo
api_router.include_router(
g_selo_grupo_endpoint.router, prefix="/administrativo/g_selo_grupo", tags=["Grupos de Selos"]
)