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

Merge branch 'MVPTN-106' into release(MVP/Sprint6)

Browse source
This commit is contained in:
Kenio 2025-10-28 15:47:33 -03:00
commit 9ab315d273
37 changed files with 1318 additions and 111 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

27
packages/v1/administrativo/actions/g_emolumento/g_emolumento_index_action.py Normal file
View file

@ -0,0 +1,27 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoSistemaIdSchema
# O repositório IndexRepository deve ser substituído pelo GEmolumentoIndexRepository.
from packages.v1.administrativo.repositories.g_emolumento.g_emolumento_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_EMOLUMENTO.
"""
def execute(self, emolumento_schema: GEmolumentoSistemaIdSchema):
"""
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(emolumento_schema)
# Retorno da informação
return response

44
packages/v1/administrativo/actions/g_emolumento_item/g_emolumento_item_index_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 GEmolumentoItemEmolumentoIdSchema
# O repositório IndexRepository deve ser substituído pelo GEmolumentoItemIndexRepository.
from packages.v1.administrativo.repositories.g_emolumento_item.g_emolumento_item_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_EMOLUMENTO_ITEM,
utilizando a DDL fornecida.
"""
# Mantendo o padrão de nome de método do arquivo original
def execute(self, emolumento_item_schema: GEmolumentoItemEmolumentoIdSchema):
"""
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 = IndexRepository()
# 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

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

30
packages/v1/administrativo/controllers/g_emolumento_controller.py Normal file
View file

@ -0,0 +1,30 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoSistemaIdSchema
class GEmolumentoController:
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_emolumento")
pass
# Lista todos os registros de emolumento por sistema
def index(self, emolumento_schema: GEmolumentoSistemaIdSchema):
# Importação da classe desejada
indexService = self.dynamic_import.service("g_emolumento_index_service", "IndexService")
# Instância da classe service
self.indexService = indexService()
# Lista todos os registros de emolumento
return {
'message': 'Registros de emolumento localizados com sucesso',
'data': self.indexService.execute(emolumento_schema)
}

53
packages/v1/administrativo/controllers/g_emolumento_item_controller.py Normal file
View file

@ -0,0 +1,53 @@
from actions.dynamic_import.dynamic_import import DynamicImport
# O Schema precisa ser adaptado para GEmolumentoItem.
# Assumindo o padrão de nomenclatura anterior:
from packages.v1.administrativo.schemas.g_emolumento_item_schema import (
GEmolumentoItemEmolumentoIdSchema,
GEmolumentoItemValorSchema
)
class GEmolumentoItemController:
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 (adaptado para g_emolumento_item)
self.dynamic_import.set_table("g_emolumento_item")
pass
# Lista todos os registros de item de emolumento por Emolumento ID
def index(self, emolumento_item_schema: GEmolumentoItemEmolumentoIdSchema):
# Importação da classe desejada (adaptado o nome da service)
indexService = self.dynamic_import.service("g_emolumento_item_index_service", "IndexService")
# Instância da classe service
self.indexService = indexService()
# Lista todos os registros de item de emolumento
# O nome do parâmetro foi adaptado para 'emolumento_item_schema'
return {
'message': 'Registros de item de emolumento localizados com sucesso',
'data': self.indexService.execute(emolumento_item_schema)
}
# Lista todos os registros de item de emolumento por Emolumento ID e faixa de valor
def get_by_valor(self, emolumento_item_schema: GEmolumentoItemValorSchema):
# Importação da classe desejada (adaptado o nome da service)
valorService = self.dynamic_import.service("g_emolumento_item_get_by_valor_service", "ValorService")
# Instância da classe service
self.valorService = valorService()
# Lista todos os registros de item de emolumento
# O nome do parâmetro foi adaptado para 'emolumento_item_schema'
return {
'message': 'Registros de item de emolumento localizados com sucesso',
'data': self.valorService.execute(emolumento_item_schema)
}

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,

28
packages/v1/administrativo/endpoints/g_emolumento_endpoint.py Normal file
View file

