Merge remote-tracking branch 'origin/release(MVP/Sprint5)' into release(MVP)

This commit is contained in:
Keven Willian Pereira de Souza 2025-09-26 17:30:45 -03:00
commit f4dea586a7
44 changed files with 3640 additions and 6 deletions

File diff suppressed because it is too large Load diff

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -0,0 +1,26 @@
from packages.v1.administrativo.schemas.g_natureza_schema import GNaturezaUpdateSchema
from packages.v1.administrativo.repositories.g_natureza.g_natureza_update_repository import UpdateRepository
class UpdateAction:
"""
Service responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela g_natureza.
"""
def execute(self, natureza_id: int, natureza_schema: GNaturezaUpdateSchema):
"""
Executa a operação de atualização.
Args:
natureza_id (int): O ID do registro a ser atualizado.
natureza_schema (GNaturezaUpdateSchema): O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# Instância o repositório de atualização
update_repository = UpdateRepository()
# Chama o método de execução do repositório para realizar a atualização
return update_repository.execute(natureza_id, natureza_schema)

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -0,0 +1,26 @@
from packages.v1.administrativo.schemas.g_tb_documentotipo_schema import GTbDocumentoTipoUpdateSchema
from packages.v1.administrativo.repositories.g_tb_documentotipo.g_tb_documentotipo_update_repository import UpdateRepository
class UpdateAction:
"""
Service responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela g_tb_documentotipo.
"""
def execute(self, documento_tipo_id: int, documento_tipo_schema: GTbDocumentoTipoUpdateSchema):
"""
Executa a operação de atualização.
Args:
documento_tipo_id (int): O ID do registro a ser atualizado.
documento_tipo_schema (GTbDocumentoTipoUpdateSchema): O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# Instância o repositório de atualização
update_repository = UpdateRepository()
# Chama o método de execução do repositório para realizar a atualização
return update_repository.execute(documento_tipo_id, documento_tipo_schema)

View file

