[MVPTN-16] feat(CRUD): Gerado CRUD completo para as seguintes tabelas: t_servico_tipo, g_marcacao_tipo, t_servico_etiqueta

This commit is contained in:
Kenio 2025-10-01 17:02:50 -03:00
parent b449227d07
commit da4c68723b
62 changed files with 3556 additions and 0 deletions

View file

@ -0,0 +1,29 @@
from abstracts.action import BaseAction
# O schema TServicoTipoIdSchema deve ser substituído por GMarcacaoTipoIdSchema
# que contém o campo-chave MARCACAO_TIPO_ID.
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoIdSchema
# O repositório TServicoTipoDeleteRepository deve ser substituído pelo GMarcacaoTipoDeleteRepository.
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_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_MARCACAO_TIPO.
"""
def execute(self, marcacao_tipo_schema: GMarcacaoTipoIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
marcacao_tipo_schema (GMarcacaoTipoIdSchema): O esquema com o ID (MARCACAO_TIPO_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(marcacao_tipo_schema)

View file

@ -0,0 +1,31 @@
from abstracts.action import BaseAction
# Ajuste do schema de entrada
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoDescricaoSchema
# Ajuste do repositório
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_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_MARCACAO_TIPO por descrição.
"""
def execute(self, marcacao_tipo_schema: GMarcacaoTipoDescricaoSchema):
"""
Executa a operação de busca no banco de dados.
Args:
marcacao_tipo_schema (GMarcacaoTipoDescricaoSchema): 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(marcacao_tipo_schema)
# Retorno da informação
return response

View file

@ -0,0 +1,25 @@
from abstracts.action import BaseAction
# O repositório TServicoTipoIndexRepository deve ser substituído pelo GMarcacaoTipoIndexRepository.
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_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_MARCACAO_TIPO.
"""
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,30 @@
from abstracts.action import BaseAction
# Ajuste do schema de entrada
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoSaveSchema
# Ajuste do repositório
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_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_MARCACAO_TIPO.
"""
def execute(self, marcacao_tipo_schema: GMarcacaoTipoSaveSchema):
"""
Executa a operação de salvamento.
Args:
marcacao_tipo_schema (GMarcacaoTipoSaveSchema): 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(marcacao_tipo_schema)
# Retorno da informação
return response

View file