@ -0,0 +1,28 @@
# 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_emolumento_controller import GEmolumentoController
from packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoSistemaIdSchema
# Inicializa o roteador para as rotas do emolumento
router = APIRouter()
# Instanciamento do controller desejado
g_emolumento_controller = GEmolumentoController()
# Localiza um registro de emolumento pelo ID
@router.get('/{sistema_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de emolumento em específico pelo ID',
response_description='Busca um registro de emolumento em específico')
async def index(sistema_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
emolumento_schema = GEmolumentoSistemaIdSchema(sistema_id=sistema_id)
# Busca um registro de emolumento específico pelo ID
response = g_emolumento_controller.index(emolumento_schema)
# Retorna os dados localizados
return response

50
packages/v1/administrativo/endpoints/g_emolumento_item_endpoint.py Normal file
View file

@ -0,0 +1,50 @@
# 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
# Importações adaptadas para G_EMOLUMENTO_ITEM
from packages.v1.administrativo.controllers.g_emolumento_item_controller import GEmolumentoItemController
from packages.v1.administrativo.schemas.g_emolumento_item_schema import (
GEmolumentoItemEmolumentoIdSchema,
GEmolumentoItemValorSchema
)
# Inicializa o roteador para as rotas do item de emolumento
# Poderia ser renomeado para 'router_item' ou similar se houver chance de conflito de namespace
router = APIRouter()
# Instanciamento do controller desejado
g_emolumento_item_controller = GEmolumentoItemController()
# Localiza registros de item de emolumento pelo emolumento_id e faixa de valores
@router.get('/faixa/{emolumento_id}/{valor}',
status_code=status.HTTP_200_OK,
summary='Busca registros de item de emolumento em específico, possivelmente filtrando pelo ID do sistema/pai',
response_description='Busca registros de item de emolumento em específico')
async def get_by_valor(emolumento_id : int, valor : float, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos (adaptado para GEmolumentoItem)
emolumento_item_schema = GEmolumentoItemValorSchema(emolumento_id=emolumento_id, valor=valor)
# Busca os registros de item de emolumento (adaptado para o novo controller)
response = g_emolumento_item_controller.get_by_valor(emolumento_item_schema)
# Retorna os dados localizados
return response
# Localiza registros de item de emolumento pelo emolumento_id
@router.get('/{emolumento_id}',
status_code=status.HTTP_200_OK,
summary='Busca registros de item de emolumento em específico, possivelmente filtrando pelo ID do sistema/pai',
response_description='Busca registros de item de emolumento em específico')
async def index(emolumento_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos (adaptado para GEmolumentoItem)
emolumento_item_schema = GEmolumentoItemEmolumentoIdSchema(emolumento_id=emolumento_id)
# Busca os registros de item de emolumento (adaptado para o novo controller)
response = g_emolumento_item_controller.index(emolumento_item_schema)
# Retorna os dados localizados
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)

48
packages/v1/administrativo/repositories/g_emolumento/g_emolumento_index_repository.py Normal file
View file

@ -0,0 +1,48 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoSistemaIdSchema
class IndexRepository(BaseRepository):
"""
Repositório para a operação de listagem de todos os registros
na tabela G_EMOLUMENTO.
"""
def execute(self, emolumento_schema: GEmolumentoSistemaIdSchema):
"""
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_EMOLUMENTO
sql = """ SELECT EMOLUMENTO_ID,
DESCRICAO,
TIPO,
SISTEMA_ID,
SELO_GRUPO_ID,
REG_AVERB,
PRE_DEFINIDO,
SITUACAO,
SITUACAO_RI,
COM_REDUCAO,
MOTIVO_REDUCAO,
VALOR_MAXIMO_CERTIDAO,
TIPO_OBJETIVO,
MODELO_TAG,
CODIGO_NOTA_ID,
CONVENIO_CODHAB,
ITEM_DF
FROM G_EMOLUMENTO
WHERE SISTEMA_ID = :sistema_id AND (SITUACAO = 'A' OR SITUACAO IS NULL) ORDER BY DESCRICAO """
# Preenchimento de parâmetros
params = {
"sistema_id": emolumento_schema.sistema_id
}
# Execução do sql
response = self.fetch_all(sql, params)
# Retorna os dados localizados
return response

68
packages/v1/administrativo/repositories/g_emolumento_item/g_emolumento_item_index_repository.py Normal file
View file

@ -0,0 +1,68 @@
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 GEmolumentoItemEmolumentoIdSchema
class IndexRepository(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: GEmolumentoItemEmolumentoIdSchema):
"""
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
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
}
# Execução do sql
response = self.fetch_all(sql, params)
# Retorna os dados localizados
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)
# ----------------------------------------------------

75
packages/v1/administrativo/schemas/g_emolumento_item_schema.py Normal file
View file