@ -0,0 +1,113 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.g_natureza_schema import (
GNaturezaSchema,
GNaturezaSaveSchema,
GNaturezaUpdateSchema,
GNaturezaIdSchema,
GNaturezaDescricaoSchema
)
class GNaturezaController:
def __init__(self):
# Action responsável por carregar as services de acordo com o estado
self.dynamic_import = DynamicImport()
# Define o pacote que deve ser carregado
self.dynamic_import.set_package("administrativo")
# Define a tabela que o pacote pertence
self.dynamic_import.set_table("g_natureza")
pass
# Lista todos os registros de natureza
def index(self):
# Importação da classe desejada
indexService = self.dynamic_import.service("g_natureza_index_service", "IndexService")
# Instância da classe service
self.indexService = indexService()
# Lista todos os registros de natureza
return {
'message': 'Registros de natureza localizados com sucesso',
'data': self.indexService.execute()
}
# Busca um registro de natureza específico pelo ID
def show(self, natureza_schema: GNaturezaIdSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('g_natureza_show_service', 'ShowService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de natureza desejado
return {
'message': 'Registro de natureza localizado com sucesso',
'data': self.show_service.execute(natureza_schema)
}
# Busca um registro de natureza pela descrição
def get_by_descricao(self, natureza_schema: GNaturezaDescricaoSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('g_natureza_get_descricao_service', 'GetByDescricaoService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de natureza desejado
return {
'message': 'Registro de natureza localizado com sucesso',
'data': self.show_service.execute(natureza_schema, True)
}
# Cadastra um novo registro de natureza
def save(self, natureza_schema: GNaturezaSaveSchema):
#Importação da classe desejada
save_service = self.dynamic_import.service('g_natureza_save_service', 'SaveService')
# Instância da classe desejada
self.save_service = save_service()
# Busca e retorna o registro de natureza desejado
return {
'message': 'Registro de natureza salvo com sucesso',
'data': self.save_service.execute(natureza_schema)
}
# Atualiza os dados de um registro de natureza
def update(self, natureza_id: int, natureza_schema: GNaturezaUpdateSchema):
#Importação da classe desejada
update_service = self.dynamic_import.service('g_natureza_update_service', 'UpdateService')
# Instância da classe desejada
self.update_service = update_service()
# Busca e retorna o registro de natureza desejado
return {
'message': 'Registro de natureza atualizado com sucesso',
'data': self.update_service.execute(natureza_id, natureza_schema)
}
# Exclui um registro de natureza
def delete(self, natureza_schema: GNaturezaIdSchema):
#Importação da classe desejada
delete_service = self.dynamic_import.service('g_natureza_delete_service', 'DeleteService')
# Instância da classe desejada
self.delete_service = delete_service()
# Busca e retorna o registro de natureza desejado
return {
'message': 'Registro de natureza removido com sucesso',
'data': self.delete_service.execute(natureza_schema)
}

View file

@ -0,0 +1,113 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.g_tb_documentotipo_schema import (
GTbDocumentoTipoSchema,
GTbDocumentoTipoSaveSchema,
GTbDocumentoTipoUpdateSchema,
GTbDocumentoTipoIdSchema,
GTbDocumentoTipoDescricaoSchema
)
class GTbDocumentoTipoController:
def __init__(self):
# Action responsável por carregar as services de acordo com o estado
self.dynamic_import = DynamicImport()
# Define o pacote que deve ser carregado
self.dynamic_import.set_package("administrativo")
# Define a tabela que o pacote pertence
self.dynamic_import.set_table("g_tb_documentotipo")
pass
# Lista todos os registros de documento_tipo
def index(self):
# Importação da classe desejada
indexService = self.dynamic_import.service("g_tb_documentotipo_index_service", "IndexService")
# Instância da classe service
self.indexService = indexService()
# Lista todos os registros de documento_tipo
return {
'message': 'Registros de documento_tipo localizados com sucesso',
'data': self.indexService.execute()
}
# Busca um registro de documento_tipo específico pelo ID
def show(self, documento_tipo_schema: GTbDocumentoTipoIdSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('g_tb_documentotipo_show_service', 'ShowService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de documento_tipo desejado
return {
'message': 'Registro de documento_tipo localizado com sucesso',
'data': self.show_service.execute(documento_tipo_schema)
}
# Busca um registro de documento_tipo pela descrição
def get_by_descricao(self, documento_tipo_schema: GTbDocumentoTipoDescricaoSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('g_tb_documentotipo_get_by_descricao_service', 'GetByDescricaoService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de documento_tipo desejado
return {
'message': 'Registro de documento_tipo localizado com sucesso',
'data': self.show_service.execute(documento_tipo_schema, True)
}
# Cadastra um novo registro de documento_tipo
def save(self, documento_tipo_schema: GTbDocumentoTipoSaveSchema):
#Importação da classe desejada
save_service = self.dynamic_import.service('g_tb_documentotipo_save_service', 'SaveService')
# Instância da classe desejada
self.save_service = save_service()
# Busca e retorna o registro de documento_tipo desejado
return {
'message': 'Registro de documento_tipo salvo com sucesso',
'data': self.save_service.execute(documento_tipo_schema)
}
# Atualiza os dados de um registro de documento_tipo
def update(self, documento_tipo_id: int, documento_tipo_schema: GTbDocumentoTipoUpdateSchema):
#Importação da classe desejada
update_service = self.dynamic_import.service('g_tb_documentotipo_update_service', 'UpdateService')
# Instância da classe desejada
self.update_service = update_service()
# Busca e retorna o registro de documento_tipo desejado
return {
'message': 'Registro de documento_tipo atualizado com sucesso',
'data': self.update_service.execute(documento_tipo_id, documento_tipo_schema)
}
# Exclui um registro de documento_tipo
def delete(self, documento_tipo_schema: GTbDocumentoTipoIdSchema):
#Importação da classe desejada
delete_service = self.dynamic_import.service('g_tb_documentotipo_delete_service', 'DeleteService')
# Instância da classe desejada
self.delete_service = delete_service()
# Busca e retorna o registro de documento_tipo desejado
return {
'message': 'Registro de documento_tipo removido com sucesso',
'data': self.delete_service.execute(documento_tipo_schema)
}

View file

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

View file

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

View file

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

View file

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

View file

@ -0,0 +1,23 @@
from abstracts.repository import BaseRepository
class IndexRepository(BaseRepository):
"""
Repositório para a operação de listagem de todos os registros
na tabela g_natureza.
"""
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 G_NATUREZA """
# Execução do sql
response = self.fetch_all(sql)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,62 @@
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.g_natureza_schema import GNaturezaSaveSchema
class SaveRepository(BaseRepository):
"""
Repositório para a operação de salvamento de um novo registro na tabela g_natureza.
"""
def execute(self, natureza_schema: GNaturezaSaveSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
natureza_schema (GNaturezaSaveSchema): 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 G_NATUREZA(
NATUREZA_ID,
DESCRICAO,
SITUACAO,
SISTEMA_ID,
PEDIR_NUMERO_IMOVEL,
CONTROLE_FRENTEVERSO
) VALUES (
:natureza_id,
:descricao,
:situacao,
:sistema_id,
:pedir_numero_imovel,
:controle_frenteverso
) RETURNING *;"""
# Preenchimento de parâmetros
params = {
'natureza_id': natureza_schema.natureza_id,
'descricao': natureza_schema.descricao,
'situacao': natureza_schema.situacao,
'sistema_id': natureza_schema.sistema_id,
'pedir_numero_imovel': natureza_schema.pedir_numero_imovel,
'controle_frenteverso': natureza_schema.controle_frenteverso
}
# Execução do sql
return self.run_and_return(sql, params)
except Exception as e:
# Informa que houve uma falha no salvamento do registro
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao salvar NATUREZA: {e}"
)

View file

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

View file

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

View file

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

View file

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

View file

@ -0,0 +1,31 @@
from abstracts.repository import BaseRepository
class IndexRepository(BaseRepository):
"""
Repositório para a operação de listagem de todos os registros
na tabela G_TB_DOCUMENTOTIPO.
"""
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 TB_DOCUMENTOTIPO_ID,
DESCRICAO,
SITUACAO,
POSSUI_NUMERACAO,
ORGAO_PADRAO,
DESCRICAO_SIMPLIFICADA,
TIPO,
DESCRICAO_SINTER
FROM G_TB_DOCUMENTOTIPO """
# Execução do sql
response = self.fetch_all(sql)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,71 @@
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.g_tb_documentotipo_schema import GTbDocumentoTipoSaveSchema
class SaveRepository(BaseRepository):
"""
Repositório para a operação de salvamento de um novo registro na tabela G_TB_DOCUMENTOTIPO.
"""
def execute(self, documentotipo_schema: GTbDocumentoTipoSaveSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
documentotipo_schema (GtbDocumentotipoSaveSchema): 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 G_TB_DOCUMENTOTIPO(
TB_DOCUMENTOTIPO_ID,
DESCRICAO,
TEXTO,
SITUACAO,
POSSUI_NUMERACAO,
ORGAO_PADRAO,
DESCRICAO_SIMPLIFICADA,
TIPO,
DESCRICAO_SINTER
) VALUES (
:tb_documentotipo_id,
:descricao,
:texto,
:situacao,
:possui_numeracao,
:orgao_padrao,
:descricao_simplificada,
:tipo,
:descricao_sinter
) RETURNING *;"""
# Preenchimento de parâmetros
params = {
'tb_documentotipo_id': documentotipo_schema.tb_documentotipo_id,
'descricao': documentotipo_schema.descricao,
'texto': documentotipo_schema.texto,
'situacao': documentotipo_schema.situacao,
'possui_numeracao': documentotipo_schema.possui_numeracao,
'orgao_padrao': documentotipo_schema.orgao_padrao,
'descricao_simplificada': documentotipo_schema.descricao_simplificada,
'tipo': documentotipo_schema.tipo,
'descricao_sinter': documentotipo_schema.descricao_sinter
}
# Execução do sql
return self.run_and_return(sql, params)
except Exception as e:
# Informa que houve uma falha no salvamento do registro
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao salvar G_TB_DOCUMENTOTIPO: {e}"
)

View file

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

View file

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

View file

@ -0,0 +1,110 @@
from pydantic import BaseModel, field_validator, model_validator
from fastapi import HTTPException, status
from typing import Optional
# Funções para sanitização de entradas (evitar XSS, SQLi etc.)
from actions.validations.text import Text
# ----------------------------------------------------
# Schema base
# ----------------------------------------------------
class GNaturezaSchema(BaseModel):
natureza_id: Optional[int] = None
descricao: Optional[str] = None
situacao: Optional[str] = None
sistema_id: Optional[int] = None
pedir_numero_imovel: Optional[str] = None
controle_frenteverso: Optional[str] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar uma NATUREZA especifica pelo ID (GET)
# ----------------------------------------------------
class GNaturezaIdSchema(BaseModel):
natureza_id: int
# ----------------------------------------------------
# Schema para localizar uma NATUREZA especifica pela descrição (GET)
# ----------------------------------------------------
class GNaturezaDescricaoSchema(BaseModel):
descricao: str
# ----------------------------------------------------
# Schema para criação de nova NATUREZA (POST)
# ----------------------------------------------------
class GNaturezaSaveSchema(BaseModel):
natureza_id: Optional[int] = None
descricao: str
situacao: str
sistema_id: Optional[int] = None
pedir_numero_imovel: Optional[str] = None
controle_frenteverso: Optional[str] = None
# Sanitiza os inputs enviados
@field_validator('descricao', 'situacao', 'pedir_numero_imovel', 'controle_frenteverso')
def sanitize_fields(cls, v):
if v:
return Text.sanitize_input(v)
return v
# Verifica se os campos obrigatórios foram enviados
@model_validator(mode='after')
def validate_all_fields(self):
errors = []
if not self.descricao or len(self.descricao.strip()) == 0:
errors.append({'input': 'descricao', 'message': 'A descrição é obrigatória.'})
if not self.situacao or len(self.situacao.strip()) == 0:
errors.append({'input': 'situacao', 'message': 'A situação é obrigatória.'})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self
# ----------------------------------------------------
# Schema para atualizar NATUREZA (PUT)
# ----------------------------------------------------
class GNaturezaUpdateSchema(BaseModel):
descricao: Optional[str] = None
situacao: Optional[str] = None
sistema_id: Optional[int] = None
pedir_numero_imovel: Optional[str] = None
controle_frenteverso: Optional[str] = None
# Sanitiza os inputs enviados
@field_validator('descricao', 'situacao', 'pedir_numero_imovel', 'controle_frenteverso')
def sanitize_fields(cls, v):
if v:
return Text.sanitize_input(v)
return v
# Verifica se os campos obrigatórios foram enviados
@model_validator(mode='after')
def validate_all_fields(self):
errors = []
# A validação de campos obrigatórios é diferente para atualização, pois os campos são opcionais.
# A lógica para a validação pode ser mais complexa dependendo das regras de negócio.
# Para este exemplo, não há campos obrigatórios na atualização.
# Se você precisar de validação de campos obrigatórios, pode adicionar aqui.
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self

View file

@ -0,0 +1,148 @@
from pydantic import BaseModel, field_validator, model_validator
from fastapi import HTTPException, status
from typing import Optional
# Funções para sanitização de entradas (evitar XSS, SQLi etc.)
from actions.validations.text import Text
# ----------------------------------------------------
# Schema base
# ----------------------------------------------------
class GTbDocumentoTipoSchema(BaseModel):
tb_documentotipo_id: Optional[int] = None
descricao: Optional[str] = None
texto: Optional[bytes] = None
situacao: Optional[str] = None
possui_numeracao: Optional[str] = None
orgao_padrao: Optional[str] = None
descricao_simplificada: Optional[str] = None
tipo: Optional[str] = None
descricao_sinter: Optional[str] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar um G_TB_DOCUMENTOTIPO especifico pelo ID (GET)
# ----------------------------------------------------
class GTbDocumentoTipoIdSchema(BaseModel):
tb_documentotipo_id: int
# ----------------------------------------------------
# Schema para localizar um G_TB_DOCUMENTOTIPO especifico pela descrição (GET)
# ----------------------------------------------------
class GTbDocumentoTipoDescricaoSchema(BaseModel):
descricao: str
# ----------------------------------------------------
# Schema para criação de novo G_TB_DOCUMENTOTIPO (POST)
# ----------------------------------------------------
class GTbDocumentoTipoSaveSchema(BaseModel):
tb_documentotipo_id: Optional[int] = None
descricao: str
texto: Optional[bytes] = None
situacao: str
possui_numeracao: str
orgao_padrao: str
descricao_simplificada: str
tipo: str
descricao_sinter: str
# Sanitiza os inputs enviados
@field_validator('descricao', 'situacao', 'possui_numeracao', 'orgao_padrao', 'descricao_simplificada', 'tipo', 'descricao_sinter')
def sanitize_fields(cls, v):
if v:
return Text.sanitize_input(v)
return v
# Verifica se os campos obrigatórios foram enviados
@model_validator(mode='after')
def validate_all_fields(self):
errors = []
if not self.descricao or len(self.descricao.strip()) == 0:
errors.append({'input': 'descricao', 'message': 'A descrição é obrigatória.'})
if not self.situacao or len(self.situacao.strip()) == 0:
errors.append({'input': 'situacao', 'message': 'A situação é obrigatória.'})
if not self.possui_numeracao or len(self.possui_numeracao.strip()) == 0:
errors.append({'input': 'possui_numeracao', 'message': 'A informação se possui numeração é obrigatória.'})
if not self.orgao_padrao or len(self.orgao_padrao.strip()) == 0:
errors.append({'input': 'orgao_padrao', 'message': 'O orgão padrão é obrigatório.'})
if not self.descricao_simplificada or len(self.descricao_simplificada.strip()) == 0:
errors.append({'input': 'descricao_simplificada', 'message': 'A descrição simplificada é obrigatória.'})
if not self.tipo or len(self.tipo.strip()) == 0:
errors.append({'input': 'tipo', 'message': 'O tipo é obrigatório.'})
if not self.descricao_sinter or len(self.descricao_sinter.strip()) == 0:
errors.append({'input': 'descricao_sinter', 'message': 'A descrição do SINTER é obrigatória.'})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self
# ----------------------------------------------------
# Schema para atualizar G_TB_DOCUMENTOTIPO (PUT)
# ----------------------------------------------------
class GTbDocumentoTipoUpdateSchema(BaseModel):
descricao: Optional[str] = None
texto: Optional[bytes] = None
situacao: Optional[str] = None
possui_numeracao: Optional[str] = None
orgao_padrao: Optional[str] = None
descricao_simplificada: Optional[str] = None
tipo: Optional[str] = None
descricao_sinter: Optional[str] = None
# Sanitiza os inputs enviados
@field_validator('descricao', 'situacao', 'possui_numeracao', 'orgao_padrao', 'descricao_simplificada', 'tipo', 'descricao_sinter')
def sanitize_fields(cls, v):
if v:
return Text.sanitize_input(v)
return v
# Verifica se os campos obrigatórios foram enviados
@model_validator(mode='after')
def validate_all_fields(self):
errors = []
if self.descricao is not None and len(self.descricao.strip()) == 0:
errors.append({'input': 'descricao', 'message': 'A descrição é obrigatória.'})
if self.situacao is not None and len(self.situacao.strip()) == 0:
errors.append({'input': 'situacao', 'message': 'A situação é obrigatória.'})
if self.possui_numeracao is not None and len(self.possui_numeracao.strip()) == 0:
errors.append({'input': 'possui_numeracao', 'message': 'A informação se possui numeração é obrigatória.'})
if self.orgao_padrao is not None and len(self.orgao_padrao.strip()) == 0:
errors.append({'input': 'orgao_padrao', 'message': 'O orgão padrão é obrigatório.'})
if self.descricao_simplificada is not None and len(self.descricao_simplificada.strip()) == 0:
errors.append({'input': 'descricao_simplificada', 'message': 'A descrição simplificada é obrigatória.'})
if self.tipo is not None and len(self.tipo.strip()) == 0:
errors.append({'input': 'tipo', 'message': 'O tipo é obrigatório.'})
if self.descricao_sinter is not None and len(self.descricao_sinter.strip()) == 0:
errors.append({'input': 'descricao_sinter', 'message': 'A descrição do SINTER é obrigatória.'})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -0,0 +1,23 @@
from packages.v1.administrativo.schemas.g_natureza_schema import GNaturezaUpdateSchema
from packages.v1.administrativo.actions.g_natureza.g_natureza_update_action import UpdateAction
class UpdateService:
"""
Serviço para a operação de atualização de um registro na tabela
g_natureza.
"""
def execute(self, natureza_id : int, natureza_schema: GNaturezaUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
natureza_schema (GNaturezaUpdateSchema): O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# Instanciamento de ações
update_action = UpdateAction()
# Retorna o resultado da operação
return update_action.execute(natureza_id, natureza_schema)

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -0,0 +1,23 @@
from packages.v1.administrativo.schemas.g_tb_documentotipo_schema import GTbDocumentoTipoUpdateSchema
from packages.v1.administrativo.actions.g_tb_documentotipo.g_tb_documentotipo_update_action import UpdateAction
class UpdateService:
"""
Serviço para a operação de atualização de um registro na tabela
G_TB_DOCUMENTOTIPO.
"""
def execute(self, tb_documentotipo_id : int, documentotipo_schema: GTbDocumentoTipoUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
documentotipo_schema (GtbDocumentotipoUpdateSchema): O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# Instanciamento de ações
update_action = UpdateAction()
# Retorna o resultado da operação
return update_action.execute(tb_documentotipo_id, documentotipo_schema)

View file

@ -17,6 +17,8 @@ from packages.v1.administrativo.endpoints import t_censec_qualidade_endpoint
from packages.v1.administrativo.endpoints import g_tb_estadocivil_endpoint
from packages.v1.administrativo.endpoints import g_medida_tipo_endpoint
from packages.v1.administrativo.endpoints import t_minuta_endpoint
from packages.v1.administrativo.endpoints import g_tb_documentotipo_endpoint
from packages.v1.administrativo.endpoints import g_natureza_endpoint
from packages.v1.administrativo.endpoints import g_tb_bairro_endpoint
from packages.v1.administrativo.endpoints import g_tb_tipologradouro_endpoint
@ -101,6 +103,16 @@ api_router.include_router(
t_minuta_endpoint.router, prefix="/administrativo/t_minuta", tags=["Minutas"]
)
# Inclui as rotas de g_tb_documentotipo
api_router.include_router(
g_tb_documentotipo_endpoint.router, prefix="/administrativo/g_tb_documentotipo", tags=["Documento tipo"]
)
# Inclui as rotas de g_natureza
api_router.include_router(
g_natureza_endpoint.router, prefix="/administrativo/g_natureza", tags=["Natureza"]
)
# Inclui as rotas de t_censec_naturezalitigio
api_router.include_router(