@ -0,0 +1,30 @@
from abstracts.action import BaseAction
# Ajuste do schema de entrada
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoIdSchema
# Ajuste do repositório
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_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_MARCACAO_TIPO.
"""
def execute(self, marcacao_tipo_schema: GMarcacaoTipoIdSchema):
"""
Executa a operação de exibição.
Args:
marcacao_tipo_schema (GMarcacaoTipoIdSchema): O esquema com o ID (MARCACAO_TIPO_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(marcacao_tipo_schema)
# Retorno da informação
return response

View file

@ -0,0 +1,28 @@
# Importação do Schema ajustada
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoUpdateSchema
# Importação do Repositório ajustada
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_update_repository import UpdateRepository
# A classe UpdateAction não herda de BaseAction no arquivo original, mantemos o padrão.
class UpdateAction:
"""
Service responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela G_MARCACAO_TIPO.
"""
def execute(self, marcacao_tipo_id: int, marcacao_tipo_schema: GMarcacaoTipoUpdateSchema):
"""
Executa a operação de atualização.
Args:
marcacao_tipo_id (int): O ID (MARCACAO_TIPO_ID) do registro a ser atualizado.
marcacao_tipo_schema (GMarcacaoTipoUpdateSchema): 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(marcacao_tipo_id, marcacao_tipo_schema)

View file

@ -0,0 +1,26 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import TServicoEtiquetaIdSchema
from packages.v1.administrativo.repositories.t_servico_etiqueta.t_servico_etiqueta_delete_repository import DeleteRepository
class DeleteAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela T_SERVICO_ETIQUETA.
"""
def execute(self, servico_etiqueta_schema: TServicoEtiquetaIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
servico_etiqueta_schema (TServicoEtiquetaIdSchema): O esquema com o ID (SERVICO_ETIQUETA_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(servico_etiqueta_schema)

View file

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

View file

@ -0,0 +1,28 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import TServicoEtiquetaSaveSchema
from packages.v1.administrativo.repositories.t_servico_etiqueta.t_servico_etiqueta_save_repository import SaveRepository
class SaveAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvar um novo registro na tabela T_SERVICO_ETIQUETA.
"""
def execute(self, servico_etiqueta_schema: TServicoEtiquetaSaveSchema):
"""
Executa a operação de salvamento.
Args:
servico_etiqueta_schema (TServicoEtiquetaSaveSchema): 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(servico_etiqueta_schema)
# Retorno da informação
return response

View file

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

View file

@ -0,0 +1,26 @@
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import TServicoEtiquetaUpdateSchema
from packages.v1.administrativo.repositories.t_servico_etiqueta.t_servico_etiqueta_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 T_SERVICO_ETIQUETA.
"""
def execute(self, servico_etiqueta_id: int, servico_etiqueta_schema: TServicoEtiquetaUpdateSchema):
"""
Executa a operação de atualização.
Args:
servico_etiqueta_id (int): O ID (SERVICO_ETIQUETA_ID) do registro a ser atualizado.
servico_etiqueta_schema (TServicoEtiquetaUpdateSchema): 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(servico_etiqueta_id, servico_etiqueta_schema)

View file

@ -0,0 +1,26 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_servico_tipo_schema import TServicoTipoIdSchema
from packages.v1.administrativo.repositories.t_servico_tipo.t_servico_tipo_delete_repository import DeleteRepository
class DeleteAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela T_SERVICO_TIPO.
"""
def execute(self, servico_tipo_schema: TServicoTipoIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
servico_tipo_schema (TServicoTipoIdSchema): O esquema com o ID (SERVICO_TIPO_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(servico_tipo_schema)

View file

@ -0,0 +1,29 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_servico_tipo_schema import TServicoTipoDescricaoSchema
from packages.v1.administrativo.repositories.t_servico_tipo.t_servico_tipo_get_by_descricao_repository import GetByDescricaoRepository
class GetByDescricaoAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela T_SERVICO_TIPO por descrição.
"""
def execute(self, servico_tipo_schema: TServicoTipoDescricaoSchema):
"""
Executa a operação de busca no banco de dados.
Args:
servico_tipo_schema (TServicoTipoDescricaoSchema): 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(servico_tipo_schema)
# Retorno da informação
return response

View file

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

View file

@ -0,0 +1,28 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_servico_tipo_schema import TServicoTipoSaveSchema
from packages.v1.administrativo.repositories.t_servico_tipo.t_servico_tipo_save_repository import SaveRepository
class SaveAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvar um novo registro na tabela T_SERVICO_TIPO.
"""
def execute(self, servico_tipo_schema: TServicoTipoSaveSchema):
"""
Executa a operação de salvamento.
Args:
servico_tipo_schema (TServicoTipoSaveSchema): 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(servico_tipo_schema)
# Retorno da informação
return response

View file

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

View file

@ -0,0 +1,26 @@
from packages.v1.administrativo.schemas.t_servico_tipo_schema import TServicoTipoUpdateSchema
from packages.v1.administrativo.repositories.t_servico_tipo.t_servico_tipo_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 T_SERVICO_TIPO.
"""
def execute(self, servico_tipo_id: int, servico_tipo_schema: TServicoTipoUpdateSchema):
"""
Executa a operação de atualização.
Args:
servico_tipo_id (int): O ID (SERVICO_TIPO_ID) do registro a ser atualizado.
servico_tipo_schema (TServicoTipoUpdateSchema): 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(servico_tipo_id, servico_tipo_schema)

View file

@ -0,0 +1,113 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import (
GMarcacaoTipoSchema,
GMarcacaoTipoSaveSchema,
GMarcacaoTipoUpdateSchema,
GMarcacaoTipoIdSchema,
GMarcacaoTipoDescricaoSchema
)
class GMarcacaoTipoController:
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_marcacao_tipo")
pass
# Lista todos os registros de marcacao tipo
def index(self):
# Importação da classe desejada
indexService = self.dynamic_import.service("g_marcacao_tipo_index_service", "IndexService")
# Instância da classe service
self.indexService = indexService()
# Lista todos os registros de marcacao tipo
return {
'message': 'Registros de marcação tipo localizados com sucesso',
'data': self.indexService.execute()
}
# Busca um registro de marcacao tipo específico pelo ID
def show(self, marcacao_tipo_schema: GMarcacaoTipoIdSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('g_marcacao_tipo_show_service', 'ShowService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de marcacao tipo desejado
return {
'message': 'Registro de marcação tipo localizado com sucesso',
'data': self.show_service.execute(marcacao_tipo_schema)
}
# Busca um registro de marcacao tipo pela descrição
def get_by_descricao(self, marcacao_tipo_schema: GMarcacaoTipoDescricaoSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('g_marcacao_tipo_get_by_descricao_service', 'GetByDescricaoService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de marcacao tipo desejado
return {
'message': 'Registro de marcação tipo localizado com sucesso',
'data': self.show_service.execute(marcacao_tipo_schema, True)
}
# Cadastra um novo registro de marcacao tipo
def save(self, marcacao_tipo_schema: GMarcacaoTipoSaveSchema):
#Importação da classe desejada
save_service = self.dynamic_import.service('g_marcacao_tipo_save_service', 'SaveService')
# Instância da classe desejada
self.save_service = save_service()
# Busca e retorna o registro de marcacao tipo desejado
return {
'message': 'Registro de marcação tipo salvo com sucesso',
'data': self.save_service.execute(marcacao_tipo_schema)
}
# Atualiza os dados de um registro de marcacao tipo
def update(self, marcacao_tipo_id: int, marcacao_tipo_schema: GMarcacaoTipoUpdateSchema):
#Importação da classe desejada
update_service = self.dynamic_import.service('g_marcacao_tipo_update_service', 'UpdateService')
# Instância da classe desejada
self.update_service = update_service()
# Busca e retorna o registro de marcacao tipo desejado
return {
'message': 'Registro de marcação tipo atualizado com sucesso',
'data': self.update_service.execute(marcacao_tipo_id, marcacao_tipo_schema)
}
# Exclui um registro de marcacao tipo
def delete(self, marcacao_tipo_schema: GMarcacaoTipoIdSchema):
#Importação da classe desejada
delete_service = self.dynamic_import.service('g_marcacao_tipo_delete_service', 'DeleteService')
# Instância da classe desejada
self.delete_service = delete_service()
# Busca e retorna o registro de marcacao tipo desejado
return {
'message': 'Registro de marcação tipo removido com sucesso',
'data': self.delete_service.execute(marcacao_tipo_schema)
}

View file

@ -0,0 +1,96 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import (
TServicoEtiquetaSchema,
TServicoEtiquetaSaveSchema,
TServicoEtiquetaUpdateSchema,
TServicoEtiquetaIdSchema,
TServicoEtiquetaServicoTipoIdSchema
)
class TServicoEtiquetaController:
def __init__(self):
# Action responsável por carregar as services de acordo com o estado
self.dynamic_import = DynamicImport()
# Define o pacote que deve ser carregado
self.dynamic_import.set_package("administrativo")
# Define a tabela que o pacote pertence
self.dynamic_import.set_table("t_servico_etiqueta")
pass
# Lista todos os registros de servico etiqueta
def index(self):
# Importação da classe desejada
index_service = self.dynamic_import.service("t_servico_etiqueta_index_service", "IndexService")
# Instância da classe service
self.index_service = index_service()
# Lista todos os registros de servico etiqueta
return {
'message': 'Registros de servico etiqueta localizados com sucesso',
'data': self.index_service.execute()
}
# Busca um registro de servico etiqueta específico pelo ID
def show(self, servico_etiqueta_schema: TServicoEtiquetaIdSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('t_servico_etiqueta_show_service', 'ShowService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de servico etiqueta desejado
return {
'message': 'Registro de servico etiqueta localizado com sucesso',
'data': self.show_service.execute(servico_etiqueta_schema)
}
# Cadastra um novo registro de servico etiqueta
def save(self, servico_etiqueta_schema: TServicoEtiquetaSaveSchema):
#Importação da classe desejada
save_service = self.dynamic_import.service('t_servico_etiqueta_save_service', 'SaveService')
# Instância da classe desejada
self.save_service = save_service()
# Busca e retorna o registro de servico etiqueta desejado
return {
'message': 'Registro de servico etiqueta salvo com sucesso',
'data': self.save_service.execute(servico_etiqueta_schema)
}
# Atualiza os dados de um registro de servico etiqueta
def update(self, servico_etiqueta_id: int, servico_etiqueta_schema: TServicoEtiquetaUpdateSchema):
#Importação da classe desejada
update_service = self.dynamic_import.service('t_servico_etiqueta_update_service', 'UpdateService')
# Instância da classe desejada
self.update_service = update_service()
# Busca e retorna o registro de servico etiqueta desejado
return {
'message': 'Registro de servico etiqueta atualizado com sucesso',
'data': self.update_service.execute(servico_etiqueta_id, servico_etiqueta_schema)
}
# Exclui um registro de servico etiqueta
def delete(self, servico_etiqueta_schema: TServicoEtiquetaIdSchema):
#Importação da classe desejada
delete_service = self.dynamic_import.service('t_servico_etiqueta_delete_service', 'DeleteService')
# Instância da classe desejada
self.delete_service = delete_service()
# Busca e retorna o registro de servico etiqueta desejado
return {
'message': 'Registro de servico etiqueta removido com sucesso',
'data': self.delete_service.execute(servico_etiqueta_schema)
}

View file

@ -0,0 +1,113 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.t_servico_tipo_schema import (
TServicoTipoSchema,
TServicoTipoSaveSchema,
TServicoTipoUpdateSchema,
TServicoTipoIdSchema,
TServicoTipoDescricaoSchema
)
class TServicoTipoController:
def __init__(self):
# Action responsável por carregar as services de acordo com o estado
self.dynamic_import = DynamicImport()
# Define o pacote que deve ser carregado
self.dynamic_import.set_package("administrativo")
# Define a tabela que o pacote pertence
self.dynamic_import.set_table("t_servico_tipo")
pass
# Lista todos os registros de servico tipo
def index(self):
# Importação da classe desejada
indexService = self.dynamic_import.service("t_servico_tipo_index_service", "IndexService")
# Instância da classe service
self.indexService = indexService()
# Lista todos os registros de servico tipo
return {
'message': 'Registros de servico tipo localizados com sucesso',
'data': self.indexService.execute()
}
# Busca um registro de servico tipo específico pelo ID
def show(self, servico_tipo_schema: TServicoTipoIdSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('t_servico_tipo_show_service', 'ShowService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de servico tipo desejado
return {
'message': 'Registro de servico tipo localizado com sucesso',
'data': self.show_service.execute(servico_tipo_schema)
}
# Busca um registro de servico tipo pela descrição
def get_by_descricao(self, servico_tipo_schema: TServicoTipoDescricaoSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('t_servico_tipo_get_by_descricao_service', 'GetByDescricaoService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de servico tipo desejado
return {
'message': 'Registro de servico tipo localizado com sucesso',
'data': self.show_service.execute(servico_tipo_schema, True)
}
# Cadastra um novo registro de servico tipo
def save(self, servico_tipo_schema: TServicoTipoSaveSchema):
#Importação da classe desejada
save_service = self.dynamic_import.service('t_servico_tipo_save_service', 'SaveService')
# Instância da classe desejada
self.save_service = save_service()
# Busca e retorna o registro de servico tipo desejado
return {
'message': 'Registro de servico tipo salvo com sucesso',
'data': self.save_service.execute(servico_tipo_schema)
}
# Atualiza os dados de um registro de servico tipo
def update(self, servico_tipo_id: int, servico_tipo_schema: TServicoTipoUpdateSchema):
#Importação da classe desejada
update_service = self.dynamic_import.service('t_servico_tipo_update_service', 'UpdateService')
# Instância da classe desejada
self.update_service = update_service()
# Busca e retorna o registro de servico tipo desejado
return {
'message': 'Registro de servico tipo atualizado com sucesso',
'data': self.update_service.execute(servico_tipo_id, servico_tipo_schema)
}
# Exclui um registro de servico tipo
def delete(self, servico_tipo_schema: TServicoTipoIdSchema):
#Importação da classe desejada
delete_service = self.dynamic_import.service('t_servico_tipo_delete_service', 'DeleteService')
# Instância da classe desejada
self.delete_service = delete_service()
# Busca e retorna o registro de servico tipo desejado
return {
'message': 'Registro de servico tipo removido com sucesso',
'data': self.delete_service.execute(servico_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_marcacao_tipo_controller import GMarcacaoTipoController
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import (
GMarcacaoTipoSchema,
GMarcacaoTipoSaveSchema,
GMarcacaoTipoUpdateSchema,
GMarcacaoTipoIdSchema
)
# Inicializa o roteador para as rotas do tipo de marcação
router = APIRouter()
# Instanciamento do controller desejado
g_marcacao_tipo_controller = GMarcacaoTipoController()
# Lista todos os registros de marcação tipo
@router.get('/',
status_code=status.HTTP_200_OK,
summary='Lista todos os registros de marcação tipo cadastrados',
response_description='Lista todos os registros de marcação tipo cadastrados')
async def index(current_user: dict = Depends(get_current_user)):
# Busca todos os registros de marcação tipo cadastrados
response = g_marcacao_tipo_controller.index()
# Retorna os dados localizados
return response
# Localiza um registro de marcação tipo pela descrição
@router.get('/descricao',
status_code=status.HTTP_200_OK,
summary='Busca um registro de marcação tipo em específico pela descrição',
response_description='Busca um registro de marcação tipo em específico')
async def get_by_descricao(descricao : str, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos. Nota: Assumindo que GMarcacaoTipoSchema pode ser usado para descrição.
marcacao_tipo_schema = GMarcacaoTipoSchema(descricao=descricao)
# Busca um registro de marcação tipo específico pela descrição
response = g_marcacao_tipo_controller.get_by_descricao(marcacao_tipo_schema)
# Retorna os dados localizados
return response
# Localiza um registro de marcação tipo pelo ID
@router.get('/{marcacao_tipo_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de marcação tipo em específico pelo ID',
response_description='Busca um registro de marcação tipo em específico')
async def show(marcacao_tipo_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
marcacao_tipo_schema = GMarcacaoTipoIdSchema(marcacao_tipo_id=marcacao_tipo_id)
# Busca um registro de marcação tipo específico pelo ID
response = g_marcacao_tipo_controller.show(marcacao_tipo_schema)
# Retorna os dados localizados
return response
# Cadastro de registro de marcação tipo
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um registro de marcação tipo',
response_description='Cadastra um registro de marcação tipo')
async def save(marcacao_tipo_schema: GMarcacaoTipoSaveSchema, current_user: dict = Depends(get_current_user)):
# Efetua o cadastro no banco de dados
response = g_marcacao_tipo_controller.save(marcacao_tipo_schema)
# Retorna os dados localizados
return response
# Atualiza os dados de um registro de marcação tipo
@router.put('/{marcacao_tipo_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza um registro de marcação tipo',
response_description='Atualiza um registro de marcação tipo')
async def update(marcacao_tipo_id: int, marcacao_tipo_schema: GMarcacaoTipoUpdateSchema, current_user: dict = Depends(get_current_user)):
# Efetua a atualização dos dados
response = g_marcacao_tipo_controller.update(marcacao_tipo_id, marcacao_tipo_schema)
# Retorna os dados localizados
return response
# Exclui um determinado registro de marcação tipo
@router.delete('/{marcacao_tipo_id}',
status_code=status.HTTP_200_OK,
summary='Remove um registro de marcação tipo',
response_description='Remove um registro de marcação tipo')
async def delete(marcacao_tipo_id: int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
marcacao_tipo_schema = GMarcacaoTipoIdSchema(marcacao_tipo_id=marcacao_tipo_id)
# Efetua a exclusão do registro de marcação tipo
response = g_marcacao_tipo_controller.delete(marcacao_tipo_schema)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,91 @@
# Importação de bibliotecas
from typing import Optional
from fastapi import APIRouter, Body, Depends, status
from actions.jwt.get_current_user import get_current_user
from packages.v1.administrativo.controllers.t_servico_etiqueta_controller import TServicoEtiquetaController
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import (
TServicoEtiquetaSaveSchema,
TServicoEtiquetaUpdateSchema,
TServicoEtiquetaIdSchema,
TServicoEtiquetaServicoTipoIdSchema
)
# Inicializa o roteador para as rotas de etiqueta de serviço
router = APIRouter()
# Instanciamento do controller desejado
t_servico_etiqueta_controller = TServicoEtiquetaController()
# Lista todos os registros de servico etiqueta
@router.get('/',
status_code=status.HTTP_200_OK,
summary='Lista todos os registros de servico etiqueta cadastrados',
response_description='Lista todos os registros de servico etiqueta cadastrados')
async def index(current_user: dict = Depends(get_current_user)):
# Busca todos os registros de servico etiqueta cadastrados
response = t_servico_etiqueta_controller.index()
# Retorna os dados localizados
return response
# Localiza um registro de servico etiqueta pelo ID
@router.get('/{servico_etiqueta_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de servico etiqueta em específico pelo ID',
response_description='Busca um registro de servico etiqueta em específico')
async def show(servico_etiqueta_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
servico_etiqueta_schema = TServicoEtiquetaIdSchema(servico_etiqueta_id=servico_etiqueta_id)
# Busca um registro de servico etiqueta específico pelo ID
response = t_servico_etiqueta_controller.show(servico_etiqueta_schema)
# Retorna os dados localizados
return response
# Cadastro de registro de servico etiqueta
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um registro de servico etiqueta',
response_description='Cadastra um registro de servico etiqueta')
async def save(servico_etiqueta_schema: TServicoEtiquetaSaveSchema, current_user: dict = Depends(get_current_user)):
# Efetua o cadastro no banco de dados
response = t_servico_etiqueta_controller.save(servico_etiqueta_schema)
# Retorna os dados localizados
return response
# Atualiza os dados de um registro de servico etiqueta
@router.put('/{servico_etiqueta_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza um registro de servico etiqueta',
response_description='Atualiza um registro de servico etiqueta')
async def update(servico_etiqueta_id: int, servico_etiqueta_schema: TServicoEtiquetaUpdateSchema, current_user: dict = Depends(get_current_user)):
# Efetua a atualização dos dados
response = t_servico_etiqueta_controller.update(servico_etiqueta_id, servico_etiqueta_schema)
# Retorna os dados localizados
return response
# Exclui um determinado registro de servico etiqueta
@router.delete('/{servico_etiqueta_id}',
status_code=status.HTTP_200_OK,
summary='Remove um registro de servico etiqueta',
response_description='Remove um registro de servico etiqueta')
async def delete(servico_etiqueta_id: int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
servico_etiqueta_schema = TServicoEtiquetaIdSchema(servico_etiqueta_id=servico_etiqueta_id)
# Efetua a exclusão do registro de servico etiqueta
response = t_servico_etiqueta_controller.delete(servico_etiqueta_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.t_servico_tipo_controller import TServicoTipoController
from packages.v1.administrativo.schemas.t_servico_tipo_schema import (
TServicoTipoSchema,
TServicoTipoSaveSchema,
TServicoTipoUpdateSchema,
TServicoTipoIdSchema
)
# Inicializa o roteador para as rotas do tipo de serviço
router = APIRouter()
# Instanciamento do controller desejado
t_servico_tipo_controller = TServicoTipoController()
# Lista todos os registros de servico tipo
@router.get('/',
status_code=status.HTTP_200_OK,
summary='Lista todos os registros de servico tipo cadastrados',
response_description='Lista todos os registros de servico tipo cadastrados')
async def index(current_user: dict = Depends(get_current_user)):
# Busca todos os registros de servico tipo cadastrados
response = t_servico_tipo_controller.index()
# Retorna os dados localizados
return response
# Localiza um registro de servico tipo pela descrição
@router.get('/descricao',
status_code=status.HTTP_200_OK,
summary='Busca um registro de servico tipo em específico pela descrição',
response_description='Busca um registro de servico tipo em específico')
async def get_by_descricao(descricao : str, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos. Nota: Assumindo que TServicoTipoSchema pode ser usado para descrição na rota get.
servico_tipo_schema = TServicoTipoSchema(descricao=descricao)
# Busca um registro de servico tipo específico pela descrição
response = t_servico_tipo_controller.get_by_descricao(servico_tipo_schema)
# Retorna os dados localizados
return response
# Localiza um registro de servico tipo pelo ID
@router.get('/{servico_tipo_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de servico tipo em específico pelo ID',
response_description='Busca um registro de servico tipo em específico')
async def show(servico_tipo_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
servico_tipo_schema = TServicoTipoIdSchema(servico_tipo_id=servico_tipo_id)
# Busca um registro de servico tipo específico pelo ID
response = t_servico_tipo_controller.show(servico_tipo_schema)
# Retorna os dados localizados
return response
# Cadastro de registro de servico tipo
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um registro de servico tipo',
response_description='Cadastra um registro de servico tipo')
async def save(servico_tipo_schema: TServicoTipoSaveSchema, current_user: dict = Depends(get_current_user)):
# Efetua o cadastro no banco de dados
response = t_servico_tipo_controller.save(servico_tipo_schema)
# Retorna os dados localizados
return response
# Atualiza os dados de um registro de servico tipo
@router.put('/{servico_tipo_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza um registro de servico tipo',
response_description='Atualiza um registro de servico tipo')
async def update(servico_tipo_id: int, servico_tipo_schema: TServicoTipoUpdateSchema, current_user: dict = Depends(get_current_user)):
# Efetua a atualização dos dados
response = t_servico_tipo_controller.update(servico_tipo_id, servico_tipo_schema)
# Retorna os dados localizados
return response
# Exclui um determinado registro de servico tipo
@router.delete('/{servico_tipo_id}',
status_code=status.HTTP_200_OK,
summary='Remove um registro de servico tipo',
response_description='Remove um registro de servico tipo')
async def delete(servico_tipo_id: int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
servico_tipo_schema = TServicoTipoIdSchema(servico_tipo_id=servico_tipo_id)
# Efetua a exclusão do registro de servico tipo
response = t_servico_tipo_controller.delete(servico_tipo_schema)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,28 @@
from abstracts.action import BaseAction
# O schema TServicoTipoIdSchema deve ser substituído por GMarcacaoTipoIdSchema
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoIdSchema
# O repositório TServicoTipoDeleteRepository deve ser substituído pelo GMarcacaoTipoDeleteRepository.
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_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_MARCACAO_TIPO.
"""
def execute(self, marcacao_tipo_schema: GMarcacaoTipoIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
marcacao_tipo_schema (GMarcacaoTipoIdSchema): O esquema com o ID (MARCACAO_TIPO_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(marcacao_tipo_schema)

View file

@ -0,0 +1,42 @@
from abstracts.repository import BaseRepository
# O schema TServicoTipoIdSchema deve ser substituído por GMarcacaoTipoIdSchema.
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoIdSchema
from fastapi import HTTPException, status
class DeleteRepository(BaseRepository):
"""
Repositório para a operação de exclusão de um registro na tabela
G_MARCACAO_TIPO.
"""
def execute(self, marcacao_tipo_schema: GMarcacaoTipoIdSchema):
"""
Executa a consulta SQL para remover um registro pelo ID.
Args:
marcacao_tipo_schema (GMarcacaoTipoIdSchema): O esquema com o ID (MARCACAO_TIPO_ID) a ser removido.
Returns:
O resultado da operação de exclusão.
"""
try:
# Montagem do sql
sql = """ DELETE FROM G_MARCACAO_TIPO WHERE MARCACAO_TIPO_ID = :marcacao_tipo_id """
# Preenchimento de parâmetros
params = {
"marcacao_tipo_id": marcacao_tipo_schema.marcacao_tipo_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 G_MARCACAO_TIPO: {e}"
)

View file

@ -0,0 +1,47 @@
from abstracts.repository import BaseRepository
# O schema de entrada deve ser o de descrição para a tabela G_MARCACAO_TIPO
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoDescricaoSchema
class GetByDescricaoRepository(BaseRepository):
"""
Repositório para a operação de busca de um registro na tabela
G_MARCACAO_TIPO por descrição.
"""
def execute(self, marcacao_tipo_schema: GMarcacaoTipoDescricaoSchema):
"""
Executa a consulta SQL para buscar um registro pela descrição.
Args:
marcacao_tipo_schema (GMarcacaoTipoDescricaoSchema): 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 MARCACAO_TIPO_ID,
DESCRICAO,
NOME,
GRUPO,
SITUACAO,
SISTEMA_ID,
GRUPO_TIPO,
TIPO_QUALIFICACAO,
CONDICAO_SQL,
SEPARADOR_1,
SEPARADOR_2,
SEPARADOR_3,
TIPO_VALOR,
ATUALIZAR,
PROTEGIDA,
ATIVAR_SEPARADOR,
SQL_COMPLETO
FROM G_MARCACAO_TIPO WHERE DESCRICAO = :descricao """
# Preenchimento de parâmetros
params = {
'descricao': marcacao_tipo_schema.descricao
}
# Execução do sql
return self.fetch_one(sql, params)

View file

@ -0,0 +1,41 @@
from abstracts.repository import BaseRepository
class IndexRepository(BaseRepository):
"""
Repositório para a operação de listagem de todos os registros
na tabela G_MARCACAO_TIPO.
"""
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
# Tabela ajustada para G_MARCACAO_TIPO
sql = """ SELECT MARCACAO_TIPO_ID,
DESCRICAO,
NOME,
GRUPO,
SITUACAO,
SISTEMA_ID,
GRUPO_TIPO,
TIPO_QUALIFICACAO,
CONDICAO_SQL,
SEPARADOR_1,
SEPARADOR_2,
SEPARADOR_3,
TIPO_VALOR,
ATUALIZAR,
PROTEGIDA,
ATIVAR_SEPARADOR,
SQL_COMPLETO
FROM G_MARCACAO_TIPO """
# Execução do sql
response = self.fetch_all(sql)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,96 @@
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
# Assumindo que o novo schema de save está no mesmo padrão de importação
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoSaveSchema
class SaveRepository(BaseRepository):
"""
Repositório para a operação de salvamento de um novo registro na tabela G_MARCACAO_TIPO.
"""
def execute(self, marcacao_tipo_schema: GMarcacaoTipoSaveSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
marcacao_tipo_schema (GMarcacaoTipoSaveSchema): 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_MARCACAO_TIPO(
MARCACAO_TIPO_ID,
DESCRICAO,
NOME,
GRUPO,
SITUACAO,
SISTEMA_ID,
GRUPO_TIPO,
TIPO_QUALIFICACAO,
CONDICAO_SQL,
SEPARADOR_1,
SEPARADOR_2,
SEPARADOR_3,
TIPO_VALOR,
ATUALIZAR,
PROTEGIDA,
ATIVAR_SEPARADOR,
SQL_COMPLETO
) VALUES (
:marcacao_tipo_id,
:descricao,
:nome,
:grupo,
:situacao,
:sistema_id,
:grupo_tipo,
:tipo_qualificacao,
:condicao_sql,
:separador_1,
:separador_2,
:separador_3,
:tipo_valor,
:atualizar,
:protegida,
:ativar_separador,
:sql_completo
) RETURNING *;"""
# Preenchimento de parâmetros
params = {
'marcacao_tipo_id': marcacao_tipo_schema.marcacao_tipo_id,
'descricao': marcacao_tipo_schema.descricao,
'nome': marcacao_tipo_schema.nome,
'grupo': marcacao_tipo_schema.grupo,
'situacao': marcacao_tipo_schema.situacao,
'sistema_id': marcacao_tipo_schema.sistema_id,
'grupo_tipo': marcacao_tipo_schema.grupo_tipo,
'tipo_qualificacao': marcacao_tipo_schema.tipo_qualificacao,
'condicao_sql': marcacao_tipo_schema.condicao_sql,
'separador_1': marcacao_tipo_schema.separador_1,
'separador_2': marcacao_tipo_schema.separador_2,
'separador_3': marcacao_tipo_schema.separador_3,
'tipo_valor': marcacao_tipo_schema.tipo_valor,
'atualizar': marcacao_tipo_schema.atualizar,
'protegida': marcacao_tipo_schema.protegida,
'ativar_separador': marcacao_tipo_schema.ativar_separador,
'sql_completo': marcacao_tipo_schema.sql_completo,
}
# 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 MARCAÇÃO TIPO: {e}"
)

View file

@ -0,0 +1,47 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoIdSchema
from fastapi import HTTPException, status
class ShowRepository(BaseRepository):
"""
Repositório para a operação de exibição de um registro na tabela G_MARCACAO_TIPO.
"""
def execute(self, marcacao_tipo_schema: GMarcacaoTipoIdSchema):
"""
Busca um registro específico de G_MARCACAO_TIPO pelo ID.
Args:
marcacao_tipo_schema (GMarcacaoTipoIdSchema): O esquema que contém o ID (MARCACAO_TIPO_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_MARCACAO_TIPO WHERE MARCACAO_TIPO_ID = :marcacao_tipo_id"
# Preenchimento de parâmetros
params = {
'marcacao_tipo_id': marcacao_tipo_schema.marcacao_tipo_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 de G_MARCACAO_TIPO não encontrado"
)
return result
except Exception as e:
# Tratamento de erro
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao buscar o registro de G_MARCACAO_TIPO: {e}"
)

View file

@ -0,0 +1,128 @@
from abstracts.repository import BaseRepository
# O schema TServicoTipoUpdateSchema deve ser substituído por GMarcacaoTipoUpdateSchema
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoUpdateSchema
from fastapi import HTTPException, status
class UpdateRepository(BaseRepository):
"""
Repositório para a operação de atualização na tabela G_MARCACAO_TIPO.
"""
def execute(self, marcacao_tipo_id : int, marcacao_tipo_schema: GMarcacaoTipoUpdateSchema):
"""
Executa a atualização de um registro na tabela.
Args:
marcacao_tipo_id (int): O ID do registro a ser atualizado.
marcacao_tipo_schema (GMarcacaoTipoUpdateSchema): 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 = {}
# --- Campos da G_MARCACAO_TIPO que podem ser atualizados (com base na DDL) ---
if marcacao_tipo_schema.descricao is not None:
updates.append("DESCRICAO = :descricao")
params["descricao"] = marcacao_tipo_schema.descricao
if marcacao_tipo_schema.texto is not None:
updates.append("TEXTO = :texto")
params["texto"] = marcacao_tipo_schema.texto
if marcacao_tipo_schema.nome is not None:
updates.append("NOME = :nome")
params["nome"] = marcacao_tipo_schema.nome
if marcacao_tipo_schema.grupo is not None:
updates.append("GRUPO = :grupo")
params["grupo"] = marcacao_tipo_schema.grupo
if marcacao_tipo_schema.situacao is not None:
updates.append("SITUACAO = :situacao")
params["situacao"] = marcacao_tipo_schema.situacao
if marcacao_tipo_schema.sistema_id is not None:
updates.append("SISTEMA_ID = :sistema_id")
params["sistema_id"] = marcacao_tipo_schema.sistema_id
if marcacao_tipo_schema.grupo_tipo is not None:
updates.append("GRUPO_TIPO = :grupo_tipo")
params["grupo_tipo"] = marcacao_tipo_schema.grupo_tipo
if marcacao_tipo_schema.tipo_qualificacao is not None:
updates.append("TIPO_QUALIFICACAO = :tipo_qualificacao")
params["tipo_qualificacao"] = marcacao_tipo_schema.tipo_qualificacao
if marcacao_tipo_schema.condicao_sql is not None:
updates.append("CONDICAO_SQL = :condicao_sql")
params["condicao_sql"] = marcacao_tipo_schema.condicao_sql
if marcacao_tipo_schema.separador_1 is not None:
updates.append("SEPARADOR_1 = :separador_1")
params["separador_1"] = marcacao_tipo_schema.separador_1
if marcacao_tipo_schema.separador_2 is not None:
updates.append("SEPARADOR_2 = :separador_2")
params["separador_2"] = marcacao_tipo_schema.separador_2
if marcacao_tipo_schema.separador_3 is not None:
updates.append("SEPARADOR_3 = :separador_3")
params["separador_3"] = marcacao_tipo_schema.separador_3
if marcacao_tipo_schema.tipo_valor is not None:
updates.append("TIPO_VALOR = :tipo_valor")
params["tipo_valor"] = marcacao_tipo_schema.tipo_valor
if marcacao_tipo_schema.atualizar is not None:
updates.append("ATUALIZAR = :atualizar")
params["atualizar"] = marcacao_tipo_schema.atualizar
if marcacao_tipo_schema.protegida is not None:
updates.append("PROTEGIDA = :protegida")
params["protegida"] = marcacao_tipo_schema.protegida
if marcacao_tipo_schema.ativar_separador is not None:
updates.append("ATIVAR_SEPARADOR = :ativar_separador")
params["ativar_separador"] = marcacao_tipo_schema.ativar_separador
if marcacao_tipo_schema.sql_completo is not None:
updates.append("SQL_COMPLETO = :sql_completo")
params["sql_completo"] = marcacao_tipo_schema.sql_completo
# -------------------------------------------------------------------------
if not updates:
return False
params["marcacao_tipo_id"] = marcacao_tipo_id
sql = f"UPDATE G_MARCACAO_TIPO SET {', '.join(updates)} WHERE MARCACAO_TIPO_ID = :marcacao_tipo_id RETURNING *;"
# Executa a query
result = self.run_and_return(sql, params)
# O result é uma tupla/objeto com os campos do registro,
# verificamos a existência do campo chave para confirmação
if not result or not getattr(result, 'marcacao_tipo_id', None):
# Informa que não existe o registro a ser modificado
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail='Nenhum MARCAÇÃO TIPO localizado para esta solicitação'
)
# Se houver um resultado e ele tiver o ID, 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 MARCAÇÃO TIPO: {e}"
)

View file

@ -0,0 +1,41 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import TServicoEtiquetaIdSchema
from fastapi import HTTPException, status
class DeleteRepository(BaseRepository):
"""
Repositório para a operação de exclusão de um registro na tabela
T_SERVICO_ETIQUETA.
"""
def execute(self, servico_etiqueta_schema: TServicoEtiquetaIdSchema):
"""
Executa a consulta SQL para remover um registro pelo ID.
Args:
servico_etiqueta_schema (TServicoEtiquetaIdSchema): O esquema com o ID (SERVICO_ETIQUETA_ID) a ser removido.
Returns:
O resultado da operação de exclusão.
"""
try:
# Montagem do sql
sql = """ DELETE FROM T_SERVICO_ETIQUETA WHERE SERVICO_ETIQUETA_ID = :servico_etiqueta_id """
# Preenchimento de parâmetros
params = {
"servico_etiqueta_id": servico_etiqueta_schema.servico_etiqueta_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 T_SERVICO_ETIQUETA: {e}"
)

View file

@ -0,0 +1,30 @@
from abstracts.repository import BaseRepository
class IndexRepository(BaseRepository):
"""
Repositório para a operação de listagem de todos os registros
na tabela T_SERVICO_ETIQUETA.
"""
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
# Tabela e colunas ajustadas para T_SERVICO_ETIQUETA
sql = """
SELECT
SERVICO_ETIQUETA_ID,
ETIQUETA_MODELO_ID,
SERVICO_TIPO_ID
FROM T_SERVICO_ETIQUETA
"""
# Execução do sql
response = self.fetch_all(sql)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,47 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import TServicoEtiquetaSaveSchema
from fastapi import HTTPException, status
class SaveRepository(BaseRepository):
"""
Repositório para a operação de salvamento de um novo registro na tabela
T_SERVICO_ETIQUETA.
"""
def execute(self, servico_etiqueta_schema: TServicoEtiquetaSaveSchema):
"""
Executa a consulta SQL para inserir um novo registro.
Args:
servico_etiqueta_schema (TServicoEtiquetaSaveSchema): O esquema com os dados a serem salvos.
Returns:
O resultado da operação de salvamento, geralmente o ID do novo registro ou um status de sucesso.
"""
try:
# Montagem do sql. SERVICO_ETIQUETA_ID é a chave primária e presumivelmente auto-gerada.
sql = """
INSERT INTO T_SERVICO_ETIQUETA (SERVICO_ETIQUETA_ID, ETIQUETA_MODELO_ID, SERVICO_TIPO_ID)
VALUES (:servico_etiqueta_id, :etiqueta_modelo_id, :servico_tipo_id) RETURNING *;
"""
# Preenchimento de parâmetros
params = {
"servico_etiqueta_id": servico_etiqueta_schema.servico_etiqueta_id,
"etiqueta_modelo_id": servico_etiqueta_schema.etiqueta_modelo_id,
"servico_tipo_id": servico_etiqueta_schema.servico_tipo_id
}
# Execução do sql
# Assumimos que o self.run lida com a execução do INSERT.
response = self.run_and_return(sql, params)
# Retorna o resultado
return response
except Exception as e:
# Informa que houve uma falha no salvamento
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao salvar T_SERVICO_ETIQUETA: {e}"
)

View file

@ -0,0 +1,49 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import TServicoEtiquetaIdSchema # Schema ajustado para T_SERVICO_ETIQUETA
from fastapi import HTTPException, status
class ShowRepository(BaseRepository):
"""
Repositório para a operação de exibição de um registro na tabela T_SERVICO_ETIQUETA.
"""
def execute(self, servico_etiqueta_schema: TServicoEtiquetaIdSchema): # Nome do parâmetro e tipo ajustados
"""
Busca um registro específico de T_SERVICO_ETIQUETA pelo ID.
Args:
servico_etiqueta_schema (TServicoEtiquetaIdSchema): O esquema que contém o ID do registro. # Nome do tipo ajustado
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
# Tabela e coluna da chave primária ajustadas para T_SERVICO_ETIQUETA
sql = "SELECT SERVICO_ETIQUETA_ID, ETIQUETA_MODELO_ID, SERVICO_TIPO_ID FROM T_SERVICO_ETIQUETA WHERE SERVICO_ETIQUETA_ID = :servico_etiqueta_id"
# Preenchimento de parâmetros
params = {
# Nome do parâmetro ajustado para SERVICO_ETIQUETA_ID
'servico_etiqueta_id': servico_etiqueta_schema.servico_etiqueta_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:
# Tratamento de erro ajustado
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Erro ao buscar registro T_SERVICO_ETIQUETA: {str(e)}"
)

View file

@ -0,0 +1,73 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import TServicoEtiquetaUpdateSchema # Schema ajustado para TServicoEtiqueta
from fastapi import HTTPException, status
class UpdateRepository(BaseRepository):
"""
Repositório para a operação de atualização na tabela T_SERVICO_ETIQUETA.
"""
def execute(self, servico_etiqueta_id : int, servico_etiqueta_schema: TServicoEtiquetaUpdateSchema):
"""
Executa a atualização de um registro na tabela.
Args:
servico_etiqueta_id (int): O ID do registro a ser atualizado (SERVICO_ETIQUETA_ID).
servico_etiqueta_schema (TServicoEtiquetaUpdateSchema): 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 = {}
# --- Campos da T_SERVICO_ETIQUETA que podem ser atualizados (com base na DDL) ---
# ETIQUETA_MODELO_ID NUMERIC(10,2)
if servico_etiqueta_schema.etiqueta_modelo_id is not None:
updates.append("ETIQUETA_MODELO_ID = :etiqueta_modelo_id")
params["etiqueta_modelo_id"] = servico_etiqueta_schema.etiqueta_modelo_id
# SERVICO_TIPO_ID NUMERIC(10,2)
if servico_etiqueta_schema.servico_tipo_id is not None:
updates.append("SERVICO_TIPO_ID = :servico_tipo_id")
params["servico_tipo_id"] = servico_etiqueta_schema.servico_tipo_id
# -------------------------------------------------------------------------
if not updates:
# Se não houver campos para atualizar, retorna False (ou um comportamento de sucesso sem alteração)
return False
# Parâmetro da chave primária para a cláusula WHERE
params["servico_etiqueta_id"] = servico_etiqueta_id
# Montagem e execução do SQL
sql = f"UPDATE T_SERVICO_ETIQUETA SET {', '.join(updates)} WHERE SERVICO_ETIQUETA_ID = :servico_etiqueta_id RETURNING *;"
# Executa a query e retorna o registro atualizado
result = self.run_and_return(sql, params)
# O result é uma tupla/objeto com os campos do registro,
# verificamos a existência do campo chave para confirmação
if not result or not getattr(result, 'servico_etiqueta_id', None):
# Informa que não existe o registro a ser modificado
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail='Nenhuma ETIQUETA DE SERVIÇO localizada para esta solicitação'
)
# Se houver um resultado e ele tiver o ID, 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 a ETIQUETA DE SERVIÇO: {e}"
)

View file

@ -0,0 +1,41 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_servico_tipo_schema import TServicoTipoIdSchema
from fastapi import HTTPException, status
class DeleteRepository(BaseRepository):
"""
Repositório para a operação de exclusão de um registro na tabela
T_SERVICO_TIPO.
"""
def execute(self, servico_tipo_schema: TServicoTipoIdSchema):
"""
Executa a consulta SQL para remover um registro pelo ID.
Args:
servico_tipo_schema (TServicoTipoIdSchema): O esquema com o ID (SERVICO_TIPO_ID) a ser removido.
Returns:
O resultado da operação de exclusão.
"""
try:
# Montagem do sql
sql = """ DELETE FROM T_SERVICO_TIPO WHERE SERVICO_TIPO_ID = :servico_tipo_id """
# Preenchimento de parâmetros
params = {
"servico_tipo_id": servico_tipo_schema.servico_tipo_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 T_SERVICO_TIPO: {e}"
)

View file

@ -0,0 +1,31 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_servico_tipo_schema import TServicoTipoDescricaoSchema # Nome do schema ajustado
class GetByDescricaoRepository(BaseRepository):
"""
Repositório para a operação de busca de um registro na tabela
T_SERVICO_TIPO por descrição.
"""
def execute(self, servico_tipo_schema: TServicoTipoDescricaoSchema): # Nome do parâmetro ajustado
"""
Executa a consulta SQL para buscar um registro pela descrição.
Args:
servico_tipo_schema (TServicoTipoDescricaoSchema): O esquema com a descrição a ser buscada. # Nome do tipo ajustado
Returns:
Um dicionário contendo os dados do registro ou None se não for encontrado.
"""
# Montagem do SQL
# Tabela e coluna ajustadas
sql = """ SELECT * FROM T_SERVICO_TIPO WHERE DESCRICAO = :descricao """
# Preenchimento de parâmetros
params = {
# Ajuste para usar o atributo 'descricao' do novo schema
'descricao': servico_tipo_schema.descricao
}
# Execução do sql
return self.fetch_one(sql, params)

View file

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

View file

@ -0,0 +1,150 @@
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
# Assumindo que o novo schema de save está no mesmo padrão de importação do anterior
from packages.v1.administrativo.schemas.t_servico_tipo_schema import TServicoTipoSaveSchema
class SaveRepository(BaseRepository):
"""
Repositório para a operação de salvamento de um novo registro na tabela t_servico_tipo.
"""
def execute(self, servico_tipo_schema: TServicoTipoSaveSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
servico_tipo_schema (TServicoSaveSchema): O esquema com os dados a serem salvos.
Returns:
O registro recém-criado.
Raises:
HTTPException: Caso ocorra um erro na execução da query.
"""
try:
# Montagem do SQL
sql = """ INSERT INTO T_SERVICO_TIPO(
SERVICO_TIPO_ID,
DESCRICAO,
VALOR,
TIPO_ITEM,
REQUER_AUTORIZACAO,
REQUER_BIOMETRIA,
TIPO_PESSOA,
TB_RECONHECIMENTOTIPO_ID,
TIPO_PERMISSAO_CPF,
REQUER_ABONADOR,
REQUER_REPRESENTANTE,
SITUACAO,
REQUER_CPF,
SERVICO_PADRAO,
MAXIMO_PESSOA,
ALTERAR_VALOR,
SERVICO_CAIXA_ID,
LANCAR_TAXA,
LANCAR_FUNDESP,
LIBERAR_DESCONTO,
FUNDESP_AUTOMATICA,
LANCAR_VALOR_DOCUMENTO,
VALOR_FIXO,
EMOLUMENTO_ID,
ATO_PRATICADO,
SELAR,
FRENTEVERSO,
PAGINA_ACRESCIDA,
EMOLUMENTO_OBRIGATORIO,
APRESENTANTE_SELO,
RENOVACAO_CARTAO,
ETIQUETA_UNICA,
TRANSFERENCIA_VEICULO,
USAR_A4,
AVERBACAO
) VALUES (
:servico_tipo_id,
:descricao,
:valor,
:tipo_item,
:requer_autorizacao,
:requer_biometria,
:tipo_pessoa,
:tb_reconhecimentotipo_id,
:tipo_permissao_cpf,
:requer_abonador,
:requer_representante,
:situacao,
:requer_cpf,
:servico_padrao,
:maximo_pessoa,
:alterar_valor,
:servico_caixa_id,
:lancar_taxa,
:lancar_fundesp,
:liberar_desconto,
:fundesp_automatica,
:lancar_valor_documento,
:valor_fixo,
:emolumento_id,
:ato_praticado,
:selar,
:frenteverso,
:pagina_acrescida,
:emolumento_obrigatorio,
:apresentante_selo,
:renovacao_cartao,
:etiqueta_unica,
:transferencia_veiculo,
:usar_a4,
:averbacao
) RETURNING *;"""
# Preenchimento de parâmetros
params = {
'servico_tipo_id': servico_tipo_schema.servico_tipo_id,
'descricao': servico_tipo_schema.descricao,
'valor': servico_tipo_schema.valor,
'tipo_item': servico_tipo_schema.tipo_item,
'requer_autorizacao': servico_tipo_schema.requer_autorizacao,
'requer_biometria': servico_tipo_schema.requer_biometria,
'tipo_pessoa': servico_tipo_schema.tipo_pessoa,
'tb_reconhecimentotipo_id': servico_tipo_schema.tb_reconhecimentotipo_id,
'tipo_permissao_cpf': servico_tipo_schema.tipo_permissao_cpf,
'requer_abonador': servico_tipo_schema.requer_abonador,
'requer_representante': servico_tipo_schema.requer_representante,
'situacao': servico_tipo_schema.situacao,
'requer_cpf': servico_tipo_schema.requer_cpf,
'servico_padrao': servico_tipo_schema.servico_padrao,
'maximo_pessoa': servico_tipo_schema.maximo_pessoa,
'alterar_valor': servico_tipo_schema.alterar_valor,
'servico_caixa_id': servico_tipo_schema.servico_caixa_id,
'lancar_taxa': servico_tipo_schema.lancar_taxa,
'lancar_fundesp': servico_tipo_schema.lancar_fundesp,
'liberar_desconto': servico_tipo_schema.liberar_desconto,
'fundesp_automatica': servico_tipo_schema.fundesp_automatica,
'lancar_valor_documento': servico_tipo_schema.lancar_valor_documento,
'valor_fixo': servico_tipo_schema.valor_fixo,
'emolumento_id': servico_tipo_schema.emolumento_id,
'ato_praticado': servico_tipo_schema.ato_praticado,
'selar': servico_tipo_schema.selar,
'frenteverso': servico_tipo_schema.frenteverso,
'pagina_acrescida': servico_tipo_schema.pagina_acrescida,
'emolumento_obrigatorio': servico_tipo_schema.emolumento_obrigatorio,
'apresentante_selo': servico_tipo_schema.apresentante_selo,
'renovacao_cartao': servico_tipo_schema.renovacao_cartao,
'etiqueta_unica': servico_tipo_schema.etiqueta_unica,
'transferencia_veiculo': servico_tipo_schema.transferencia_veiculo,
'usar_a4': servico_tipo_schema.usar_a4,
'averbacao': servico_tipo_schema.averbacao,
}
# 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 SERVICO TIPO: {e}"
)

View file

@ -0,0 +1,49 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_servico_tipo_schema import TServicoTipoIdSchema # Nome do schema ajustado
from fastapi import HTTPException, status
class ShowRepository(BaseRepository):
"""
Repositório para a operação de exibição de um registro na tabela T_SERVICO_TIPO.
"""
def execute(self, servico_tipo_schema: TServicoTipoIdSchema): # Nome do parâmetro ajustado
"""
Busca um registro específico de T_SERVICO_TIPO pelo ID.
Args:
servico_tipo_schema (TServicoTipoIdSchema): O esquema que contém o ID do registro. # Nome do tipo ajustado
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
# Tabela e coluna da chave primária ajustadas
sql = "SELECT * FROM T_SERVICO_TIPO WHERE SERVICO_TIPO_ID = :servico_tipo_id"
# Preenchimento de parâmetros
params = {
# Nome do parâmetro ajustado
'servico_tipo_id': servico_tipo_schema.servico_tipo_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:
# Tratamento de erro ajustado
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Erro ao buscar registro T_SERVICO_TIPO: {str(e)}"
)

View file

@ -0,0 +1,195 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_servico_tipo_schema import TServicoTipoUpdateSchema
from fastapi import HTTPException, status
class UpdateRepository(BaseRepository):
"""
Repositório para a operação de atualização na tabela T_SERVICO_TIPO.
"""
def execute(self, servico_tipo_id : int, servico_tipo_schema: TServicoTipoUpdateSchema):
"""
Executa a atualização de um registro na tabela.
Args:
servico_tipo_id (int): O ID do registro a ser atualizado.
servico_tipo_schema (TServicoTipoUpdateSchema): 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 = {}
# --- Campos da T_SERVICO_TIPO que podem ser atualizados (com base na DDL) ---
if servico_tipo_schema.descricao is not None:
updates.append("DESCRICAO = :descricao")
params["descricao"] = servico_tipo_schema.descricao
if servico_tipo_schema.valor is not None:
updates.append("VALOR = :valor")
params["valor"] = servico_tipo_schema.valor
if servico_tipo_schema.tipo_item is not None:
updates.append("TIPO_ITEM = :tipo_item")
params["tipo_item"] = servico_tipo_schema.tipo_item
if servico_tipo_schema.requer_autorizacao is not None:
updates.append("REQUER_AUTORIZACAO = :requer_autorizacao")
params["requer_autorizacao"] = servico_tipo_schema.requer_autorizacao
if servico_tipo_schema.requer_biometria is not None:
updates.append("REQUER_BIOMETRIA = :requer_biometria")
params["requer_biometria"] = servico_tipo_schema.requer_biometria
if servico_tipo_schema.tipo_pessoa is not None:
updates.append("TIPO_PESSOA = :tipo_pessoa")
params["tipo_pessoa"] = servico_tipo_schema.tipo_pessoa
# if servico_tipo_schema.tb_reconhecimentotipo_id is not None:
# updates.append("TB_RECONHECIMENTOTIPO_ID = :tb_reconhecimentotipo_id")
# params["tb_reconhecimentotipo_id"] = servico_tipo_schema.tb_reconhecimentotipo_id
if servico_tipo_schema.tipo_permissao_cpf is not None:
updates.append("TIPO_PERMISSAO_CPF = :tipo_permissao_cpf")
params["tipo_permissao_cpf"] = servico_tipo_schema.tipo_permissao_cpf
if servico_tipo_schema.requer_abonador is not None:
updates.append("REQUER_ABONADOR = :requer_abonador")
params["requer_abonador"] = servico_tipo_schema.requer_abonador
if servico_tipo_schema.requer_representante is not None:
updates.append("REQUER_REPRESENTANTE = :requer_representante")
params["requer_representante"] = servico_tipo_schema.requer_representante
if servico_tipo_schema.situacao is not None:
updates.append("SITUACAO = :situacao")
params["situacao"] = servico_tipo_schema.situacao
if servico_tipo_schema.requer_cpf is not None:
updates.append("REQUER_CPF = :requer_cpf")
params["requer_cpf"] = servico_tipo_schema.requer_cpf
if servico_tipo_schema.servico_padrao is not None:
updates.append("SERVICO_PADRAO = :servico_padrao")
params["servico_padrao"] = servico_tipo_schema.servico_padrao
if servico_tipo_schema.maximo_pessoa is not None:
updates.append("MAXIMO_PESSOA = :maximo_pessoa")
params["maximo_pessoa"] = servico_tipo_schema.maximo_pessoa
if servico_tipo_schema.alterar_valor is not None:
updates.append("ALTERAR_VALOR = :alterar_valor")
params["alterar_valor"] = servico_tipo_schema.alterar_valor
if servico_tipo_schema.servico_caixa_id is not None:
updates.append("SERVICO_CAIXA_ID = :servico_caixa_id")
params["servico_caixa_id"] = servico_tipo_schema.servico_caixa_id
if servico_tipo_schema.lancar_taxa is not None:
updates.append("LANCAR_TAXA = :lancar_taxa")
params["lancar_taxa"] = servico_tipo_schema.lancar_taxa
if servico_tipo_schema.lancar_fundesp is not None:
updates.append("LANCAR_FUNDESP = :lancar_fundesp")
params["lancar_fundesp"] = servico_tipo_schema.lancar_fundesp
if servico_tipo_schema.liberar_desconto is not None:
updates.append("LIBERAR_DESCONTO = :liberar_desconto")
params["liberar_desconto"] = servico_tipo_schema.liberar_desconto
if servico_tipo_schema.fundesp_automatica is not None:
updates.append("FUNDESP_AUTOMATICA = :fundesp_automatica")
params["fundesp_automatica"] = servico_tipo_schema.fundesp_automatica
if servico_tipo_schema.lancar_valor_documento is not None:
updates.append("LANCAR_VALOR_DOCUMENTO = :lancar_valor_documento")
params["lancar_valor_documento"] = servico_tipo_schema.lancar_valor_documento
if servico_tipo_schema.valor_fixo is not None:
updates.append("VALOR_FIXO = :valor_fixo")
params["valor_fixo"] = servico_tipo_schema.valor_fixo
# if servico_tipo_schema.emolumento_id is not None:
# updates.append("EMOLUMENTO_ID = :emolumento_id")
# params["emolumento_id"] = servico_tipo_schema.emolumento_id
if servico_tipo_schema.ato_praticado is not None:
updates.append("ATO_PRATICADO = :ato_praticado")
params["ato_praticado"] = servico_tipo_schema.ato_praticado
if servico_tipo_schema.selar is not None:
updates.append("SELAR = :selar")
params["selar"] = servico_tipo_schema.selar
if servico_tipo_schema.frenteverso is not None:
updates.append("FRENTEVERSO = :frenteverso")
params["frenteverso"] = servico_tipo_schema.frenteverso
if servico_tipo_schema.pagina_acrescida is not None:
updates.append("PAGINA_ACRESCIDA = :pagina_acrescida")
params["pagina_acrescida"] = servico_tipo_schema.pagina_acrescida
if servico_tipo_schema.emolumento_obrigatorio is not None:
updates.append("EMOLUMENTO_OBRIGATORIO = :emolumento_obrigatorio")
params["emolumento_obrigatorio"] = servico_tipo_schema.emolumento_obrigatorio
if servico_tipo_schema.apresentante_selo is not None:
updates.append("APRESENTANTE_SELO = :apresentante_selo")
params["apresentante_selo"] = servico_tipo_schema.apresentante_selo
if servico_tipo_schema.renovacao_cartao is not None:
updates.append("RENOVACAO_CARTAO = :renovacao_cartao")
params["renovacao_cartao"] = servico_tipo_schema.renovacao_cartao
if servico_tipo_schema.etiqueta_unica is not None:
updates.append("ETIQUETA_UNICA = :etiqueta_unica")
params["etiqueta_unica"] = servico_tipo_schema.etiqueta_unica
if servico_tipo_schema.transferencia_veiculo is not None:
updates.append("TRANSFERENCIA_VEICULO = :transferencia_veiculo")
params["transferencia_veiculo"] = servico_tipo_schema.transferencia_veiculo
if servico_tipo_schema.usar_a4 is not None:
updates.append("USAR_A4 = :usar_a4")
params["usar_a4"] = servico_tipo_schema.usar_a4
if servico_tipo_schema.averbacao is not None:
updates.append("AVERBACAO = :averbacao")
params["averbacao"] = servico_tipo_schema.averbacao
# -------------------------------------------------------------------------
if not updates:
return False
params["servico_tipo_id"] = servico_tipo_id
sql = f"UPDATE T_SERVICO_TIPO SET {', '.join(updates)} WHERE SERVICO_TIPO_ID = :servico_tipo_id RETURNING *;"
# Executa a query
result = self.run_and_return(sql, params)
# O result é uma tupla/objeto com os campos do registro,
# verificamos a existência do campo chave para confirmação
if not result or not getattr(result, 'servico_tipo_id', None):
# Informa que não existe o registro a ser modificado
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail='Nenhum SERVIÇO TIPO localizado para esta solicitação'
)
# Se houver um resultado e ele tiver o ID, 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 SERVIÇO TIPO: {e}"
)

View file

@ -0,0 +1,187 @@
from pydantic import BaseModel, field_validator, model_validator
from fastapi import HTTPException, status
from typing import Optional, ClassVar, Dict
from decimal import Decimal
# Assumindo que a classe Text está disponível no caminho 'actions.validations.text'
from actions.validations.text import Text
# ----------------------------------------------------
# Schema base
# ----------------------------------------------------
class GMarcacaoTipoSchema(BaseModel):
marcacao_tipo_id: Optional[int] = None # NUMERIC(10,2) PK
descricao: Optional[str] = None # VARCHAR(120)
texto: Optional[str] = None # BLOB SUB_TYPE BINARY
nome: Optional[str] = None # VARCHAR(30)
grupo: Optional[str] = None # VARCHAR(30)
situacao: Optional[str] = None # VARCHAR(1)
sistema_id: Optional[int] = None # NUMERIC(10,2)
grupo_tipo: Optional[str] = None # VARCHAR(1)
tipo_qualificacao: Optional[str] = None # VARCHAR(3)
condicao_sql: Optional[str] = None # VARCHAR(260)
separador_1: Optional[str] = None # VARCHAR(10)
separador_2: Optional[str] = None # VARCHAR(10)
separador_3: Optional[str] = None # VARCHAR(10)
tipo_valor: Optional[str] = None # VARCHAR(1)
atualizar: Optional[str] = None # VARCHAR(1)
protegida: Optional[str] = None # VARCHAR(1)
ativar_separador: Optional[str] = None # VARCHAR(1)
sql_completo: Optional[str] = None # VARCHAR(1000)
# Sanitiza campos de texto
@field_validator(
'descricao', 'texto', 'nome', 'grupo', 'situacao', 'grupo_tipo',
'tipo_qualificacao', 'condicao_sql', 'separador_1', 'separador_2',
'separador_3', 'tipo_valor', 'atualizar', 'protegida',
'ativar_separador', 'sql_completo'
)
def sanitize_fields(cls, v):
if isinstance(v, str) and v:
return Text.sanitize_input(v)
return v
# ----------------------------------------------------
# Schema para localizar um registro pelo ID (GET)
# ----------------------------------------------------
class GMarcacaoTipoIdSchema(BaseModel):
marcacao_tipo_id: int
# ----------------------------------------------------
# Schema para localizar um registro pela descrição (GET)
# ----------------------------------------------------
class GMarcacaoTipoDescricaoSchema(BaseModel):
descricao: str
# ----------------------------------------------------
# Schema para criação de novo registro (POST)
# ----------------------------------------------------
class GMarcacaoTipoSaveSchema(BaseModel):
marcacao_tipo_id: Optional[int] = None
descricao: str
nome: str
grupo: str
situacao: str
sistema_id: int
grupo_tipo: str
tipo_qualificacao: str
condicao_sql: str
separador_1: str
separador_2: str
separador_3: str
tipo_valor: str
atualizar: str
protegida: str
ativar_separador: str
sql_completo: str
# Sanitiza entradas
@field_validator(
'descricao', 'nome', 'grupo', 'situacao', 'grupo_tipo',
'tipo_qualificacao', 'condicao_sql', 'separador_1', 'separador_2',
'separador_3', 'tipo_valor', 'atualizar', 'protegida',
'ativar_separador', 'sql_completo'
)
def sanitize_fields(cls, v):
if isinstance(v, str) and v:
return Text.sanitize_input(v)
return v
# Verifica campos obrigatórios
@model_validator(mode='after')
def validate_required_fields(self):
required_fields = {
'descricao': 'A descrição é obrigatória.',
'nome': 'O nome é obrigatório.',
'grupo': 'O grupo é obrigatório.',
'situacao': 'A situação é obrigatória.',
'sistema_id': 'O sistema_id é obrigatório.',
'grupo_tipo': 'O grupo_tipo é obrigatório.',
'tipo_qualificacao': 'O tipo_qualificacao é obrigatório.',
'condicao_sql': 'A condição SQL é obrigatória.',
'separador_1': 'O separador_1 é obrigatório.',
'separador_2': 'O separador_2 é obrigatório.',
'separador_3': 'O separador_3 é obrigatório.',
'tipo_valor': 'O tipo_valor é obrigatório.',
'atualizar': 'O campo atualizar é obrigatório.',
'protegida': 'O campo protegida é obrigatório.',
'ativar_separador': 'O campo ativar_separador é obrigatório.',
'sql_completo': 'O sql_completo é obrigatório.',
}
errors = []
for field_name, message in required_fields.items():
field_value = getattr(self, field_name, None)
if field_value is None or (isinstance(field_value, str) and not field_value.strip()):
errors.append({'input': field_name, 'message': message})
if errors:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=errors)
return self
# ----------------------------------------------------
# Schema para atualização de registro (PUT)
# ----------------------------------------------------
class GMarcacaoTipoUpdateSchema(BaseModel):
descricao: Optional[str] = None
texto: Optional[str] = None
nome: Optional[str] = None
grupo: Optional[str] = None
situacao: Optional[str] = None
sistema_id: Optional[int] = None
grupo_tipo: Optional[str] = None
tipo_qualificacao: Optional[str] = None
condicao_sql: Optional[str] = None
separador_1: Optional[str] = None
separador_2: Optional[str] = None
separador_3: Optional[str] = None
tipo_valor: Optional[str] = None
atualizar: Optional[str] = None
protegida: Optional[str] = None
ativar_separador: Optional[str] = None
sql_completo: Optional[str] = None
updatable_fields: ClassVar[Dict[str, str]] = {
'descricao': 'A descrição não pode ser vazia.',
'texto': 'O campo texto não pode ser vazio.',
'nome': 'O nome não pode ser vazio.',
'grupo': 'O grupo não pode ser vazio.',
'situacao': 'A situação não pode ser vazia.',
'sistema_id': 'O sistema_id não pode ser vazio.',
'grupo_tipo': 'O grupo_tipo não pode ser vazio.',
'tipo_qualificacao': 'O tipo_qualificacao não pode ser vazio.',
'condicao_sql': 'A condição SQL não pode ser vazia.',
'separador_1': 'O separador_1 não pode ser vazio.',
'separador_2': 'O separador_2 não pode ser vazio.',
'separador_3': 'O separador_3 não pode ser vazio.',
'tipo_valor': 'O tipo_valor não pode ser vazio.',
'atualizar': 'O campo atualizar não pode ser vazio.',
'protegida': 'O campo protegida não pode ser vazio.',
'ativar_separador': 'O campo ativar_separador não pode ser vazio.',
'sql_completo': 'O sql_completo não pode ser vazio.',
}
@model_validator(mode='after')
def validate_update_fields(self):
has_data = False
errors = []
for field_name, message in self.updatable_fields.items():
field_value = getattr(self, field_name, None)
if field_value is not None:
has_data = True
if isinstance(field_value, str) and not field_value.strip():
errors.append({'input': field_name, 'message': message})
if not has_data:
errors.append({'message': 'Pelo menos um campo deve ser fornecido para atualização.'})
if errors:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=errors)
return self

View file

@ -0,0 +1,127 @@
from pydantic import BaseModel, field_validator, model_validator
from fastapi import HTTPException, status
from typing import Optional
from decimal import Decimal
# Funções para sanitização de entradas (evitar XSS, SQLi etc.)
# Assumindo que a classe Text está disponível no caminho 'actions.validations.text'
from actions.validations.text import Text
# ----------------------------------------------------
# Schema base
# ----------------------------------------------------
class TServicoEtiquetaSchema(BaseModel):
# SERVICO_ETIQUETA_ID NUMERIC(10,2) NOT NULL (Chave Primária)
servico_etiqueta_id: Optional[int] = None
# ETIQUETA_MODELO_ID NUMERIC(10,2)
etiqueta_modelo_id: Optional[int] = None
# SERVICO_TIPO_ID NUMERIC(10,2) (Chave Estrangeira)
servico_tipo_id: Optional[int] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar um SERVICO_ETIQUETA especifico pelo ID (GET)
# ----------------------------------------------------
class TServicoEtiquetaIdSchema(BaseModel):
servico_etiqueta_id: int
# ----------------------------------------------------
# Schema para localizar um SERVICO_ETIQUETA pelo SERVICO_TIPO_ID (GET)
# Útil para buscar todas as etiquetas de um determinado tipo de serviço.
# ----------------------------------------------------
class TServicoEtiquetaServicoTipoIdSchema(BaseModel):
servico_tipo_id: int
# ----------------------------------------------------
# Schema para criação de novo SERVICO_ETIQUETA (POST)
# ----------------------------------------------------
class TServicoEtiquetaSaveSchema(BaseModel):
# SERVICO_ETIQUETA_ID é geralmente definido pelo banco de dados ou é opcional para o POST
servico_etiqueta_id: Optional[int] = None
# ETIQUETA_MODELO_ID é obrigatório na criação (assumindo que deve ser associado a um modelo)
etiqueta_modelo_id: int
# SERVICO_TIPO_ID é obrigatório na criação (assumindo que deve ser associado a um serviço)
servico_tipo_id: int
# Não há campos string na DDL para sanitização, mas manter a estrutura de validação por boa prática
@model_validator(mode='after')
def validate_all_fields(self):
errors = []
required_fields = {
'etiqueta_modelo_id': 'O ID do Modelo de Etiqueta (etiqueta_modelo_id) é obrigatório.',
'servico_tipo_id': 'O ID do Tipo de Serviço (servico_tipo_id) é obrigatório.',
}
for field_name, message in required_fields.items():
field_value = getattr(self, field_name, None)
if field_value is None:
errors.append({'input': field_name, 'message': message})
# A DDL usa NUMERIC(10,2), então verificamos se é um número inteiro válido (e > 0 se for PK/FK)
elif not isinstance(field_value, int) or field_value <= 0:
errors.append({'input': field_name, 'message': f"{message} Deve ser um número inteiro positivo."})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self
# ----------------------------------------------------
# Schema para atualizar SERVICO_ETIQUETA (PUT)
# ----------------------------------------------------
class TServicoEtiquetaUpdateSchema(BaseModel):
etiqueta_modelo_id: Optional[int] = None
servico_tipo_id: Optional[int] = None
# Não há campos string na DDL para sanitização, mas a estrutura de validação é adaptada.
# Verifica se pelo menos um campo foi enviado, se não, lança exceção.
@model_validator(mode='after')
def validate_all_fields(self):
updatable_fields = {
'etiqueta_modelo_id': 'O ID do Modelo de Etiqueta (etiqueta_modelo_id) não pode ser nulo ou zero.',
'servico_tipo_id': 'O ID do Tipo de Serviço (servico_tipo_id) não pode ser nulo ou zero.',
}
has_data = False
errors = []
for field_name, message in updatable_fields.items():
field_value = getattr(self, field_name, None)
if field_value is not None:
has_data = True
# Para campos NUMERIC(10,2) que são FKs, verifica se são inteiros e positivos
if isinstance(field_value, int) and field_value <= 0:
errors.append({'input': field_name, 'message': f"O campo {field_name} deve ser um número inteiro positivo."})
if not has_data:
errors.append({'message': 'Pelo menos um campo deve ser fornecido para a atualização.'})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self

View file

@ -0,0 +1,305 @@
from pydantic import BaseModel, field_validator, model_validator
from fastapi import HTTPException, status
from typing import Optional
from decimal import Decimal
# Funções para sanitização de entradas (evitar XSS, SQLi etc.)
# Assumindo que a classe Text está disponível no caminho 'actions.validations.text'
from actions.validations.text import Text
# ----------------------------------------------------
# Schema base
# ----------------------------------------------------
class TServicoTipoSchema(BaseModel):
servico_tipo_id: Optional[int] = None
descricao: Optional[str] = None
valor: Optional[float] = None # Numeric(14,3)
tipo_item: Optional[str] = None # Varchar(1)
requer_autorizacao: Optional[str] = None # Varchar(1)
requer_biometria: Optional[str] = None # Varchar(1)
tipo_pessoa: Optional[str] = None # Varchar(1)
tb_reconhecimentotipo_id: Optional[int] = None # Numeric(10,2)
tipo_permissao_cpf: Optional[str] = None # Varchar(1)
requer_abonador: Optional[str] = None # Varchar(1)
requer_representante: Optional[str] = None # Varchar(1)
situacao: Optional[str] = None # Varchar(1)
requer_cpf: Optional[str] = None # Varchar(1)
servico_padrao: Optional[str] = None # Varchar(1)
maximo_pessoa: Optional[int] = None # Numeric(10,2)
alterar_valor: Optional[str] = None # Varchar(1)
servico_caixa_id: Optional[int] = None # Numeric(10,2)
lancar_taxa: Optional[str] = None # Varchar(1)
lancar_fundesp: Optional[str] = None # Varchar(1)
liberar_desconto: Optional[str] = None # Varchar(1)
fundesp_automatica: Optional[str] = None # Varchar(1)
lancar_valor_documento: Optional[str] = None # Varchar(1)
valor_fixo: Optional[str] = None # Varchar(1)
emolumento_id: Optional[int] = None # Numeric(10,2)
ato_praticado: Optional[str] = None # Varchar(1)
selar: Optional[str] = None # Varchar(1)
frenteverso: Optional[str] = None # Varchar(1)
pagina_acrescida: Optional[str] = None # Varchar(1)
emolumento_obrigatorio: Optional[int] = None # Numeric(10,2)
apresentante_selo: Optional[str] = None # Varchar(1)
renovacao_cartao: Optional[str] = None # Varchar(1)
etiqueta_unica: Optional[str] = None # Varchar(1)
transferencia_veiculo: Optional[str] = None # Varchar(1)
usar_a4: Optional[str] = None # Varchar(1)
averbacao: Optional[str] = None # Varchar(1)
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar um SERVICO especifico pelo ID (GET)
# ----------------------------------------------------
class TServicoTipoIdSchema(BaseModel):
servico_tipo_id: int
# ----------------------------------------------------
# Schema para localizar um SERVICO especifico pela descrição (GET)
# ----------------------------------------------------
class TServicoTipoDescricaoSchema(BaseModel):
descricao: str
# ----------------------------------------------------
# Schema para criação de novo SERVICO (POST)
# ----------------------------------------------------
class TServicoTipoSaveSchema(BaseModel):
servico_tipo_id: Optional[int] = None
descricao: str
valor: float # Numeric(14,3) - Considerado obrigatório para criação
tipo_item: str # Varchar(1)
requer_autorizacao: str # Varchar(1)
requer_biometria: str # Varchar(1)
tipo_pessoa: str # Varchar(1)
tb_reconhecimentotipo_id: Optional[int] = None # Opcional por ser FK e poder ser nulo
tipo_permissao_cpf: str # Varchar(1)
requer_abonador: str # Varchar(1)
requer_representante: str # Varchar(1)
situacao: str # Varchar(1)
requer_cpf: str # Varchar(1)
servico_padrao: str # Varchar(1)
maximo_pessoa: int # Numeric(10,2)
alterar_valor: str # Varchar(1)
servico_caixa_id: Optional[int] = None # Opcional por ser FK e poder ser nulo
lancar_taxa: str # Varchar(1)
lancar_fundesp: str # Varchar(1)
liberar_desconto: str # Varchar(1)
fundesp_automatica: str # Varchar(1)
lancar_valor_documento: str # Varchar(1)
valor_fixo: str # Varchar(1)
emolumento_id: Optional[int] = None # Opcional por ser FK e poder ser nulo
ato_praticado: str # Varchar(1)
selar: str # Varchar(1)
frenteverso: str # Varchar(1)
pagina_acrescida: str # Varchar(1)
emolumento_obrigatorio: Decimal # Numeric(10,2)
apresentante_selo: str # Varchar(1)
renovacao_cartao: str # Varchar(1)
etiqueta_unica: str # Varchar(1)
transferencia_veiculo: str # Varchar(1)
usar_a4: str # Varchar(1)
averbacao: str # Varchar(1)
# Campos que devem ser sanitizados (strings)
@field_validator(
'descricao', 'tipo_item', 'requer_autorizacao', 'requer_biometria',
'tipo_pessoa', 'tipo_permissao_cpf', 'requer_abonador', 'requer_representante',
'situacao', 'requer_cpf', 'servico_padrao', 'alterar_valor', 'lancar_taxa',
'lancar_fundesp', 'liberar_desconto', 'fundesp_automatica',
'lancar_valor_documento', 'valor_fixo', 'ato_praticado', 'selar',
'frenteverso', 'pagina_acrescida', 'apresentante_selo', 'renovacao_cartao',
'etiqueta_unica', 'transferencia_veiculo', 'usar_a4', 'averbacao'
)
def sanitize_fields(cls, v):
if isinstance(v, str) and v:
return Text.sanitize_input(v)
return v
# Verifica se os campos obrigatórios foram enviados (baseado no TServicoTipoSaveSchema original e na lógica da DDL)
@model_validator(mode='after')
def validate_all_fields(self):
errors = []
# Consideraremos 'descricao', 'valor', 'situacao' e todos os VARCHAR(1) como obrigatórios,
# seguindo o padrão do TCensecSaveSchema e campos que fazem sentido como requeridos em um POST.
required_fields = {
'descricao': 'A descrição é obrigatória.',
'valor': 'O valor é obrigatório.',
'tipo_item': 'O Tipo do Item é obrigatório.',
'requer_autorizacao': 'O Requer Autorização é obrigatório.',
'requer_biometria': 'O Requer Biometria é obrigatório.',
'tipo_pessoa': 'O Tipo Pessoa é obrigatório.',
'tipo_permissao_cpf': 'O Tipo Permissão CPF é obrigatório.',
'requer_abonador': 'O Requer Abonador é obrigatório.',
'requer_representante': 'O Requer Representante é obrigatório.',
'situacao': 'A situação é obrigatória.',
'requer_cpf': 'O Requer CPF é obrigatório.',
'servico_padrao': 'O Serviço Padrão é obrigatório.',
'maximo_pessoa': 'O Máximo Pessoa é obrigatório.',
'alterar_valor': 'O Alterar Valor é obrigatório.',
'lancar_taxa': 'O Lançar Taxa é obrigatório.',
'lancar_fundesp': 'O Lançar FUNDESP é obrigatório.',
'liberar_desconto': 'O Liberar Desconto é obrigatório.',
'fundesp_automatica': 'O FUNDESP Automática é obrigatório.',
'lancar_valor_documento': 'O Lançar Valor Documento é obrigatório.',
'valor_fixo': 'O Valor Fixo é obrigatório.',
'ato_praticado': 'O Ato Praticado é obrigatório.',
'selar': 'O Selar é obrigatório.',
'frenteverso': 'O FrenteVerso é obrigatório.',
'pagina_acrescida': 'A Página Acrescida é obrigatória.',
'emolumento_obrigatorio': 'O Emolumento Obrigatório é obrigatório.',
'apresentante_selo': 'O Apresentante Selo é obrigatório.',
'renovacao_cartao': 'A Renovação Cartão é obrigatória.',
'etiqueta_unica': 'A Etiqueta Única é obrigatória.',
'transferencia_veiculo': 'A Transferência Veículo é obrigatória.',
'usar_a4': 'O Usar A4 é obrigatório.',
'averbacao': 'A Averbação é obrigatória.',
}
for field_name, message in required_fields.items():
field_value = getattr(self, field_name, None)
if field_value is None:
errors.append({'input': field_name, 'message': message})
elif isinstance(field_value, str) and len(field_value.strip()) == 0:
errors.append({'input': field_name, 'message': message})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self
# ----------------------------------------------------
# Schema para atualizar SERVICO (PUT)
# ----------------------------------------------------
class TServicoTipoUpdateSchema(BaseModel):
descricao: Optional[str] = None
valor: Optional[float] = None
tipo_item: Optional[str] = None
requer_autorizacao: Optional[str] = None
requer_biometria: Optional[str] = None
tipo_pessoa: Optional[str] = None
tb_reconhecimentotipo_id: Optional[int] = None
tipo_permissao_cpf: Optional[str] = None
requer_abonador: Optional[str] = None
requer_representante: Optional[str] = None
situacao: Optional[str] = None
requer_cpf: Optional[str] = None
servico_padrao: Optional[str] = None
maximo_pessoa: Optional[int] = None
alterar_valor: Optional[str] = None
servico_caixa_id: Optional[int] = None
lancar_taxa: Optional[str] = None
lancar_fundesp: Optional[str] = None
liberar_desconto: Optional[str] = None
fundesp_automatica: Optional[str] = None
lancar_valor_documento: Optional[str] = None
valor_fixo: Optional[str] = None
emolumento_id: Optional[int] = None
ato_praticado: Optional[str] = None
selar: Optional[str] = None
frenteverso: Optional[str] = None
pagina_acrescida: Optional[str] = None
emolumento_obrigatorio: Optional[Decimal] = None
apresentante_selo: Optional[str] = None
renovacao_cartao: Optional[str] = None
etiqueta_unica: Optional[str] = None
transferencia_veiculo: Optional[str] = None
usar_a4: Optional[str] = None
averbacao: Optional[str] = None
# Campos que devem ser sanitizados (strings)
@field_validator(
'descricao', 'tipo_item', 'requer_autorizacao', 'requer_biometria',
'tipo_pessoa', 'tipo_permissao_cpf', 'requer_abonador', 'requer_representante',
'situacao', 'requer_cpf', 'servico_padrao', 'alterar_valor', 'lancar_taxa',
'lancar_fundesp', 'liberar_desconto', 'fundesp_automatica',
'lancar_valor_documento', 'valor_fixo', 'ato_praticado', 'selar',
'frenteverso', 'pagina_acrescida', 'apresentante_selo', 'renovacao_cartao',
'etiqueta_unica', 'transferencia_veiculo', 'usar_a4', 'averbacao'
)
def sanitize_fields(cls, v):
if isinstance(v, str) and v:
return Text.sanitize_input(v)
return v
# Verifica se pelo menos um campo foi enviado, se não, lança exceção.
# Se algum campo for enviado, ele deve ser válido (ex: 'descricao' não pode ser string vazia).
@model_validator(mode='after')
def validate_all_fields(self):
# Campos que, se fornecidos, não podem ser None ou strings vazias/apenas espaços.
# Todos os campos da DDL, exceto os IDs que são Opcionais (FKs), são verificados.
updatable_fields = {
'descricao': 'A descrição não pode ser vazia.',
'valor': 'O valor não pode ser vazio.',
'tipo_item': 'O Tipo do Item não pode ser vazio.',
'requer_autorizacao': 'O Requer Autorização não pode ser vazio.',
'requer_biometria': 'O Requer Biometria não pode ser vazio.',
'tipo_pessoa': 'O Tipo Pessoa não pode ser vazio.',
'tipo_permissao_cpf': 'O Tipo Permissão CPF não pode ser vazio.',
'requer_abonador': 'O Requer Abonador não pode ser vazio.',
'requer_representante': 'O Requer Representante não pode ser vazio.',
'situacao': 'A situação não pode ser vazia.',
'requer_cpf': 'O Requer CPF não pode ser vazio.',
'servico_padrao': 'O Serviço Padrão não pode ser vazio.',
'maximo_pessoa': 'O Máximo Pessoa não pode ser vazio.',
'alterar_valor': 'O Alterar Valor não pode ser vazio.',
'servico_caixa_id': 'O Servico Caixa ID não pode ser vazio.',
'lancar_taxa': 'O Lançar Taxa não pode ser vazio.',
'lancar_fundesp': 'O Lançar FUNDESP não pode ser vazio.',
'liberar_desconto': 'O Liberar Desconto não pode ser vazio.',
'fundesp_automatica': 'O FUNDESP Automática não pode ser vazio.',
'lancar_valor_documento': 'O Lançar Valor Documento não pode ser vazio.',
'valor_fixo': 'O Valor Fixo não pode ser vazio.',
'emolumento_id': 'O Emolumento ID não pode ser vazio.',
'ato_praticado': 'O Ato Praticado não pode ser vazio.',
'selar': 'O Selar não pode ser vazio.',
'frenteverso': 'O FrenteVerso não pode ser vazio.',
'pagina_acrescida': 'A Página Acrescida não pode ser vazia.',
'emolumento_obrigatorio': 'O Emolumento Obrigatório não pode ser vazio.',
'apresentante_selo': 'O Apresentante Selo não pode ser vazio.',
'renovacao_cartao': 'A Renovação Cartão não pode ser vazia.',
'etiqueta_unica': 'A Etiqueta Única não pode ser vazia.',
'transferencia_veiculo': 'A Transferência Veículo não pode ser vazia.',
'usar_a4': 'O Usar A4 não pode ser vazio.',
'averbacao': 'A Averbação não pode ser vazia.',
}
has_data = False
errors = []
for field_name, message in updatable_fields.items():
field_value = getattr(self, field_name, None)
if field_value is not None:
has_data = True
# Verifica se o campo é uma string vazia ou só espaços
if isinstance(field_value, str) and len(field_value.strip()) == 0:
errors.append({'input': field_name, 'message': message})
# Caso seja um campo numérico, verifica se é zero (opcional: a regra de negócio pode permitir zero)
# elif isinstance(field_value, (int, float)) and field_value <= 0:
# errors.append({'input': field_name, 'message': message})
if not has_data:
errors.append({'message': 'Pelo menos um campo deve ser fornecido para a atualização.'})
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_marcacao_tipo_schema import GMarcacaoTipoIdSchema
from packages.v1.administrativo.actions.g_marcacao_tipo.g_marcacao_tipo_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_MARCACAO_TIPO.
"""
def execute(self, marcacao_tipo_schema: GMarcacaoTipoIdSchema):
"""
Executa a operação de exclusão do registro no banco de dados.
Args:
marcacao_tipo_schema (GMarcacaoTipoIdSchema): 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(marcacao_tipo_schema)
# Retorno da informação
return data

View file

@ -0,0 +1,38 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoDescricaoSchema
from packages.v1.administrativo.actions.g_marcacao_tipo.g_marcacao_tipo_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_MARCACAO_TIPO pela sua descrição.
"""
def execute(self, marcacao_tipo_schema: GMarcacaoTipoDescricaoSchema, messageValidate: bool):
"""
Executa a operação de busca no banco de dados.
Args:
marcacao_tipo_schema (GMarcacaoTipoDescricaoSchema): 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(marcacao_tipo_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_MARCACAO_TIPO'
)
# Retorno da informação
return data

View file

@ -0,0 +1,33 @@
from fastapi import HTTPException, status
# Importação da Action ajustada para o novo prefixo
from packages.v1.administrativo.actions.g_marcacao_tipo.g_marcacao_tipo_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_MARCACAO_TIPO.
"""
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_MARCACAO_TIPO'
)
# Retorna as informações localizadas
return data

View file

@ -0,0 +1,76 @@
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
# Ajuste das importações para o novo schema
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoSaveSchema, GMarcacaoTipoDescricaoSchema
# Ajuste da importação para a nova action
from packages.v1.administrativo.actions.g_marcacao_tipo.g_marcacao_tipo_save_action import SaveAction
from fastapi import HTTPException, status
class SaveService:
"""
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_MARCACAO_TIPO.
"""
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_marcacao_tipo") # Tabela ajustada
pass
# Cadastra o novo G_MARCACAO_TIPO
def execute(self, marcacao_tipo_schema: GMarcacaoTipoSaveSchema): # Nome do parâmetro e tipo ajustados
# Armazenamento de possíveis erros de validação
errors = []
# Importa o service de busca por descrição dinamicamente
GetByDescricaoService = self.dynamic_import.service('g_marcacao_tipo_get_by_descricao_service', 'GetByDescricaoService')
get_by_descricao_service = GetByDescricaoService()
# Verifica se o registro de marcacao tipo já existe por descrição
# Uso do GMarcacaoTipoDescricaoSchema para a busca
self.response = get_by_descricao_service.execute(GMarcacaoTipoDescricaoSchema(descricao=marcacao_tipo_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
# Coluna primária ajustada (MARCACAO_TIPO_ID)
if not marcacao_tipo_schema.marcacao_tipo_id:
# Crio um objeto de sequencia
sequencia_schema = GSequenciaSchema()
# Define os dados para atualizar a sequencia
sequencia_schema.tabela = 'G_MARCACAO_TIPO' # Nome da tabela ajustado
# Busco a sequência atualizada
generate = GenerateService()
# Busco a sequência atualizada
sequencia = generate.execute(sequencia_schema)
# Atualiza os dados da chave primária
marcacao_tipo_schema.marcacao_tipo_id = sequencia.sequencia # Coluna primária ajustada
# Instanciamento de ações
# Ação já é importada como SaveAction
save_action = SaveAction()
# Retorna o resultado da ação
return save_action.execute(marcacao_tipo_schema)

View file

@ -0,0 +1,37 @@
from fastapi import HTTPException, status
# Importação do Schema ajustada
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoIdSchema
# Importação da Action ajustada
from packages.v1.administrativo.actions.g_marcacao_tipo.g_marcacao_tipo_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_MARCACAO_TIPO.
"""
def execute(self, marcacao_tipo_schema: GMarcacaoTipoIdSchema):
"""
Executa a operação de busca no banco de dados.
Args:
marcacao_tipo_schema (GMarcacaoTipoIdSchema): O esquema com o ID (MARCACAO_TIPO_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(marcacao_tipo_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_MARCACAO_TIPO'
)
# Retorno da informação
return data

View file

@ -0,0 +1,24 @@
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoUpdateSchema
from packages.v1.administrativo.actions.g_marcacao_tipo.g_marcacao_tipo_update_action import UpdateAction
class UpdateService:
"""
Serviço para a operação de atualização de um registro na tabela
G_MARCACAO_TIPO.
"""
def execute(self, marcacao_tipo_id : int, marcacao_tipo_schema: GMarcacaoTipoUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
marcacao_tipo_id (int): O ID do registro a ser atualizado (MARCACAO_TIPO_ID).
marcacao_tipo_schema (GMarcacaoTipoUpdateSchema): 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(marcacao_tipo_id, marcacao_tipo_schema)

View file

@ -0,0 +1,27 @@
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import TServicoEtiquetaIdSchema # Importação do Schema ajustada para T_SERVICO_ETIQUETA
from packages.v1.administrativo.actions.t_servico_etiqueta.t_servico_etiqueta_delete_action import DeleteAction # Importação da Action ajustada para T_SERVICO_ETIQUETA
class DeleteService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela T_SERVICO_ETIQUETA. # Nome da tabela ajustado
"""
def execute(self, servico_etiqueta_schema: TServicoEtiquetaIdSchema): # Nome do parâmetro e tipo ajustados
"""
Executa a operação de exclusão do registro no banco de dados.
Args:
servico_etiqueta_schema (TServicoEtiquetaIdSchema): O esquema com o ID do registro a ser excluído. # Nome do tipo ajustado
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(servico_etiqueta_schema) # Nome do parâmetro ajustado
# Retorno da informação
return data

View file

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

View file

@ -0,0 +1,68 @@
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
# Ajuste das importações para o novo schema
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import TServicoEtiquetaSaveSchema
# Ajuste da importação para a nova action
from packages.v1.administrativo.actions.t_servico_etiqueta.t_servico_etiqueta_save_action import SaveAction
from fastapi import HTTPException, status
class SaveService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvamento de um novo registro na tabela T_SERVICO_ETIQUETA.
"""
def __init__(self):
# Ação responsável por carregar as services de acordo com o estado
self.dynamic_import = DynamicImport()
# Define o pacote que deve ser carregado
self.dynamic_import.set_package("administrativo")
# Define a tabela que o pacote pertence
self.dynamic_import.set_table("t_servico_etiqueta") # Tabela ajustada
pass
# Cadastra o novo T_SERVICO_ETIQUETA
def execute(self, servico_etiqueta_schema: TServicoEtiquetaSaveSchema): # Nome do parâmetro e tipo ajustados
# Armazena possíveis erros
errors = []
# A verificação de unicidade por "descricao" foi removida, pois a tabela
# T_SERVICO_ETIQUETA não possui esse campo. Se a unicidade for necessária,
# deve ser implementada uma lógica de checagem pela combinação de ETIQUETA_MODELO_ID e SERVICO_TIPO_ID.
# 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
# Coluna primária ajustada
if not servico_etiqueta_schema.servico_etiqueta_id:
# Crio um objeto de sequencia
sequencia_schema = GSequenciaSchema()
# Define os dados para atualizar a sequencia
sequencia_schema.tabela = 'T_SERVICO_ETIQUETA' # Nome da tabela ajustado
# Busco a sequência atualizada
generate = GenerateService()
# Busco a sequência atualizada
sequencia = generate.execute(sequencia_schema)
# Atualiza os dados da chave primária
servico_etiqueta_schema.servico_etiqueta_id = sequencia.sequencia # Coluna primária ajustada
# Instanciamento de ações
# Ação já é importada como SaveAction (nome ajustado acima)
save_action = SaveAction()
# Retorna o resultado da operação
return save_action.execute(servico_etiqueta_schema) # Nome do parâmetro ajustado

View file

@ -0,0 +1,38 @@
from fastapi import HTTPException, status
# Importação do Schema ajustada para o de ID
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import TServicoEtiquetaIdSchema
# Importação da Action ajustada para ShowAction
from packages.v1.administrativo.actions.t_servico_etiqueta.t_servico_etiqueta_show_action import ShowAction
class ShowService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela T_SERVICO_ETIQUETA pelo seu ID.
"""
def execute(self, servico_etiqueta_schema: TServicoEtiquetaIdSchema): # Nome do parâmetro e tipo ajustados
"""
Executa a operação de busca no banco de dados.
Args:
servico_etiqueta_schema (TServicoEtiquetaIdSchema): O esquema com o ID a ser buscado.
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 = ShowAction() # Nome da Action ajustado
# Executa a ação em questão
data = show_action.execute(servico_etiqueta_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 T_SERVICO_ETIQUETA' # Mensagem de erro ajustada
)
# Retorno da informação
return data

View file

@ -0,0 +1,24 @@
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import TServicoEtiquetaUpdateSchema # Importação do Schema ajustada
from packages.v1.administrativo.actions.t_servico_etiqueta.t_servico_etiqueta_update_action import UpdateAction # Importação da Action ajustada
class UpdateService:
"""
Serviço para a operação de atualização de um registro na tabela
T_SERVICO_ETIQUETA.
"""
def execute(self, servico_etiqueta_id : int, servico_etiqueta_schema: TServicoEtiquetaUpdateSchema): # Nomes dos parâmetros ajustados
"""
Executa a operação de atualização no banco de dados.
Args:
servico_etiqueta_id (int): O ID do registro a ser atualizado (SERVICO_ETIQUETA_ID).
servico_etiqueta_schema (TServicoEtiquetaUpdateSchema): 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(servico_etiqueta_id, servico_etiqueta_schema) # Parâmetros ajustados

View file

@ -0,0 +1,27 @@
from packages.v1.administrativo.schemas.t_servico_tipo_schema import TServicoTipoIdSchema # Importação do Schema ajustada
from packages.v1.administrativo.actions.t_servico_tipo.t_servico_tipo_delete_action import DeleteAction # Importação da Action ajustada
class DeleteService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela T_SERVICO_TIPO. # Nome da tabela ajustado
"""
def execute(self, servico_tipo_schema: TServicoTipoIdSchema): # Nome do parâmetro e tipo ajustados
"""
Executa a operação de exclusão do registro no banco de dados.
Args:
servico_tipo_schema (TServicoTipoIdSchema): O esquema com o ID do registro a ser excluído. # Nome do tipo ajustado
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(servico_tipo_schema) # Nome do parâmetro ajustado
# Retorno da informação
return data

View file

@ -0,0 +1,38 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.schemas.t_servico_tipo_schema import TServicoTipoDescricaoSchema # Importação do Schema ajustada
from packages.v1.administrativo.actions.t_servico_tipo.t_servico_tipo_get_by_descricao_action import GetByDescricaoAction # Importação da Action ajustada
class GetByDescricaoService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela T_SERVICO_TIPO pela sua descrição. # Nome da tabela ajustado
"""
def execute(self, servico_tipo_schema: TServicoTipoDescricaoSchema, messageValidate: bool): # Nome do parâmetro e tipo ajustados
"""
Executa a operação de busca no banco de dados.
Args:
servico_tipo_schema (TServicoTipoDescricaoSchema): O esquema com a descrição a ser buscada. # Nome do tipo ajustado
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(servico_tipo_schema) # Nome do parâmetro ajustado
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 T_SERVICO_TIPO' # Mensagem de erro ajustada
)
# Retorno da informação
return data

View file

@ -0,0 +1,33 @@
from fastapi import HTTPException, status
# Importação da Action ajustada para o novo prefixo
from packages.v1.administrativo.actions.t_servico_tipo.t_servico_tipo_index_action import IndexAction
class IndexService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de registros na tabela T_SERVICO_TIPO.
"""
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 T_SERVICO_TIPO' # Mensagem de erro ajustada
)
# Retorna as informações localizadas
return data