@ -0,0 +1,75 @@
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 para G_EMOLUMENTO_ITEM
# ----------------------------------------------------
class GEmolumentoItemSchema(BaseModel):
# Colunas NUMERIC(14,3) devem ser mapeadas para Decimal para precisão
valor_emolumento: Optional[Decimal] = None
# Coluna NUMERIC(10,2) PK
emolumento_item_id: Optional[int] = None
# Colunas NUMERIC(10,2) FK ou comuns
emolumento_id: Optional[int] = None
emolumento_periodo_id: Optional[int] = None
codigo: Optional[int] = None
pagina_extra: Optional[int] = None
selo_grupo_id: Optional[int] = None
vrc_ext: Optional[int] = None # VRCEXT NUMERIC(10,2)
# Colunas NUMERIC(14,3)
valor_inicio: Optional[Decimal] = None
valor_fim: Optional[Decimal] = None
valor_taxa_judiciaria: Optional[Decimal] = None
valor_pagina_extra: Optional[Decimal] = None
valor_outra_taxa1: Optional[Decimal] = None
valor_fundo_ri: Optional[Decimal] = None
emolumento_acresce: Optional[Decimal] = None
taxa_acresce: Optional[Decimal] = None
funcivil_acresce: Optional[Decimal] = None
valor_fracao: Optional[Decimal] = None
valor_por_excedente_emol: Optional[Decimal] = None
valor_por_excedente_tj: Optional[Decimal] = None
valor_por_excedente_fundo: Optional[Decimal] = None
valor_limite_excedente_emol: Optional[Decimal] = None
valor_limite_excedente_tj: Optional[Decimal] = None
valor_limite_excedente_fundo: Optional[Decimal] = None
fundo_selo: Optional[Decimal] = None
distribuicao: Optional[Decimal] = None
# Colunas VARCHAR
codigo_selo: Optional[str] = None # VARCHAR(30)
codigo_tabela: Optional[str] = None # VARCHAR(30)
codigo_km: Optional[str] = None # VARCHAR(30)
# Sanitiza campos de texto (apenas os VARCHARs da DDL)
@field_validator(
'codigo_selo', 'codigo_tabela', 'codigo_km'
)
def sanitize_fields(cls, v):
if isinstance(v, str) and v:
return Text.sanitize_input(v)
return v
# ----------------------------------------------------
# Schema para localizar um registro pelo emolumento_id (GET)
# ----------------------------------------------------
class GEmolumentoItemEmolumentoIdSchema(BaseModel):
emolumento_id: int
# ----------------------------------------------------
# Schema para localizar um registro pelo emolumento_id e valor (GET)
# ----------------------------------------------------
class GEmolumentoItemValorSchema(BaseModel):
emolumento_id: int
valor: float

48
packages/v1/administrativo/schemas/g_emolumento_schema.py Normal file
View file

@ -0,0 +1,48 @@
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 GEmolumentoSchema(BaseModel):
emolumento_id: Optional[int] = None # NUMERIC(10,2) PK
descricao: Optional[str] = None # VARCHAR(260)
tipo: Optional[str] = None # VARCHAR(1)
sistema_id: Optional[int] = None # NUMERIC(10,2)
selo_grupo_id: Optional[int] = None # NUMERIC(10,2)
reg_averb: Optional[str] = None # VARCHAR(1)
pre_definido: Optional[str] = None # VARCHAR(1)
situacao: Optional[str] = None # VARCHAR(1)
situacao_ri: Optional[str] = None # VARCHAR(1)
com_reducao: Optional[str] = None # VARCHAR(1)
motivo_reducao: Optional[str] = None # VARCHAR(120)
valor_maximo_certidao: Optional[Decimal] = None # NUMERIC(14,3)
tipo_objetivo: Optional[str] = None # VARCHAR(3)
modelo_tag: Optional[str] = None # VARCHAR(3)
codigo_nota_id: Optional[int] = None # NUMERIC(10,2)
convenio_codhab: Optional[str] = None # VARCHAR(1)
item_df: Optional[str] = None # VARCHAR(10)
# Sanitiza campos de texto
@field_validator(
'descricao', 'tipo', 'reg_averb', 'pre_definido', 'situacao',
'situacao_ri', 'com_reducao', 'motivo_reducao', 'tipo_objetivo',
'modelo_tag', 'convenio_codhab', 'item_df'
)
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 GEmolumentoSistemaIdSchema(BaseModel):
sistema_id: int

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

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

@ -0,0 +1,34 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.schemas.g_emolumento_schema import GEmolumentoSistemaIdSchema
# Importação da Action ajustada para o novo prefixo
from packages.v1.administrativo.actions.g_emolumento.g_emolumento_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_EMOLUMENTO.
"""
def execute(self, emolumento_schema: GEmolumentoSistemaIdSchema):
"""
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(emolumento_schema)
# 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_EMOLUMENTO'
)
# Retorna as informações localizadas
return data

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_emolumento_item/go/g_emolumento_item_index_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 GEmolumentoItemEmolumentoIdSchema
# Importação da Action ajustada para o novo prefixo
from packages.v1.administrativo.actions.g_emolumento_item.g_emolumento_item_index_action import IndexAction
# 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 IndexService:
"""
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: GEmolumentoItemEmolumentoIdSchema):
"""
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)
index_action = IndexAction()
# Executa a busca de todas as ações (adaptando o nome do parâmetro)
data = index_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

12
packages/v1/api.py
View file

@ -32,6 +32,8 @@ from packages.v1.administrativo.endpoints import t_imovel_endpoint
from packages.v1.administrativo.endpoints import t_imovel_unidade_endpoint
from packages.v1.administrativo.endpoints import t_pessoa_endpoint
from packages.v1.administrativo.endpoints import t_pessoa_representante_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()
@ -188,4 +190,14 @@ api_router.include_router(
# Inclui as rotas de t_imovel_unidade
api_router.include_router(
t_censec_qualidadeato_endpoint.router, prefix="/administrativo/t_censec_qualidadeato", tags=["CENSEC Qualidade Ato"]
)
# 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"]
)