Merge branch 'release(MVP/Sprint3)' into release(MVP)

This commit is contained in:
Kenio 2025-09-15 13:01:15 -03:00
commit 4a3316a21b
23 changed files with 981 additions and 1 deletions

5
.gitignore vendored
View file

@ -38,4 +38,7 @@ build/
pip-wheel-metadata/
*.egg
.cache/
.tox/
.tox/
# Arquivo s de conexão
config/database/firebird.json

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -0,0 +1,27 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.g_tb_profissao_schema import GTbProfissaoUpdateSchema
from packages.v1.administrativo.repositories.g_tb_profissao.g_tb_profissao_update_repository import UpdateRepository
class GTbProfissaoUpdateAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela G_TB_PROFISSAO.
"""
def execute(self, tb_profissao_id: float, profissao_schema: GTbProfissaoUpdateSchema):
"""
Executa a operação de atualização.
Args:
tb_profissao_id (float): O ID da profissão a ser atualizada.
profissao_schema (GTbProfissaoUpdateSchema): O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# Instância o repositório de atualização
update_repository = UpdateRepository()
# Chama o método de execução do repositório para realizar a atualização
return update_repository.execute(tb_profissao_id, profissao_schema)

View file

@ -0,0 +1,113 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.g_tb_profissao_schema import (
GTbProfissaoSchema,
GTbProfissaoSaveSchema,
GTbProfissaoUpdateSchema,
GTbProfissaoIdSchema,
GTbProfissaoDescricaoSchema
)
class GTbProfissaoController:
def __init__(self):
# Action responsável por carregar as services de acordo com o estado
self.dynamic_import = DynamicImport()
# Define o pacote que deve ser carregado
self.dynamic_import.set_package("administrativo")
# Define a tabela que o pacote pertence
self.dynamic_import.set_table("g_tb_profissao")
pass
# Lista todas as profissões
def index(self):
# Importação da classe desejada
indexService = self.dynamic_import.service("g_tb_profissao_index_service", "IndexService")
# Instância da classe service
self.indexService = indexService()
# Lista todas as profissões
return {
'message': 'Profissões localizadas com sucesso',
'data': self.indexService.execute()
}
# Busca uma profissão específica pelo ID
def show(self, profissao_schema: GTbProfissaoIdSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('g_tb_profissao_show_service', 'ShowService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna a profissão desejada
return {
'message' : 'Profissão localizada com sucesso',
'data': self.show_service.execute(profissao_schema)
}
# Busca uma profissão pela descrição
def get_by_descricao(self, profissao_schema: GTbProfissaoDescricaoSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('g_tb_profissao_get_descricao_service', 'GetByDescricaoService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna a profissão desejada
return {
'message' : 'Profissão localizada com sucesso',
'data': self.show_service.execute(profissao_schema, True) #True para retornar a mensagem de erro caso não localize o serviço
}
# Cadastra uma nova profissão
def save(self, profissao_schema: GTbProfissaoSaveSchema):
#Importação da classe desejada
save_service = self.dynamic_import.service('g_tb_profissao_save_service', 'GTbProfissaoSaveService')
# Instância da classe desejada
self.save_service = save_service()
# Busca e retorna a profissão desejada
return {
'message' : 'Profissão salva com sucesso',
'data': self.save_service.execute(profissao_schema)
}
# Atualiza os dados de uma profissão
def update(self, tb_profissao_id: float, profissao_schema: GTbProfissaoUpdateSchema):
#Importação da classe desejada
update_service = self.dynamic_import.service('g_tb_profissao_update_service', 'GTbProfissaoUpdateService')
# Instância da classe desejada
self.update_service = update_service()
# Busca e retorna a profissão desejada
return {
'message' : 'Profissão atualizada com sucesso',
'data': self.update_service.execute(tb_profissao_id, profissao_schema)
}
# Exclui uma profissão
def delete(self, profissao_schema: GTbProfissaoIdSchema):
#Importação da classe desejada
delete_service = self.dynamic_import.service('g_tb_profissao_delete_service', 'DeleteService')
# Instância da classe desejada
self.delete_service = delete_service()
# Busca e retorna a profissão desejada
return {
'message' : 'Profissão removida com sucesso',
'data': self.delete_service.execute(profissao_schema)
}

View file

@ -0,0 +1,110 @@
# Importação de bibliotecas
from typing import Optional
from fastapi import APIRouter, Body, Depends, status
from actions.jwt.get_current_user import get_current_user
from packages.v1.administrativo.controllers.g_tb_profissao_controller import GTbProfissaoController
from packages.v1.administrativo.schemas.g_tb_profissao_schema import (
GTbProfissaoSchema,
GTbProfissaoSaveSchema,
GTbProfissaoUpdateSchema,
GTbProfissaoIdSchema,
GTbProfissaoDescricaoSchema
)
# Inicializa o roteador para as rotas da profissão
router = APIRouter()
# Instanciamento do controller desejado
g_tb_profissao_controller = GTbProfissaoController()
# Lista todas as profissões
@router.get('/',
status_code=status.HTTP_200_OK,
summary='Lista todas as profissões cadastradas',
response_description='Lista todas as profissões cadastradas')
async def index(current_user: dict = Depends(get_current_user)):
"""
Endpoint que retorna uma lista de todas as profissões cadastradas.
"""
# Busca todas as profissões cadastradas
response = g_tb_profissao_controller.index()
return response
# Localiza uma profissão pela descrição
@router.get('/descricao',
status_code=status.HTTP_200_OK,
summary='Busca um registro de profissão em específico pela descrição',
response_description='Busca um registro de profissão em específico')
async def get_by_descricao(descricao : str, current_user: dict = Depends(get_current_user)):
"""
Endpoint que busca uma profissão específica usando a descrição.
"""
# Cria o schema com os dados recebidos
profissao_schema = GTbProfissaoDescricaoSchema(descricao=descricao)
# Busca a profissão específica pela descrição
response = g_tb_profissao_controller.get_by_descricao(profissao_schema)
return response
# Localiza uma profissão pelo ID
@router.get('/{tb_profissao_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de profissão em específico pelo ID',
response_description='Busca um registro de profissão em específico')
async def show(tb_profissao_id : float, current_user: dict = Depends(get_current_user)):
"""
Endpoint que busca uma profissão específica pelo ID.
"""
# Cria o schema com os dados recebidos
profissao_schema = GTbProfissaoIdSchema(tb_profissao_id=tb_profissao_id)
# Busca a profissão específica pelo ID
response = g_tb_profissao_controller.show(profissao_schema)
return response
# Cadastro de profissão
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra uma nova profissão',
response_description='Cadastra uma nova profissão')
async def save(profissao_schema : GTbProfissaoSaveSchema, current_user: dict = Depends(get_current_user)):
"""
Endpoint para cadastrar uma nova profissão no banco de dados.
"""
# Efetua o cadastro no banco de dados
response = g_tb_profissao_controller.save(profissao_schema)
return response
# Atualiza os dados de uma profissão
@router.put('/{tb_profissao_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza os dados de uma profissão',
response_description='Atualiza os dados de uma profissão')
async def update(tb_profissao_id : float, profissao_schema : GTbProfissaoUpdateSchema, current_user: dict = Depends(get_current_user)):
"""
Endpoint para atualizar os dados de uma profissão existente.
"""
# Efetua a atualização dos dados
response = g_tb_profissao_controller.update(tb_profissao_id, profissao_schema)
return response
# Exclui uma profissão
@router.delete('/{tb_profissao_id}',
status_code=status.HTTP_200_OK,
summary='Remove uma profissão',
response_description='Remove uma profissão')
async def delete(tb_profissao_id : float, current_user: dict = Depends(get_current_user)):
"""
Endpoint para excluir uma profissão pelo ID.
"""
# Cria o schema com os dados recebidos
profissao_schema = GTbProfissaoIdSchema(tb_profissao_id=tb_profissao_id)
# Efetua a exclusão da profissão
response = g_tb_profissao_controller.delete(profissao_schema)
return response

View file

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

View file

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

View file

@ -0,0 +1,23 @@
from abstracts.repository import BaseRepository
class IndexRepository(BaseRepository):
"""
Repositório para a operação de listagem de todos os registros
na tabela G_TB_PROFISSAO.
"""
def execute(self):
"""
Executa a consulta SQL para buscar todos os registros.
Returns:
Uma lista de dicionários contendo os dados dos registros.
"""
# Montagem do SQL
sql = """ SELECT * FROM G_TB_PROFISSAO """
# Execução do sql
response = self.fetch_all(sql)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,56 @@
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.g_tb_profissao_schema import GTbProfissaoSaveSchema
class SaveRepository(BaseRepository):
"""
Repositório para a operação de salvamento de um novo registro na tabela G_TB_PROFISSAO.
"""
def execute(self, profissao_schema: GTbProfissaoSaveSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
profissao_schema (GTbProfissaoSaveSchema): O esquema com os dados a serem salvos.
Returns:
O registro recém-criado.
Raises:
HTTPException: Caso ocorra um erro na execução da query.
"""
try:
# Montagem do SQL
sql = """ INSERT INTO G_TB_PROFISSAO(
TB_PROFISSAO_ID,
DESCRICAO,
SITUACAO,
COD_CBO
) VALUES (
:tb_profissao_id,
:descricao,
:situacao,
:cod_cbo
) RETURNING *;"""
# Preenchimento de parâmetros
params = {
'tb_profissao_id': profissao_schema.tb_profissao_id,
'descricao': profissao_schema.descricao,
'situacao': profissao_schema.situacao,
'cod_cbo': profissao_schema.cod_cbo
}
# Execução do sql
return self.run_and_return(sql, params)
except Exception as e:
# Informa que houve uma falha no salvamento do registro
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao salvar profissão: {e}"
)

View file

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

View file

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

View file

@ -0,0 +1,101 @@
from pydantic import BaseModel, field_validator, model_validator
from fastapi import HTTPException, status
from typing import Optional
# Funções para sanitização de entradas (evitar XSS, SQLi etc.)
from actions.validations.text import Text
# ----------------------------------------------------
# Schema base para G_TB_PROFISSAO
# ----------------------------------------------------
class GTbProfissaoSchema(BaseModel):
tb_profissao_id: Optional[float] = None # Numeric(10,2) pode ser representado por float em Python
descricao: Optional[str] = None
situacao: Optional[str] = None
cod_cbo: Optional[str] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar uma profissão específica pelo ID (GET)
# ----------------------------------------------------
class GTbProfissaoIdSchema(BaseModel):
tb_profissao_id: float
# ----------------------------------------------------
# Schema para localizar uma profissão específica pela descrição (GET)
# ----------------------------------------------------
class GTbProfissaoDescricaoSchema(BaseModel):
descricao: str
# ----------------------------------------------------
# Schema para criação de nova profissão (POST)
# ----------------------------------------------------
class GTbProfissaoSaveSchema(BaseModel):
tb_profissao_id: Optional[float] = None
descricao: str
situacao: str
cod_cbo: Optional[str] = None
@field_validator('descricao', 'situacao', 'cod_cbo')
def sanitize_fields(cls, v):
if v is not None:
return Text.sanitize_input(v)
return v
@model_validator(mode='after')
def validate_all_fields(self):
errors = []
if not self.descricao or len(self.descricao.strip()) == 0:
errors.append({'input': 'descricao', 'message': 'A descrição é obrigatória.'})
if not self.situacao or len(self.situacao.strip()) == 0:
errors.append({'input': 'situacao', 'message': 'A situação é obrigatória.'})
elif len(self.situacao) != 1: # Validação adicional para o campo SITUACAO ser de 1 caractere
errors.append({'input': 'situacao', 'message': 'A situação deve conter apenas um caractere.'})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self
# ----------------------------------------------------
# Schema para atualizar profissão (PUT)
# ----------------------------------------------------
class GTbProfissaoUpdateSchema(BaseModel):
descricao: Optional[str] = None
situacao: Optional[str] = None
cod_cbo: Optional[str] = None
@field_validator('descricao', 'situacao', 'cod_cbo')
def sanitize_fields(cls, v):
if v is not None:
return Text.sanitize_input(v)
return v
@model_validator(mode='after')
def validate_all_fields(self):
errors = []
# Ao atualizar, os campos não são estritamente obrigatórios,
# mas se forem preenchidos, devem ser válidos.
if self.descricao is not None and len(self.descricao.strip()) == 0:
errors.append({'input': 'descricao', 'message': 'A descrição não pode ser vazia se informada.'})
if self.situacao is not None and len(self.situacao.strip()) == 0:
errors.append({'input': 'situacao', 'message': 'A situação não pode ser vazia se informada.'})
elif self.situacao is not None and len(self.situacao) != 1:
errors.append({'input': 'situacao', 'message': 'A situação deve conter apenas um caractere se informada.'})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self

View file

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

View file

@ -0,0 +1,38 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.schemas.g_tb_profissao_schema import GTbProfissaoDescricaoSchema
from packages.v1.administrativo.actions.g_tb_profissao.g_tb_profissao_get_by_descricao_action import GTbProfissaoGetByDescricaoAction
class GetByDescricaoService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela g_tb_profissao pela sua descrição.
"""
def execute(self, profissao_schema: GTbProfissaoDescricaoSchema, messageValidate: bool):
"""
Executa a operação de busca no banco de dados.
Args:
profissao_schema (GTbProfissaoDescricaoSchema): 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 = GTbProfissaoGetByDescricaoAction()
# Executa a ação em questão
data = show_action.execute(profissao_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 profissão'
)
# Retorno da informação
return data

View file

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

View file

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

View file

@ -0,0 +1,35 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.schemas.g_tb_profissao_schema import GTbProfissaoIdSchema
from packages.v1.administrativo.actions.g_tb_profissao.g_tb_profissao_show_action import GTbProfissaoShowAction
class ShowService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela G_TB_PROFISSAO.
"""
def execute(self, profissao_schema: GTbProfissaoIdSchema):
"""
Executa a operação de busca no banco de dados.
Args:
profissao_schema (GTbProfissaoIdSchema): O esquema com o ID a ser buscado.
Returns:
O resultado da busca.
"""
# Instanciamento da ação
show_action = GTbProfissaoShowAction()
# Executa a ação em questão
data = show_action.execute(profissao_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 profissão'
)
# Retorno da informação
return data

View file

@ -0,0 +1,24 @@
from packages.v1.administrativo.schemas.g_tb_profissao_schema import GTbProfissaoUpdateSchema
from packages.v1.administrativo.actions.g_tb_profissao.g_tb_profissao_update_action import GTbProfissaoUpdateAction
class GTbProfissaoUpdateService:
"""
Serviço para a operação de atualização de um registro na tabela
G_TB_PROFISSAO.
"""
def execute(self, tb_profissao_id : float, profissao_schema: GTbProfissaoUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
tb_profissao_id (float): O ID da profissão a ser atualizada.
profissao_schema (GTbProfissaoUpdateSchema): O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# Instanciamento de ações
update_action = GTbProfissaoUpdateAction()
# Retorna o resultado da operação
return update_action.execute(tb_profissao_id, profissao_schema)

View file

@ -2,6 +2,9 @@
from fastapi import APIRouter
# Importa os módulos de rotas específicos
from packages.v1.administrativo.endpoints import c_caixa_item
from packages.v1.administrativo.endpoints import g_usuario
from packages.v1.administrativo.endpoints import g_tb_profissao
from packages.v1.administrativo.endpoints import c_caixa_item_endpoint
from packages.v1.administrativo.endpoints import g_usuario_endpoint
from packages.v1.administrativo.endpoints import c_caixa_servico_endpoint
@ -36,3 +39,7 @@ api_router.include_router(
t_tb_andamentoservico_endpoint.router, prefix="/administrativo/t_tb_andamentoservico", tags=["Andamentos de serviço"]
)
# Inclui as rotas de g_tb_profissao
api_router.include_router(
g_tb_profissao.router, prefix="/administrativo/g_tb_profissao", tags=["Profissões"]
)