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

Merge remote-tracking branch 'origin/release(MVP/Sprint6)' into MVPTN-126

Browse source
This commit is contained in:
Keven Willian Pereira de Souza 2025-10-30 16:06:55 -03:00
commit e871333992
32 changed files with 935 additions and 221 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

136
Orius.postman_collection.json
View file

@ -70,7 +70,7 @@
"header": [],
"body": {
"mode": "raw",
"raw": "{\r\n \"login\" : \"Kenio0\",\r\n \"senha_api\": \"123123\"\r\n}",
"raw": "{\r\n \"login\" : \"Kenio\",\r\n \"senha_api\": \"123123\"\r\n}",
"options": {
"raw": {
"language": "json"
@ -8501,6 +8501,81 @@
},
"response": []
},
{
"name": "GET GRUPO",
"event": [
{
"listen": "test",
"script": {
"exec": [
"pm.test(\"Status code é 201 ou 200\", function () {\r",
" pm.expect(pm.response.code).to.be.oneOf([200, 201]);\r",
"});\r",
"\r",
"pm.test(\"Resposta contém ID ou confirmação\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData).to.have.any.keys(\"id\", \"message\", \"status\");\r",
"});\r",
"\r",
"pm.test(\"Mensagem indica sucesso\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData.message || \"\").to.match(/sucesso|criado|registrado/i);\r",
"});\r",
""
],
"type": "text/javascript",
"packages": {},
"requests": {}
}
}
],
"protocolProfileBehavior": {
"disableBodyPruning": true
},
"request": {
"auth": {
"type": "bearer",
"bearer": [
{
"key": "token",
"value": "{{BearerToken}}",
"type": "string"
}
]
},
"method": "GET",
"header": [],
"body": {
"mode": "raw",
"raw": ""
},
"url": {
"raw": "{{BaseUrlV1}}administrativo/g_marcacao_tipo/grupo?grupo=MODELO_ETIQUETA&sistema_id=2&situacao=A",
"host": [
"{{BaseUrlV1}}administrativo"
],
"path": [
"g_marcacao_tipo",
"grupo"
],
"query": [
{
"key": "grupo",
"value": "MODELO_ETIQUETA"
},
{
"key": "sistema_id",
"value": "2"
},
{
"key": "situacao",
"value": "A"
}
]
}
},
"response": []
},
{
"name": "GET ID",
"event": [
@ -8741,6 +8816,65 @@
}
]
},
{
"name": "Emolumentos",
"item": [
{
"name": "GET",
"event": [
{
"listen": "test",
"script": {
"exec": [
"pm.test(\"Status code é 201 ou 200\", function () {\r",
" pm.expect(pm.response.code).to.be.oneOf([200, 201]);\r",
"});\r",
"\r",
"pm.test(\"Resposta contém ID ou confirmação\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData).to.have.any.keys(\"id\", \"message\", \"status\");\r",
"});\r",
"\r",
"pm.test(\"Mensagem indica sucesso\", function () {\r",
" const jsonData = pm.response.json();\r",
" pm.expect(jsonData.message || \"\").to.match(/sucesso|criado|registrado/i);\r",
"});\r",
""
],
"type": "text/javascript",
"packages": {},
"requests": {}
}
}
],
"request": {
"auth": {
"type": "bearer",
"bearer": [
{
"key": "token",
"value": "{{BearerToken}}",
"type": "string"
}
]
},
"method": "GET",
"header": [],
"url": {
"raw": "{{BaseUrlV1}}administrativo/g_emolumento/1",
"host": [
"{{BaseUrlV1}}administrativo"
],
"path": [
"g_emolumento",
"1"
]
}
},
"response": []
}
]
},
{
"name": "Andamento Serviço",
"item": [

16
packages/v1/administrativo/actions/c_caixa_servico/c_caixa_servico_get_by_sistema_id_action.py Normal file
View file

@ -0,0 +1,16 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.c_caixa_servico_schema import CCaixaServicoSistemaIdSchema
from packages.v1.administrativo.repositories.c_caixa_servico.c_caixa_servico_get_by_sistema_id_repository import ShowSistemaIdRepository
class ShowSistemaIdAction(BaseAction):
def execute(self, caixa_servico_schema : CCaixaServicoSistemaIdSchema):
# Instânciamento do repositório sql
show_sistema_id_repository = ShowSistemaIdRepository()
# Execução do sql
response = show_sistema_id_repository.execute(caixa_servico_schema)
# Retorno da informação
return response

35
packages/v1/administrativo/actions/g_emolumento/g_emolumento_index_action.py
View file

@ -1,35 +0,0 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.g_emolumento.g_emolumento_index_repository import GEmolumentoIndexRepository
class GEmolumentoIndexAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de todos os registros na tabela G_NATUREZA_TITULO.
"""
def execute(self):
"""
Executa a operação de listagem no banco de dados.
Args:
g_emolumento_index_schema (GEmolumentoIndexSchema):
Esquema contendo parâmetros opcionais de filtro.
Returns:
A lista de registros encontrados.
"""
# ----------------------------------------------------
# Instanciamento do repositório
# ----------------------------------------------------
g_emolumento_index_repository = GEmolumentoIndexRepository()
# ----------------------------------------------------
# Execução do repositório
# ----------------------------------------------------
response = g_emolumento_index_repository.execute()
# ----------------------------------------------------
# Retorno da informação
# --------------------------------
return response

44
packages/v1/administrativo/actions/g_emolumento_item/g_emolumento_item_valor_action.py Normal file
View file

@ -0,0 +1,44 @@
from abstracts.action import BaseAction
# O Schema precisa ser adaptado para GEmolumentoItem, talvez recebendo um ID de Emolumento
# ou um ID do próprio Item, dependendo da necessidade de listagem.
# Vamos sugerir um Schema específico para a listagem (Index) que pode receber um Emolumento ID
# para listar todos os seus itens, mantendo o padrão do arquivo original que usava um 'SistemaIdSchema'.
# Para simplificar, vamos assumir um schema de filtro ou um schema base para Index.
# Sugerimos a criação de:
# from packages.v1.administrativo.schemas.g_emolumento_item_schema import GEmolumentoItemIndexSchema
# (ou GEmolumentoItemEmolumentoIdSchema se for o padrão da aplicação)
from packages.v1.administrativo.schemas.g_emolumento_item_schema import GEmolumentoItemValorSchema
# O repositório ValorRepository deve ser substituído pelo GEmolumentoItemValorRepository.
from packages.v1.administrativo.repositories.g_emolumento_item.g_emolumento_item_valor_repository import ValorRepository
class ValorAction(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_EMOLUMENTO_ITEM,
utilizando a DDL fornecida.
"""
# Mantendo o padrão de nome de método do arquivo original
def execute(self, emolumento_item_schema: GEmolumentoItemValorSchema):
"""
Executa a operação de listagem de G_EMOLUMENTO_ITEM no banco de dados.
Args:
emolumento_item_schema: Esquema com parâmetros de filtro/listagem
(por exemplo, ID do Emolumento pai, se a listagem for
filtrada por ele, ou parâmetros de paginação).
Returns:
A lista de todos os registros de G_EMOLUMENTO_ITEM que satisfazem o filtro.
"""
# Instanciamento do repositório
# O nome do repositório foi adaptado com o prefixo 'GEmolumentoItem'
index_repository = ValorRepository()
# Execução do repositório
# O nome do parâmetro foi adaptado para 'emolumento_item_schema'
response = index_repository.execute(emolumento_item_schema)
# Retorno da informação
return response

31
packages/v1/administrativo/actions/g_marcacao_tipo/g_marcacao_tipo_get_by_grupo_action.py Normal file
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 GMarcacaoTipoGrupoSchema
# Ajuste do repositório
from packages.v1.administrativo.repositories.g_marcacao_tipo.g_marcacao_tipo_get_by_grupo_repository import GetByGrupoRepository
class GetByGrupoAction(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 filtro.
"""
def execute(self, marcacao_tipo_schema: GMarcacaoTipoGrupoSchema):
"""
Executa a operação de busca no banco de dados.
Args:
marcacao_tipo_schema (GMarcacaoTipoGrupoSchema): O esquema com os filtros a serem buscados.
Returns:
O registro encontrado ou None.
"""
# Instanciamento do repositório
show_repository = GetByGrupoRepository()
# Execução do repositório
response = show_repository.execute(marcacao_tipo_schema)
# Retorno da informação
return response

28
packages/v1/administrativo/actions/t_servico_etiqueta/t_servico_etiqueta_show_servico_tipo_action.py Normal file
View file

@ -0,0 +1,28 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import TServicoEtiquetaServicoTipoIdSchema
from packages.v1.administrativo.repositories.t_servico_etiqueta.t_servico_etiqueta_show_servico_tipo_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_tipo_schema: TServicoEtiquetaServicoTipoIdSchema):
"""
Executa a operação de exibição.
Args:
servico_etiqueta_schema (TServicoEtiquetaServicoTipoIdSchema): 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

21
packages/v1/administrativo/controllers/c_caixa_servico_controller.py
View file

@ -4,7 +4,8 @@ from packages.v1.administrativo.schemas.c_caixa_servico_schema import (
CCaixaServicoSaveSchema,
CCaixaServicoUpdateSchema,
CCaixaServicoIdSchema,
CCaixaServicoDescricaoSchema
CCaixaServicoDescricaoSchema,
CCaixaServicoSistemaIdSchema
)
class CCaixaServicoController:
@ -65,7 +66,23 @@ class CCaixaServicoController:
return {
'message' : 'Caixa Serviço localizado com sucesso',
'data': self.show_service.execute(caixa_servico_schema, True)# True para retornar a mensagem de erro caso não localize o serviço
}
}
# Busca um caixa serviço pelp sistema_id
def getSistemaId(self, caixa_servico_schema : CCaixaServicoSistemaIdSchema):
#Importação da classe desejada
show_sistema_id_service = self.dynamic_import.service('c_caixa_servico_get_sistema_id_service', 'GetSistemaIdService')
# Instânciamento da classe desejada
self.show_sistema_id_service = show_sistema_id_service()
# Busca e retorna o usuário desejado
return {
'message' : 'Caixa Serviços localizados com sucesso',
'data': self.show_sistema_id_service.execute(caixa_servico_schema)
}
# Cadastra um novo usuário

22
packages/v1/administrativo/controllers/g_emolumento_controller.py
View file

@ -2,7 +2,7 @@ from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.g_emolumento_schema import (
GEmolumentoSaveSchema,
GEmolumentoUpdateSchema,
GEmolumentoIdSchema
GEmolumentoIdSchema,
)
@ -25,7 +25,9 @@ class GEmolumentoController:
# ----------------------------------------------------
def index(self):
# Importação da classe desejada
index_service = self.dynamic_import.service("g_emolumento_index_service", "GEmolumentoIndexService")
index_service = self.dynamic_import.service(
"g_emolumento_index_service", "GEmolumentoIndexService"
)
# Instância da classe service
self.index_service = index_service()
@ -41,7 +43,9 @@ class GEmolumentoController:
# ----------------------------------------------------
def show(self, g_emolumento_id_schema: GEmolumentoIdSchema):
# Importação da classe desejada
show_service = self.dynamic_import.service("g_emolumento_show_service", "GEmolumentoShowService")
show_service = self.dynamic_import.service(
"g_emolumento_show_service", "GEmolumentoShowService"
)
# Instância da classe service
self.show_service = show_service()
@ -57,7 +61,9 @@ class GEmolumentoController:
# ----------------------------------------------------
def save(self, g_emolumento_save_schema: GEmolumentoSaveSchema):
# Importação da classe desejada
save_service = self.dynamic_import.service("g_emolumento_save_service", "GEmolumentoSaveService")
save_service = self.dynamic_import.service(
"g_emolumento_save_service", "GEmolumentoSaveService"
)
# Instância da classe service
self.save_service = save_service()
@ -73,7 +79,9 @@ class GEmolumentoController:
# ----------------------------------------------------
def update(self, g_emolumento_update_schema: GEmolumentoUpdateSchema):
# Importação da classe desejada
update_service = self.dynamic_import.service("g_emolumento_update_service", "GEmolumentoUpdateService")
update_service = self.dynamic_import.service(
"g_emolumento_update_service", "GEmolumentoUpdateService"
)
# Instância da classe service
self.update_service = update_service()
@ -89,7 +97,9 @@ class GEmolumentoController:
# ----------------------------------------------------
def delete(self, g_emolumento_id_schema: GEmolumentoIdSchema):
# Importação da classe desejada
delete_service = self.dynamic_import.service("g_emolumento_delete_service", "GEmolumentoDeleteService")
delete_service = self.dynamic_import.service(
"g_emolumento_delete_service", "GEmolumentoDeleteService"
)
# Instância da classe service
self.delete_service = delete_service()

30
packages/v1/administrativo/controllers/g_emolumento_item_controller.py
View file

@ -3,7 +3,7 @@ from packages.v1.administrativo.schemas.g_emolumento_item_schema import (
GEmolumentoItemIndexSchema,
GEmolumentoItemSaveSchema,
GEmolumentoItemUpdateSchema,
GEmolumentoItemIdSchema
GEmolumentoItemIdSchema,
)
@ -24,10 +24,14 @@ class GEmolumentoItemController:
# ----------------------------------------------------
# Lista todos os registros de G_EMOLUMENTO_ITEM
# ----------------------------------------------------
def index(self, g_emolumento_item_emolumento_index_schema: GEmolumentoItemIndexSchema):
def index(
self, g_emolumento_item_emolumento_index_schema: GEmolumentoItemIndexSchema
):
# Importação da classe desejada
index_service = self.dynamic_import.service("g_emolumento_item_index_service", "GEmolumentoItemIndexService")
index_service = self.dynamic_import.service(
"g_emolumento_item_index_service", "GEmolumentoItemIndexService"
)
# Instância da classe service
self.index_service = index_service()
@ -35,7 +39,9 @@ class GEmolumentoItemController:
# Execução da listagem
return {
"message": "Registros de G_EMOLUMENTO_ITEM localizados com sucesso.",
"data": self.index_service.execute(g_emolumento_item_emolumento_index_schema),
"data": self.index_service.execute(
g_emolumento_item_emolumento_index_schema
),
}
# ----------------------------------------------------
@ -43,7 +49,9 @@ class GEmolumentoItemController:
# ----------------------------------------------------
def show(self, g_emolumento_item_id_schema: GEmolumentoItemIdSchema):
# Importação da classe desejada
show_service = self.dynamic_import.service("g_emolumento_item_show_service", "GEmolumentoItemShowService")
show_service = self.dynamic_import.service(
"g_emolumento_item_show_service", "GEmolumentoItemShowService"
)
# Instância da classe service
self.show_service = show_service()
@ -59,7 +67,9 @@ class GEmolumentoItemController:
# ----------------------------------------------------
def save(self, g_emolumento_item_save_schema: GEmolumentoItemSaveSchema):
# Importação da classe desejada
save_service = self.dynamic_import.service("g_emolumento_item_save_service", "GEmolumentoItemSaveService")
save_service = self.dynamic_import.service(
"g_emolumento_item_save_service", "GEmolumentoItemSaveService"
)
# Instância da classe service
self.save_service = save_service()
@ -75,7 +85,9 @@ class GEmolumentoItemController:
# ----------------------------------------------------
def update(self, g_emolumento_item_update_schema: GEmolumentoItemUpdateSchema):
# Importação da classe desejada
update_service = self.dynamic_import.service("g_emolumento_item_update_service", "GEmolumentoItemUpdateService")
update_service = self.dynamic_import.service(
"g_emolumento_item_update_service", "GEmolumentoItemUpdateService"
)
# Instância da classe service
self.update_service = update_service()
@ -91,7 +103,9 @@ class GEmolumentoItemController:
# ----------------------------------------------------
def delete(self, g_emolumento_item_id_schema: GEmolumentoItemIdSchema):
# Importação da classe desejada
delete_service = self.dynamic_import.service("g_emolumento_item_delete_service", "GEmolumentoItemDeleteService")
delete_service = self.dynamic_import.service(
"g_emolumento_item_delete_service", "GEmolumentoItemDeleteService"
)
# Instância da classe service
self.delete_service = delete_service()

21
packages/v1/administrativo/controllers/g_marcacao_tipo_controller.py
View file

@ -4,7 +4,8 @@ from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import (
GMarcacaoTipoSaveSchema,
GMarcacaoTipoUpdateSchema,
GMarcacaoTipoIdSchema,
GMarcacaoTipoDescricaoSchema
GMarcacaoTipoDescricaoSchema,
GMarcacaoTipoGrupoSchema
)
class GMarcacaoTipoController:
@ -110,4 +111,20 @@ class GMarcacaoTipoController:
return {
'message': 'Registro de marcação tipo removido com sucesso',
'data': self.delete_service.execute(marcacao_tipo_schema)
}
}
# Busca um registro de marcacao tipo por filtro
def get_by_grupo(self, marcacao_tipo_schema: GMarcacaoTipoGrupoSchema):
#Importação da classe desejada
grupo_service = self.dynamic_import.service('g_marcacao_tipo_get_by_grupo_service', 'GetByGrupoService')
# Instância da classe desejada
self.grupo_service = grupo_service()
# Busca e retorna o registro de marcacao tipo desejado
return {
'message': 'Registro de marcação tipo localizado com sucesso',
'data': self.grupo_service.execute(marcacao_tipo_schema, True)
}

15
packages/v1/administrativo/controllers/t_servico_etiqueta_controller.py
View file

@ -35,6 +35,21 @@ class TServicoEtiquetaController:
'data': self.index_service.execute()
}
# Busca um registro de servico etiqueta específico pelo ID serviço tipo
def showServicoTipo(self, servico_tipo_schema: TServicoEtiquetaServicoTipoIdSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('t_servico_etiqueta_show_servico_tipo_service', 'ShowService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de servico etiqueta desejado
return {
'message': 'Registros de serviço etiqueta localizados com sucesso',
'data': self.show_service.execute(servico_tipo_schema)
}
# Busca um registro de servico etiqueta específico pelo ID
def show(self, servico_etiqueta_schema: TServicoEtiquetaIdSchema):

20
packages/v1/administrativo/endpoints/c_caixa_servico_endpoint.py
View file

@ -7,7 +7,8 @@ from packages.v1.administrativo.schemas.c_caixa_servico_schema import (
CCaixaServicoSchema,
CCaixaServicoSaveSchema,
CCaixaServicoUpdateSchema,
CCaixaServicoDescricaoSchema
CCaixaServicoDescricaoSchema,
CCaixaServicoSistemaIdSchema
)
# Inicializa o roteador para as rotas de usuário
@ -47,6 +48,23 @@ async def getDescricao(descricao : str, current_user: dict = Depends(get_current
return response
# Localiza um caixa serviço pelo sistema_id
@router.get('/sistema/{sistema_id}',
status_code=status.HTTP_200_OK,
summary='Busca registros pelo sistema_id',
response_description='Busca registros pelo sistema_id')
async def getSistemaId(sistema_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
caixa_servico_schema = CCaixaServicoSistemaIdSchema(sistema_id=sistema_id)
# Busca um usuário especifico pelo e-mail
response = c_caixa_servico_controller.getSistemaId(caixa_servico_schema)
# Retorna os dados localizados
return response
# Localiza um serviço pelo ID
@router.get('/{caixa_servico_id}',
status_code=status.HTTP_200_OK,

70
packages/v1/administrativo/endpoints/g_emolumento_endpoint.py
View file

@ -1,11 +1,13 @@
# Importação de bibliotecas
from fastapi import APIRouter, Depends, status
from actions.jwt.get_current_user import get_current_user
from packages.v1.administrativo.controllers.g_emolumento_controller import GEmolumentoController
from packages.v1.administrativo.controllers.g_emolumento_controller import (
GEmolumentoController,
)
from packages.v1.administrativo.schemas.g_emolumento_schema import (
GEmolumentoSaveSchema,
GEmolumentoUpdateSchema,
GEmolumentoIdSchema
GEmolumentoIdSchema,
)
# ----------------------------------------------------
@ -16,13 +18,16 @@ router = APIRouter()
# Instanciamento do controller
g_emolumento_controller = GEmolumentoController()
# ----------------------------------------------------
# Lista todos os registros de G_EMOLUMENTO
# ----------------------------------------------------
@router.get('/',
status_code=status.HTTP_200_OK,
summary='Lista todos os registros de G_EMOLUMENTO cadastrados',
response_description='Lista todos os registros de G_EMOLUMENTO cadastrados')
@router.get(
"/",
status_code=status.HTTP_200_OK,
summary="Lista todos os registros de G_EMOLUMENTO cadastrados",
response_description="Lista todos os registros de G_EMOLUMENTO cadastrados",
)
async def index(current_user: dict = Depends(get_current_user)):
"""
Retorna todos os registros da tabela G_EMOLUMENTO.
@ -34,10 +39,12 @@ async def index(current_user: dict = Depends(get_current_user)):
# ----------------------------------------------------
# Busca um registro específico de G_EMOLUMENTO pelo ID
# ----------------------------------------------------
@router.get('/{emolumento_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de G_EMOLUMENTO pelo ID',
response_description='Busca um registro de G_EMOLUMENTO em específico')
@router.get(
"/{emolumento_id}",
status_code=status.HTTP_200_OK,
summary="Busca um registro de G_EMOLUMENTO pelo ID",
response_description="Busca um registro de G_EMOLUMENTO em específico",
)
async def show(emolumento_id: int, current_user: dict = Depends(get_current_user)):
"""
Retorna um registro específico de G_EMOLUMENTO com base no ID informado.
@ -50,11 +57,16 @@ async def show(emolumento_id: int, current_user: dict = Depends(get_current_user
# ----------------------------------------------------
# Cadastra um novo registro em G_EMOLUMENTO
# ----------------------------------------------------
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um novo registro em G_EMOLUMENTO',
response_description='Cadastra um novo registro em G_EMOLUMENTO')
async def save(g_emolumento_schema: GEmolumentoSaveSchema, current_user: dict = Depends(get_current_user)):
@router.post(
"/",
status_code=status.HTTP_201_CREATED,
summary="Cadastra um novo registro em G_EMOLUMENTO",
response_description="Cadastra um novo registro em G_EMOLUMENTO",
)
async def save(
g_emolumento_schema: GEmolumentoSaveSchema,
current_user: dict = Depends(get_current_user),
):
"""
Cria um novo registro na tabela G_EMOLUMENTO.
"""
@ -65,11 +77,17 @@ async def save(g_emolumento_schema: GEmolumentoSaveSchema, current_user: dict =
# ----------------------------------------------------
# Atualiza um registro existente de G_EMOLUMENTO
# ----------------------------------------------------
@router.put('/{emolumento_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza um registro existente em G_EMOLUMENTO',
response_description='Atualiza um registro existente em G_EMOLUMENTO')
async def update(emolumento_id: int, g_emolumento_update_schema: GEmolumentoUpdateSchema, current_user: dict = Depends(get_current_user)):
@router.put(
"/{emolumento_id}",
status_code=status.HTTP_200_OK,
summary="Atualiza um registro existente em G_EMOLUMENTO",
response_description="Atualiza um registro existente em G_EMOLUMENTO",
)
async def update(
emolumento_id: int,
g_emolumento_update_schema: GEmolumentoUpdateSchema,
current_user: dict = Depends(get_current_user),
):
"""
Atualiza um registro existente de G_EMOLUMENTO com base no ID informado.
"""
@ -81,14 +99,16 @@ async def update(emolumento_id: int, g_emolumento_update_schema: GEmolumentoUpda
# ----------------------------------------------------
# Exclui um registro de G_EMOLUMENTO
# ----------------------------------------------------
@router.delete('/{emolumento_id}',
status_code=status.HTTP_200_OK,
summary='Remove um registro de G_EMOLUMENTO',
response_description='Remove um registro de G_EMOLUMENTO')
@router.delete(
"/{emolumento_id}",
status_code=status.HTTP_200_OK,
summary="Remove um registro de G_EMOLUMENTO",
response_description="Remove um registro de G_EMOLUMENTO",
)
async def delete(emolumento_id: int, current_user: dict = Depends(get_current_user)):
"""
Remove um registro específico de G_EMOLUMENTO com base no ID informado.
"""
g_emolumento_id_schema = GEmolumentoIdSchema(emolumento_id=emolumento_id)
response = g_emolumento_controller.delete(g_emolumento_id_schema)
return response
return response

96
packages/v1/administrativo/endpoints/g_emolumento_item_endpoint.py
View file

@ -1,12 +1,14 @@
# Importação de bibliotecas
from fastapi import APIRouter, Depends, status
from actions.jwt.get_current_user import get_current_user
from packages.v1.administrativo.controllers.g_emolumento_item_controller import GEmolumentoItemController
from packages.v1.administrativo.controllers.g_emolumento_item_controller import (
GEmolumentoItemController,
)
from packages.v1.administrativo.schemas.g_emolumento_item_schema import (
GEmolumentoItemIndexSchema,
GEmolumentoItemSaveSchema,
GEmolumentoItemUpdateSchema,
GEmolumentoItemIdSchema
GEmolumentoItemIdSchema,
)
# ----------------------------------------------------
@ -17,34 +19,49 @@ router = APIRouter()
# Instanciamento do controller
g_emolumento_item_controller = GEmolumentoItemController()
# ----------------------------------------------------
# Lista todos os registros de G_EMOLUMENTO_ITEM
# ----------------------------------------------------
@router.get('/{emolumento_id}/{emolumento_periodo_id}',
status_code=status.HTTP_200_OK,
summary='Lista todos os registros de G_EMOLUMENTO_ITEM cadastrados',
response_description='Lista todos os registros de G_EMOLUMENTO_ITEM cadastrados')
async def index(emolumento_id : int, emolumento_periodo_id : int, current_user: dict = Depends(get_current_user)):
@router.get(
"/{emolumento_id}/{emolumento_periodo_id}",
status_code=status.HTTP_200_OK,
summary="Lista todos os registros de G_EMOLUMENTO_ITEM cadastrados",
response_description="Lista todos os registros de G_EMOLUMENTO_ITEM cadastrados",
)
async def index(
emolumento_id: int,
emolumento_periodo_id: int,
current_user: dict = Depends(get_current_user),
):
"""
Retorna todos os registros da tabela G_EMOLUMENTO_ITEM.
"""
g_emolumento_item_emolumento_index_schema = GEmolumentoItemIndexSchema(emolumento_id=emolumento_id, emolumento_periodo_id=emolumento_periodo_id)
response = g_emolumento_item_controller.index(g_emolumento_item_emolumento_index_schema)
g_emolumento_item_emolumento_index_schema = GEmolumentoItemIndexSchema(
emolumento_id=emolumento_id, emolumento_periodo_id=emolumento_periodo_id
)
response = g_emolumento_item_controller.index(
g_emolumento_item_emolumento_index_schema
)
return response
# ----------------------------------------------------
# Busca um registro específico de G_EMOLUMENTO_ITEM pelo ID
# ----------------------------------------------------
@router.get('/{emolumento_item_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de G_EMOLUMENTO_ITEM pelo ID',
response_description='Busca um registro de G_EMOLUMENTO_ITEM em específico')
@router.get(
"/{emolumento_item_id}",
status_code=status.HTTP_200_OK,
summary="Busca um registro de G_EMOLUMENTO_ITEM pelo ID",
response_description="Busca um registro de G_EMOLUMENTO_ITEM em específico",
)
async def show(emolumento_item_id: int, current_user: dict = Depends(get_current_user)):
"""
Retorna um registro específico de G_EMOLUMENTO_ITEM com base no ID informado.
"""
g_emolumento_item_id_schema = GEmolumentoItemIdSchema(emolumento_item_id=emolumento_item_id)
g_emolumento_item_id_schema = GEmolumentoItemIdSchema(
emolumento_item_id=emolumento_item_id
)
response = g_emolumento_item_controller.show(g_emolumento_item_id_schema)
return response
@ -52,11 +69,16 @@ async def show(emolumento_item_id: int, current_user: dict = Depends(get_current
# ----------------------------------------------------
# Cadastra um novo registro em G_EMOLUMENTO_ITEM
# ----------------------------------------------------
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um novo registro em G_EMOLUMENTO_ITEM',
response_description='Cadastra um novo registro em G_EMOLUMENTO_ITEM')
async def save(g_emolumento_item_schema: GEmolumentoItemSaveSchema, current_user: dict = Depends(get_current_user)):
@router.post(
"/",
status_code=status.HTTP_201_CREATED,
summary="Cadastra um novo registro em G_EMOLUMENTO_ITEM",
response_description="Cadastra um novo registro em G_EMOLUMENTO_ITEM",
)
async def save(
g_emolumento_item_schema: GEmolumentoItemSaveSchema,
current_user: dict = Depends(get_current_user),
):
"""
Cria um novo registro na tabela G_EMOLUMENTO_ITEM.
"""
@ -67,11 +89,17 @@ async def save(g_emolumento_item_schema: GEmolumentoItemSaveSchema, current_user
# ----------------------------------------------------
# Atualiza um registro existente de G_EMOLUMENTO_ITEM
# ----------------------------------------------------
@router.put('/{emolumento_item_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza um registro existente em G_EMOLUMENTO_ITEM',
response_description='Atualiza um registro existente em G_EMOLUMENTO_ITEM')
async def update(emolumento_item_id: int, g_emolumento_item_update_schema: GEmolumentoItemUpdateSchema, current_user: dict = Depends(get_current_user)):
@router.put(
"/{emolumento_item_id}",
status_code=status.HTTP_200_OK,
summary="Atualiza um registro existente em G_EMOLUMENTO_ITEM",
response_description="Atualiza um registro existente em G_EMOLUMENTO_ITEM",
)
async def update(
emolumento_item_id: int,
g_emolumento_item_update_schema: GEmolumentoItemUpdateSchema,
current_user: dict = Depends(get_current_user),
):
"""
Atualiza um registro existente de G_EMOLUMENTO_ITEM com base no ID informado.
"""
@ -83,14 +111,20 @@ async def update(emolumento_item_id: int, g_emolumento_item_update_schema: GEmol
# ----------------------------------------------------
# Exclui um registro de G_EMOLUMENTO_ITEM
# ----------------------------------------------------
@router.delete('/{emolumento_item_id}',
status_code=status.HTTP_200_OK,
summary='Remove um registro de G_EMOLUMENTO_ITEM',
response_description='Remove um registro de G_EMOLUMENTO_ITEM')
async def delete(emolumento_item_id: int, current_user: dict = Depends(get_current_user)):
@router.delete(
"/{emolumento_item_id}",
status_code=status.HTTP_200_OK,
summary="Remove um registro de G_EMOLUMENTO_ITEM",
response_description="Remove um registro de G_EMOLUMENTO_ITEM",
)
async def delete(
emolumento_item_id: int, current_user: dict = Depends(get_current_user)
):
"""
Remove um registro específico de G_EMOLUMENTO_ITEM com base no ID informado.
"""
g_emolumento_item_id_schema = GEmolumentoItemIdSchema(emolumento_item_id=emolumento_item_id)
g_emolumento_item_id_schema = GEmolumentoItemIdSchema(
emolumento_item_id=emolumento_item_id
)
response = g_emolumento_item_controller.delete(g_emolumento_item_id_schema)
return response
return response

21
packages/v1/administrativo/endpoints/g_marcacao_tipo_endpoint.py
View file

@ -7,7 +7,8 @@ from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import (
GMarcacaoTipoSchema,
GMarcacaoTipoSaveSchema,
GMarcacaoTipoUpdateSchema,
GMarcacaoTipoIdSchema
GMarcacaoTipoIdSchema,
GMarcacaoTipoGrupoSchema
)
# Inicializa o roteador para as rotas do tipo de marcação
@ -47,6 +48,23 @@ async def get_by_descricao(descricao : str, current_user: dict = Depends(get_cur
return response
# Localiza um registro de marcação tipo passando os filtros de consulta
@router.get('/grupo',
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(grupo : str, sistema_id : int, situacao : 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 = GMarcacaoTipoGrupoSchema(grupo=grupo, sistema_id=sistema_id, situacao=situacao)
# Busca um registro de marcação tipo específico pela descrição
response = g_marcacao_tipo_controller.get_by_grupo(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,
@ -63,7 +81,6 @@ async def show(marcacao_tipo_id : int, current_user: dict = Depends(get_current_
# Retorna os dados localizados
return response
# Cadastro de registro de marcação tipo
@router.post('/',
status_code=status.HTTP_201_CREATED,

17
packages/v1/administrativo/endpoints/t_servico_etiqueta_endpoint.py
View file

@ -30,6 +30,23 @@ async def index(current_user: dict = Depends(get_current_user)):
return response
# Localiza um registro de servico etiqueta pelo serviço tipo
@router.get('/servico_tipo/{servico_tipo_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de servico etiqueta em específico pelo ID do serviço ',
response_description='Busca um registro de servico etiqueta em específico')
async def showServicoTipo(servico_tipo_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
servico_tipo_schema = TServicoEtiquetaServicoTipoIdSchema(servico_tipo_id=servico_tipo_id)
# Busca um registro de servico etiqueta específico pelo ID
response = t_servico_etiqueta_controller.showServicoTipo(servico_tipo_schema)
# 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,

17
packages/v1/administrativo/repositories/c_caixa_servico/c_caixa_servico_get_by_sistema_id_repository.py Normal file
View file

@ -0,0 +1,17 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.c_caixa_servico_schema import CCaixaServicoSistemaIdSchema
class ShowSistemaIdRepository(BaseRepository):
def execute(self, caixa_servico_schema : CCaixaServicoSistemaIdSchema):
# Montagem do sql
sql = """ SELECT * FROM C_CAIXA_SERVICO ccs WHERE ccs.sistema_id = :sistema_id ORDER BY ccs.descricao ASC"""
# Preenchimento de parâmetros
params = {
'sistema_id' : caixa_servico_schema.sistema_id
}
# Execução do sql
return self.fetch_all(sql, params)

5
packages/v1/administrativo/repositories/g_emolumento/g_emolumento_index_repository.py
View file

@ -1,11 +1,12 @@
from abstracts.repository import BaseRepository
class GEmolumentoIndexRepository(BaseRepository):
"""
Repositório para a operação de listagem de todos os registros
na tabela t_censec_qualidade.
"""
def execute(self):
"""
Executa a consulta SQL para buscar todos os registros.
@ -23,4 +24,4 @@ class GEmolumentoIndexRepository(BaseRepository):
response = self.fetch_all(sql)
# Retorna os dados localizados
return response
return response

76
packages/v1/administrativo/repositories/g_emolumento_item/g_emolumento_item_valor_repository.py Normal file
View file

@ -0,0 +1,76 @@
from abstracts.repository import BaseRepository
# Adaptação do Schema para GEmolumentoItem. Assumindo o uso de um Schema de ID de Sistema
# ou, mais tipicamente para itens, um Schema de ID do Emolumento pai (EMOLUMENTO_ID).
# Para manter a similaridade do original, vamos usar um 'SistemaIdSchema' adaptado,
# mas o SQL será ajustado para o campo EMOLUMENTO_ID.
from packages.v1.administrativo.schemas.g_emolumento_item_schema import GEmolumentoItemValorSchema
class ValorRepository(BaseRepository):
"""
Repositório para a operação de listagem de todos os registros
na tabela G_EMOLUMENTO_ITEM, possivelmente filtrados por EMOLUMENTO_ID.
"""
# Mantendo o padrão de nome de método do arquivo original
def execute(self, emolumento_item_schema: GEmolumentoItemValorSchema):
"""
Executa a consulta SQL para buscar todos os registros de G_EMOLUMENTO_ITEM,
incluindo todos os campos da DDL fornecida.
Returns:
Uma lista de dicionários contendo os dados dos registros.
"""
# Montagem do SQL com TODOS os campos da DDL fornecida
sql = """ SELECT VALOR_EMOLUMENTO,
EMOLUMENTO_ITEM_ID,
EMOLUMENTO_ID,
VALOR_INICIO,
VALOR_FIM,
VALOR_TAXA_JUDICIARIA,
EMOLUMENTO_PERIODO_ID,
CODIGO,
PAGINA_EXTRA,
VALOR_PAGINA_EXTRA,
VALOR_OUTRA_TAXA1,
CODIGO_SELO,
VALOR_FUNDO_RI,
CODIGO_TABELA,
SELO_GRUPO_ID,
CODIGO_KM,
EMOLUMENTO_ACRESCE,
TAXA_ACRESCE,
FUNCIVIL_ACRESCE,
VALOR_FRACAO,
VALOR_POR_EXCEDENTE_EMOL,
VALOR_POR_EXCEDENTE_TJ,
VALOR_POR_EXCEDENTE_FUNDO,
VALOR_LIMITE_EXCEDENTE_EMOL,
VALOR_LIMITE_EXCEDENTE_TJ,
VALOR_LIMITE_EXCEDENTE_FUNDO,
FUNDO_SELO,
DISTRIBUICAO,
VRCEXT
FROM G_EMOLUMENTO_ITEM
WHERE
EMOLUMENTO_ID = :emolumento_id
AND :valor BETWEEN VALOR_INICIO AND VALOR_FIM
AND EMOLUMENTO_PERIODO_ID = (
SELECT MAX(EI2.EMOLUMENTO_PERIODO_ID)
FROM G_EMOLUMENTO_ITEM EI2
WHERE EI2.EMOLUMENTO_ID = :emolumento_id
)
ORDER BY EMOLUMENTO_ITEM_ID; """
# Preenchimento de parâmetros.
# Adaptando o nome do parâmetro para refletir a coluna da DDL (EMOLUMENTO_ID).
# Assumimos que o campo 'emolumento_id' do schema está sendo usado para passar este valor.
params = {
"emolumento_id": emolumento_item_schema.emolumento_id,
"valor": emolumento_item_schema.valor
}
# Execução do sql
response = self.fetch_all(sql, params)
# Retorna os dados localizados
return response

52
packages/v1/administrativo/repositories/g_marcacao_tipo/g_marcacao_tipo_get_by_grupo_repository.py Normal file
View file

@ -0,0 +1,52 @@
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 GMarcacaoTipoGrupoSchema
class GetByGrupoRepository(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: GMarcacaoTipoGrupoSchema):
"""
Executa a consulta SQL para buscar um registro pela descrição.
Args:
marcacao_tipo_schema (GMarcacaoTipoGrupoSchema): O esquema com os filtros a serem consultados.
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 GRUPO = :grupo AND
SISTEMA_ID = :sistema_id AND
SITUACAO = :situacao """
# Preenchimento de parâmetros
params = {
'grupo': marcacao_tipo_schema.grupo,
'sistema_id': marcacao_tipo_schema.sistema_id,
'situacao': marcacao_tipo_schema.situacao
}
# Execução do sql
return self.fetch_all(sql, params)

55
packages/v1/administrativo/repositories/t_servico_etiqueta/t_servico_etiqueta_show_servico_tipo_repository.py Normal file
View file

@ -0,0 +1,55 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_servico_etiqueta_schema import TServicoEtiquetaServicoTipoIdSchema # 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_tipo_schema: TServicoEtiquetaServicoTipoIdSchema): # 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 E.SERVICO_ETIQUETA_ID,
E.ETIQUETA_MODELO_ID,
E.SERVICO_TIPO_ID,
T.DESCRICAO
FROM T_SERVICO_ETIQUETA E
LEFT JOIN G_MARCACAO_TIPO T ON E.ETIQUETA_MODELO_ID = T.MARCACAO_TIPO_ID
WHERE E.SERVICO_TIPO_ID = :servico_tipo_id """
# Preenchimento de parâmetros
params = {
# Nome do parâmetro ajustado para SERVICO_TIPO_ID
'servico_tipo_id': servico_tipo_schema.servico_tipo_id
}
# Execução do SQL
result = self.fetch_all(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)}"
)

2
packages/v1/administrativo/repositories/t_servico_tipo/t_servico_tipo_save_repository.py
View file

@ -22,7 +22,7 @@ class SaveRepository(BaseRepository):
Raises:
HTTPException: Caso ocorra um erro na execução da query.
"""
try:
try:
# Montagem do SQL
sql = """ INSERT INTO T_SERVICO_TIPO(

12
packages/v1/administrativo/repositories/t_servico_tipo/t_servico_tipo_update_repository.py
View file

@ -51,9 +51,9 @@ class UpdateRepository(BaseRepository):
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.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")
@ -115,9 +115,9 @@ class UpdateRepository(BaseRepository):
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.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")

7
packages/v1/administrativo/schemas/c_caixa_servico_schema.py
View file

@ -209,6 +209,13 @@ class CCaixaServicoIdSchema(BaseModel):
caixa_servico_id: int
# ----------------------------------------------------
# Schema para localizar um caixa serviço especifico pelo ID (GET)
# ----------------------------------------------------
class CCaixaServicoSistemaIdSchema(BaseModel):
sistema_id: int
# ----------------------------------------------------
# Schema para localizar um caixa serviço pela descrição (GET)
# ----------------------------------------------------

11
packages/v1/administrativo/schemas/g_marcacao_tipo_schema.py
View file

@ -54,7 +54,16 @@ class GMarcacaoTipoIdSchema(BaseModel):
# Schema para localizar um registro pela descrição (GET)
# ----------------------------------------------------
class GMarcacaoTipoDescricaoSchema(BaseModel):
descricao: str
descricao: str
# ----------------------------------------------------
# Schema para localizar um registro por filtro (GET)
# ----------------------------------------------------
class GMarcacaoTipoGrupoSchema(BaseModel):
grupo: str
sistema_id: int
situacao: str
# ----------------------------------------------------

141
packages/v1/administrativo/schemas/t_servico_tipo_schema.py
View file

@ -70,115 +70,66 @@ class TServicoTipoDescricaoSchema(BaseModel):
# ----------------------------------------------------
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)
descricao: str # Obrigatório
valor: Optional[float] = None # Numeric(14,3)
tipo_item: Optional[str] = None # Varchar(1)
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)
# Sanitização dos campos string (para prevenir XSS e entradas inválidas)
@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',
'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)
# Validação de obrigatoriedade apenas para o campo 'descricao'
@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:
def validate_required_fields(self):
if not self.descricao or not self.descricao.strip():
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
detail=[{'input': 'descricao', 'message': 'A descrição é obrigatória.'}]
)
return self
# ----------------------------------------------------
# Schema para atualizar SERVICO (PUT)
# ----------------------------------------------------

17
packages/v1/administrativo/services/c_caixa_servico/go/c_caixa_servico_get_sistema_id_service.py Normal file
View file

@ -0,0 +1,17 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.schemas.c_caixa_servico_schema import CCaixaServicoSistemaIdSchema
from packages.v1.administrativo.actions.c_caixa_servico.c_caixa_servico_get_by_sistema_id_action import ShowSistemaIdAction
class GetSistemaIdService:
def execute(self, caixa_servico_schema: CCaixaServicoSistemaIdSchema):
# Instânciamento de ação
show_sistema_id_action = ShowSistemaIdAction()
# Executa a ação em questão
data = show_sistema_id_action.execute(caixa_servico_schema)
# Retorno da informação
return data

8
packages/v1/administrativo/services/g_emolumento/go/g_emolumento_index_service.py
View file

@ -1,4 +1,6 @@
from api.packages.v1.administrativo.actions.g_emolumento.g_emolumento_index_action import GEmolumentoIndexAction
from api.packages.v1.administrativo.actions.g_emolumento.g_emolumento_index_action import (
GEmolumentoIndexAction,
)
from fastapi import HTTPException, status
@ -13,7 +15,7 @@ class GEmolumentoIndexService:
Executa a operação de busca de todos os registros no banco de dados.
Args:
g_emolumento_index_schema (GEmolumentoIndexSchema):
g_emolumento_index_schema (GEmolumentoIndexSchema):
Esquema que pode conter filtros ou parâmetros de busca.
Returns:
@ -35,7 +37,7 @@ class GEmolumentoIndexService:
if not data:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Não foi possível localizar registros de G_EMOLUMENTO."
detail="Não foi possível localizar registros de G_EMOLUMENTO.",
)
# ----------------------------------------------------

38
packages/v1/administrativo/services/g_emolumento_item/go/g_emolumento_item_get_by_valor_service.py Normal file
View file

@ -0,0 +1,38 @@
from fastapi import HTTPException, status
# Adaptação do Schema
from packages.v1.administrativo.schemas.g_emolumento_item_schema import GEmolumentoItemValorSchema
# Importação da Action ajustada para o novo prefixo
from packages.v1.administrativo.actions.g_emolumento_item.g_emolumento_item_valor_action import ValorAction
# O nome da classe deve ser adaptado com o prefixo 'GEmolumentoItem' para manter o padrão
# de classes da aplicação no Controller, Repository, etc.
class ValorService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de registros na tabela G_EMOLUMENTO_ITEM.
"""
# Mantendo o padrão de nome de método do arquivo original
def execute(self, emolumento_item_schema: GEmolumentoItemValorSchema):
"""
Executa a operação de busca de todos os registros de G_EMOLUMENTO_ITEM no banco de dados.
Returns:
A lista de registros encontrados.
"""
# Instanciamento da ação (com o prefixo adaptado)
valor_action = ValorAction()
# Executa a busca de todas as ações (adaptando o nome do parâmetro)
data = valor_action.execute(emolumento_item_schema)
# Verifica se foram localizados registros
if not data:
# Retorna uma exceção (adaptando a mensagem)
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail='Não foi possível localizar os registros de G_EMOLUMENTO_ITEM'
)
# Retorna as informações localizadas
return data

38
packages/v1/administrativo/services/g_marcacao_tipo/go/g_marcacao_tipo_get_by_grupo_service.py Normal file
View file

@ -0,0 +1,38 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.schemas.g_marcacao_tipo_schema import GMarcacaoTipoGrupoSchema
from packages.v1.administrativo.actions.g_marcacao_tipo.g_marcacao_tipo_get_by_grupo_action import GetByGrupoAction
class GetByGrupoService:
"""
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: GMarcacaoTipoGrupoSchema, messageValidate: bool):
"""
Executa a operação de busca no banco de dados.
Args:
marcacao_tipo_schema (GMarcacaoTipoGrupoSchema): 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 = GetByGrupoAction()
# 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

38
packages/v1/administrativo/services/t_servico_etiqueta/go/t_servico_etiqueta_show_servico_tipo_service.py Normal file
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 TServicoEtiquetaServicoTipoIdSchema
# Importação da Action ajustada para ShowAction
from packages.v1.administrativo.actions.t_servico_etiqueta.t_servico_etiqueta_show_servico_tipo_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: TServicoEtiquetaServicoTipoIdSchema): # Nome do parâmetro e tipo ajustados
"""
Executa a operação de busca no banco de dados.
Args:
servico_etiqueta_schema (TServicoEtiquetaServicoTipoIdSchema): 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

16
packages/v1/api.py
View file

@ -50,6 +50,8 @@ from packages.v1.servicos.balcao.endpoints import (
t_servico_itempedido_endpoint,
t_servico_pedido_endpoint,
)
from packages.v1.administrativo.endpoints import g_emolumento_endpoint
from packages.v1.administrativo.endpoints import g_emolumento_item_endpoint
# Cria uma instância do APIRouter que vai agregar todas as rotas da API
api_router = APIRouter()
@ -351,3 +353,17 @@ api_router.include_router(
prefix="/servicos/balcao/t_servico_itempedido",
tags=["Itens do Pedido"],
)
# Inclui as rotas de g_emolumento
api_router.include_router(
g_emolumento_endpoint.router,
prefix="/administrativo/g_emolumento",
tags=["Emolumentos"],
)
# Inclui as rotas de g_emolumento_item
api_router.include_router(
g_emolumento_item_endpoint.router,
prefix="/administrativo/g_emolumento_item",
tags=["Emolumento Item"],
)