View file

@ -0,0 +1,79 @@
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
# Ajuste das importações para o novo schema
from packages.v1.administrativo.schemas.t_servico_tipo_schema import TServicoTipoSaveSchema, TServicoTipoDescricaoSchema
# Ajuste da importação para a nova action
from packages.v1.administrativo.actions.t_servico_tipo.t_servico_tipo_save_action import SaveAction
from fastapi import HTTPException, status
class SaveService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvamento de um novo registro na tabela T_SERVICO_TIPO.
"""
def __init__(self):
# Ação responsável por carregar as services de acordo com o estado
self.dynamic_import = DynamicImport()
# Define o pacote que deve ser carregado
self.dynamic_import.set_package("administrativo")
# Define a tabela que o pacote pertence
self.dynamic_import.set_table("t_servico_tipo") # Tabela ajustada
pass
# Cadastra o novo T_SERVICO_TIPO
def execute(self, servico_tipo_schema: TServicoTipoSaveSchema): # Nome do parâmetro e tipo ajustados
# Armazena possíveis erros
errors = []
# Verifica se a descrição já está sendo utilizada
# Importação de service ajustada
descricao_service = self.dynamic_import.service("t_servico_tipo_get_by_descricao_service", "GetByDescricaoService")
# Instanciamento da service
self.descricao_service = descricao_service()
# Verifica se a descrição já está sendo utilizada
# Uso do novo schema e parâmetro
self.response = self.descricao_service.execute(TServicoTipoDescricaoSchema(descricao=servico_tipo_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
# Coluna primária ajustada
if not servico_tipo_schema.servico_tipo_id:
# Crio um objeto de sequencia
sequencia_schema = GSequenciaSchema()
# Define os dados para atualizar a sequencia
sequencia_schema.tabela = 'T_SERVICO_TIPO' # Nome da tabela ajustado
# Busco a sequência atualizada
generate = GenerateService()
# Busco a sequência atualizada
sequencia = generate.execute(sequencia_schema)
# Atualiza os dados da chave primária
servico_tipo_schema.servico_tipo_id = sequencia.sequencia # Coluna primária ajustada
# Instanciamento de ações
# Ação já é importada como SaveAction
save_action = SaveAction()
# Retorna o resultado da operação
return save_action.execute(servico_tipo_schema) # Nome do parâmetro ajustado

View file

@ -0,0 +1,37 @@
from fastapi import HTTPException, status
# Importação do Schema ajustada
from packages.v1.administrativo.schemas.t_servico_tipo_schema import TServicoTipoIdSchema
# Importação da Action ajustada
from packages.v1.administrativo.actions.t_servico_tipo.t_servico_tipo_show_action import ShowAction
class ShowService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela T_SERVICO_TIPO.
"""
def execute(self, servico_tipo_schema: TServicoTipoIdSchema): # Nome do parâmetro e tipo ajustados
"""
Executa a operação de busca no banco de dados.
Args:
servico_tipo_schema (TServicoTipoIdSchema): 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(servico_tipo_schema) # Parâmetro ajustado
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 T_SERVICO_TIPO' # Mensagem de erro ajustada
)
# Retorno da informação
return data

View file

@ -0,0 +1,24 @@
from packages.v1.administrativo.schemas.t_servico_tipo_schema import TServicoTipoUpdateSchema # Importação do Schema ajustada
from packages.v1.administrativo.actions.t_servico_tipo.t_servico_tipo_update_action import UpdateAction # Importação da Action ajustada
class UpdateService:
"""
Serviço para a operação de atualização de um registro na tabela
T_SERVICO_TIPO. # Nome da tabela ajustado
"""
def execute(self, servico_tipo_id : int, servico_tipo_schema: TServicoTipoUpdateSchema): # Nomes dos parâmetros ajustados
"""
Executa a operação de atualização no banco de dados.
Args:
servico_tipo_id (int): O ID do registro a ser atualizado (SERVICO_TIPO_ID).
servico_tipo_schema (TServicoTipoUpdateSchema): O esquema com os dados a serem atualizados. # Nome do tipo ajustado
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(servico_tipo_id, servico_tipo_schema) # Parâmetros ajustados

View file

@ -120,4 +120,24 @@ api_router.include_router(
# Inclui as rotas de g_cidade
api_router.include_router(
g_cidade_endpoint.router, prefix="/administrativo/g_cidade", tags=["Cidades"]
)
# Inclui as rotas de t_servico_tipo
api_router.include_router(
t_servico_tipo_endpoint.router, prefix="/administrativo/t_servico_tipo", tags=["Serviço tipo"]
)
# Inclui as rotas de g_marcacao_tipo
api_router.include_router(
g_marcacao_tipo_endpoint.router, prefix="/administrativo/g_marcacao_tipo", tags=["Marcação tipo"]
)
# Inclui as rotas de g_marcacao_tipo
api_router.include_router(
t_servico_etiqueta_endpoint.router, prefix="/administrativo/t_servico_etiqueta", tags=["Serviço etiqueta"]
)
# Inclui as rotas de g_uf
api_router.include_router(
g_uf_endpoint.router, prefix="/administrativo/g_uf", tags=["Estados"]
)