[MVPTN-19] feat(Crud): CRUD completo para a tabela g_cidade

This commit is contained in:
Kenio 2025-09-26 11:38:25 -03:00
parent 5bf465c218
commit 2b4596818b
22 changed files with 993 additions and 0 deletions

View file

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

View file

@ -0,0 +1,29 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeNomeSchema
from packages.v1.administrativo.repositories.g_cidade.g_cidade_get_by_nome_repository import GetByNomeRepository
class GetByNomeAction(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_CIDADE por nome (CIDADE_NOME).
"""
def execute(self, g_cidade_schema: GCidadeNomeSchema):
"""
Executa a operação de busca no banco de dados.
Args:
g_cidade_schema (GCidadeNomeSchema): O esquema com o nome da cidade a ser buscada.
Returns:
O registro encontrado ou None.
"""
# Instanciamento do repositório
show_repository = GetByNomeRepository()
# Execução do repositório
response = show_repository.execute(g_cidade_schema)
# Retorno da informação
return response

View file

@ -0,0 +1,24 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.repositories.g_cidade.g_cidade_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_CIDADE.
"""
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_cidade_schema import GCidadeSaveSchema
from packages.v1.administrativo.repositories.g_cidade.g_cidade_save_repository import SaveRepository
class SaveAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvar (inserir ou atualizar) um registro na tabela G_CIDADE.
"""
def execute(self, g_cidade_schema: GCidadeSaveSchema):
"""
Executa a operação de salvamento.
Args:
g_cidade_schema (GCidadeSaveSchema): 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(g_cidade_schema)
# Retorno da informação
return response

View file

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

View file

@ -0,0 +1,26 @@
from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeUpdateSchema
from packages.v1.administrativo.repositories.g_cidade.g_cidade_update_repository import UpdateRepository
class UpdateAction:
"""
Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela G_CIDADE.
"""
def execute(self, cidade_id: int, g_cidade_schema: GCidadeUpdateSchema):
"""
Executa a operação de atualização.
Args:
cidade_id (int): O ID (CIDADE_ID) do registro a ser atualizado.
g_cidade_schema (GCidadeUpdateSchema): 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(cidade_id, g_cidade_schema)

View file

@ -0,0 +1,113 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.g_cidade_schema import (
GCidadeSchema,
GCidadeSaveSchema,
GCidadeUpdateSchema,
GCidadeIdSchema,
GCidadeNomeSchema
)
class GCidadeController:
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_cidade")
pass
# Lista todos os registros de cidade
def index(self):
# Importação da classe desejada
indexService = self.dynamic_import.service("g_cidade_index_service", "IndexService")
# Instância da classe service
self.indexService = indexService()
# Lista todos os registros de cidade
return {
'message': 'Registros de cidade localizados com sucesso',
'data': self.indexService.execute()
}
# Busca um registro de cidade específico pelo ID
def show(self, g_cidade_schema: GCidadeIdSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('g_cidade_show_service', 'ShowService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de cidade desejado
return {
'message': 'Registro de cidade localizado com sucesso',
'data': self.show_service.execute(g_cidade_schema)
}
# Busca um registro de cidade pelo nome (CIDADE_NOME)
def get_by_nome(self, g_cidade_schema: GCidadeNomeSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('g_cidade_get_nome_service', 'GetByNomeService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de cidade desejado
return {
'message': 'Registro de cidade localizado com sucesso',
'data': self.show_service.execute(g_cidade_schema, True)
}
# Cadastra um novo registro de cidade
def save(self, g_cidade_schema: GCidadeSaveSchema):
#Importação da classe desejada
save_service = self.dynamic_import.service('g_cidade_save_service', 'SaveService')
# Instância da classe desejada
self.save_service = save_service()
# Busca e retorna o registro de cidade desejado
return {
'message': 'Registro de cidade salvo com sucesso',
'data': self.save_service.execute(g_cidade_schema)
}
# Atualiza os dados de um registro de cidade
def update(self, cidade_id: int, g_cidade_schema: GCidadeUpdateSchema):
#Importação da classe desejada
update_service = self.dynamic_import.service('g_cidade_update_service', 'UpdateService')
# Instância da classe desejada
self.update_service = update_service()
# Busca e retorna o registro de cidade desejado
return {
'message': 'Registro de cidade atualizado com sucesso',
'data': self.update_service.execute(cidade_id, g_cidade_schema)
}
# Exclui um registro de cidade
def delete(self, g_cidade_schema: GCidadeIdSchema):
#Importação da classe desejada
delete_service = self.dynamic_import.service('g_cidade_delete_service', 'DeleteService')
# Instância da classe desejada
self.delete_service = delete_service()
# Busca e retorna o registro de cidade desejado
return {
'message': 'Registro de cidade removido com sucesso',
'data': self.delete_service.execute(g_cidade_schema)
}

View file

@ -0,0 +1,109 @@
# 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_cidade_controller import GCidadeController
from packages.v1.administrativo.schemas.g_cidade_schema import (
GCidadeSchema,
GCidadeSaveSchema,
GCidadeUpdateSchema,
GCidadeIdSchema,
GCidadeNomeSchema
)
# Inicializa o roteador para as rotas do tipo de reconhecimento
router = APIRouter()
# Instanciamento do controller desejado
g_cidade_controller = GCidadeController()
# Lista todos os registros de cidade
@router.get('/',
status_code=status.HTTP_200_OK,
summary='Lista todos os registros de cidade cadastrados',
response_description='Lista todos os registros de cidade cadastrados')
async def index(current_user: dict = Depends(get_current_user)):
# Busca todos os registros de cidade cadastrados
response = g_cidade_controller.index()
# Retorna os dados localizados
return response
# Localiza um registro de cidade pelo nome (CIDADE_NOME)
@router.get('/nome',
status_code=status.HTTP_200_OK,
summary='Busca um registro de cidade em específico pelo nome',
response_description='Busca um registro de cidade em específico')
async def get_by_nome(cidade_nome : str, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
g_cidade_schema = GCidadeNomeSchema(cidade_nome=cidade_nome)
# Busca um registro de cidade específico pelo nome
response = g_cidade_controller.get_by_nome(g_cidade_schema)
# Retorna os dados localizados
return response
# Localiza um registro de cidade pelo ID
@router.get('/{cidade_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de cidade em específico pelo ID',
response_description='Busca um registro de cidade em específico')
async def show(cidade_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
g_cidade_schema = GCidadeIdSchema(cidade_id=cidade_id)
# Busca um registro de cidade específico pelo ID
response = g_cidade_controller.show(g_cidade_schema)
# Retorna os dados localizados
return response
# Cadastro de registro de cidade
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um registro de cidade',
response_description='Cadastra um registro de cidade')
async def save(g_cidade_schema: GCidadeSaveSchema, current_user: dict = Depends(get_current_user)):
# Efetua o cadastro no banco de dados
response = g_cidade_controller.save(g_cidade_schema)
# Retorna os dados localizados
return response
# Atualiza os dados de um registro de cidade
@router.put('/{cidade_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza um registro de cidade',
response_description='Atualiza um registro de cidade')
async def update(cidade_id: int, g_cidade_schema: GCidadeUpdateSchema, current_user: dict = Depends(get_current_user)):
# Efetua a atualização dos dados
response = g_cidade_controller.update(cidade_id, g_cidade_schema)
# Retorna os dados localizados
return response
# Exclui um determinado registro de cidade
@router.delete('/{cidade_id}',
status_code=status.HTTP_200_OK,
summary='Remove um registro de cidade',
response_description='Remove um registro de cidade')
async def delete(cidade_id: int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
g_cidade_schema = GCidadeIdSchema(cidade_id=cidade_id)
# Efetua a exclusão do registro de cidade
response = g_cidade_controller.delete(g_cidade_schema)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,41 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeIdSchema
from fastapi import HTTPException, status
class DeleteRepository(BaseRepository):
"""
Repositório para a operação de exclusão de um registro na tabela
G_CIDADE.
"""
def execute(self, g_cidade_schema: GCidadeIdSchema):
"""
Executa a consulta SQL para remover um registro pelo ID (CIDADE_ID).
Args:
g_cidade_schema (GCidadeIdSchema): 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_CIDADE WHERE CIDADE_ID = :cidade_id """
# Preenchimento de parâmetros
params = {
"cidade_id": g_cidade_schema.cidade_id
}
# Execução do sql
response = self.run(sql, params)
# Retorna o resultado
return response
except Exception as e:
# Informa que houve uma falha na exclusão
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao excluir G_CIDADE: {e}"
)

View file

@ -0,0 +1,29 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeNomeSchema
class GetByNomeRepository(BaseRepository):
"""
Repositório para a operação de busca de um registro na tabela
G_CIDADE por nome (CIDADE_NOME).
"""
def execute(self, g_cidade_schema: GCidadeNomeSchema):
"""
Executa a consulta SQL para buscar um registro pelo nome da cidade.
Args:
g_cidade_schema (GCidadeNomeSchema): O esquema com o nome da cidade 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_CIDADE WHERE CIDADE_NOME = :cidade_nome """
# Preenchimento de parâmetros
params = {
'cidade_nome': g_cidade_schema.cidade_nome
}
# 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_CIDADE.
"""
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_CIDADE """
# Execução do sql
response = self.fetch_all(sql)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,59 @@
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeSaveSchema
class SaveRepository(BaseRepository):
"""
Repositório para a operação de salvamento de um novo registro na tabela G_CIDADE.
"""
def execute(self, g_cidade_schema: GCidadeSaveSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
g_cidade_schema (GCidadeSaveSchema): 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_CIDADE(
CIDADE_ID,
UF,
CIDADE_NOME,
CODIGO_IBGE,
CODIGO_GYN
) VALUES (
:cidade_id,
:uf,
:cidade_nome,
:codigo_ibge,
:codigo_gyn
) RETURNING *;"""
# Preenchimento de parâmetros
params = {
'cidade_id': g_cidade_schema.cidade_id,
'uf': g_cidade_schema.uf,
'cidade_nome': g_cidade_schema.cidade_nome,
'codigo_ibge': g_cidade_schema.codigo_ibge,
'codigo_gyn': g_cidade_schema.codigo_gyn
}
# Execução do sql
return self.run_and_return(sql, params)
except Exception as e:
# Informa que houve uma falha no salvamento do registro
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao salvar G_CIDADE: {e}"
)

View file

@ -0,0 +1,46 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeIdSchema
from fastapi import HTTPException, status
class ShowRepository(BaseRepository):
"""
Repositório para a operação de exibição de um registro na tabela G_CIDADE.
"""
def execute(self, g_cidade_schema: GCidadeIdSchema):
"""
Busca um registro específico de G_CIDADE pelo ID (CIDADE_ID).
Args:
g_cidade_schema (GCidadeIdSchema): 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_CIDADE WHERE CIDADE_ID = :cidade_id"
# Preenchimento de parâmetros
params = {
'cidade_id': g_cidade_schema.cidade_id
}
# Execução do SQL
result = self.fetch_one(sql, params)
if not result:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Registro não encontrado"
)
return result
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Erro ao buscar registro: {str(e)}"
)

View file

@ -0,0 +1,52 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeUpdateSchema
from fastapi import HTTPException, status
class UpdateRepository(BaseRepository):
"""
Repositório para a operação de atualização de um registro na tabela G_CIDADE.
"""
def execute(self, cidade_id: int, g_cidade_schema: GCidadeUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
cidade_id (int): O ID (CIDADE_ID) do registro a ser atualizado.
g_cidade_schema (GCidadeUpdateSchema): O esquema com os dados a serem atualizados.
Returns:
O registro atualizado.
Raises:
HTTPException: Caso ocorra um erro na execução da query.
"""
try:
# Montagem do SQL
sql = """ UPDATE G_CIDADE SET
UF = :uf,
CIDADE_NOME = :cidade_nome,
CODIGO_IBGE = :codigo_ibge,
CODIGO_GYN = :codigo_gyn
WHERE CIDADE_ID = :cidade_id
RETURNING *;"""
# Preenchimento de parâmetros
params = {
'cidade_id': cidade_id,
'uf': g_cidade_schema.uf,
'cidade_nome': g_cidade_schema.cidade_nome,
'codigo_ibge': g_cidade_schema.codigo_ibge,
'codigo_gyn': g_cidade_schema.codigo_gyn
}
# Execução do sql
return self.run_and_return(sql, params)
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 G_CIDADE: {e}"
)

View file

@ -0,0 +1,127 @@
from pydantic import BaseModel, field_validator, model_validator
from fastapi import HTTPException, status
from typing import Optional
# Assumindo que esta importação de validação está disponível no ambiente
from actions.validations.text import Text
# ----------------------------------------------------
# Schema base
# ----------------------------------------------------
class GCidadeSchema(BaseModel):
"""
Schema base para a tabela G_CIDADE.
Representa a estrutura completa do registro.
"""
cidade_id: int
uf: Optional[str] = None
cidade_nome: Optional[str] = None
codigo_ibge: Optional[str] = None
codigo_gyn: Optional[str] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar um G_CIDADE especifico pelo ID (GET/DELETE)
# ----------------------------------------------------
class GCidadeIdSchema(BaseModel):
"""
Schema para busca (SHOW) ou exclusão (DELETE) de registro por ID (CIDADE_ID).
"""
cidade_id: int
# ----------------------------------------------------
# Schema para localizar um G_CIDADE especifico pelo nome (GET)
# ----------------------------------------------------
class GCidadeNomeSchema(BaseModel):
"""
Schema para busca de registro por nome (CIDADE_NOME).
"""
cidade_nome: str
# ----------------------------------------------------
# Schema para criação de novo G_CIDADE (POST)
# ----------------------------------------------------
class GCidadeSaveSchema(BaseModel):
"""
Schema para salvar (POST) um novo registro de G_CIDADE.
Os campos CIDADE_ID, UF e CIDADE_NOME são considerados obrigatórios.
"""
cidade_id: Optional[int] = None
uf: str
cidade_nome: str
codigo_ibge: Optional[str] = None
codigo_gyn: Optional[str] = None
# Sanitiza os inputs enviados
@field_validator('uf', 'cidade_nome', 'codigo_ibge', 'codigo_gyn')
def sanitize_fields(cls, v):
if v:
return Text.sanitize_input(v)
return v
# Verifica se os campos obrigatórios (UF, CIDADE_NOME) foram enviados
@model_validator(mode='after')
def validate_all_fields(self):
errors = []
# O campo 'cidade_id' é checado pela tipagem do Pydantic
if not self.uf or len(self.uf.strip()) == 0:
errors.append({'input': 'uf', 'message': 'O campo UF é obrigatório.'})
if not self.cidade_nome or len(self.cidade_nome.strip()) == 0:
errors.append({'input': 'cidade_nome', 'message': 'O nome da cidade é obrigatório.'})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self
# ----------------------------------------------------
# Schema para atualizar G_CIDADE (PUT)
# ----------------------------------------------------
class GCidadeUpdateSchema(BaseModel):
"""
Schema para atualizar (PUT) um registro de G_CIDADE.
Todos os campos são opcionais, exceto o ID que virá na rota.
"""
uf: Optional[str] = None
cidade_nome: Optional[str] = None
codigo_ibge: Optional[str] = None
codigo_gyn: Optional[str] = None
# Sanitiza os inputs enviados
@field_validator('uf', 'cidade_nome', 'codigo_ibge', 'codigo_gyn')
def sanitize_fields(cls, v):
if v:
return Text.sanitize_input(v)
return v
# Valida se os campos que foram enviados não estão vazios
@model_validator(mode='after')
def validate_all_fields(self):
errors = []
# Verifica se o campo foi enviado (self.uf is not None) e se a string está vazia após strip
if self.uf is not None and len(self.uf.strip()) == 0:
errors.append({'input': 'uf', 'message': 'O campo UF não pode ser vazio.'})
if self.cidade_nome is not None and len(self.cidade_nome.strip()) == 0:
errors.append({'input': 'cidade_nome', 'message': 'O nome da cidade não pode ser vazio.'})
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_cidade_schema import GCidadeIdSchema
from packages.v1.administrativo.actions.g_cidade.g_cidade_delete_action import DeleteAction
class DeleteService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela G_CIDADE.
"""
def execute(self, g_cidade_schema: GCidadeIdSchema):
"""
Executa a operação de exclusão do registro no banco de dados.
Args:
g_cidade_schema (GCidadeIdSchema): O esquema com o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# Instanciamento da ação
delete_action = DeleteAction()
# Executa a ação em questão
data = delete_action.execute(g_cidade_schema)
# Retorno da informação
return data

View file

@ -0,0 +1,38 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeNomeSchema
from packages.v1.administrativo.actions.g_cidade.g_cidade_get_by_nome_action import GetByNomeAction
class GetByNomeService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela G_CIDADE pelo nome da cidade (CIDADE_NOME).
"""
def execute(self, g_cidade_schema: GCidadeNomeSchema, messageValidate: bool):
"""
Executa a operação de busca no banco de dados.
Args:
g_cidade_schema (GCidadeNomeSchema): O esquema com o nome da cidade 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 = GetByNomeAction()
# Executa a ação em questão
data = show_action.execute(g_cidade_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 CIDADE'
)
# Retorno da informação
return data

View file

@ -0,0 +1,32 @@
from fastapi import HTTPException, status
from packages.v1.administrativo.actions.g_cidade.g_cidade_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_CIDADE.
"""
def execute(self):
"""
Executa a operação de busca de todos os registros no banco de dados.
Returns:
A lista de registros encontrados.
"""
# Instanciamento da ação
index_action = IndexAction()
# Executa a busca de todas as ações
data = index_action.execute()
# Verifica se foram localizados registros
if not data:
# Retorna uma exceção
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail='Não foi possível localizar os registros de CIDADE'
)
# 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_cidade_schema import GCidadeSaveSchema, GCidadeNomeSchema
from packages.v1.administrativo.actions.g_cidade.g_cidade_save_action import SaveAction
from fastapi import HTTPException, status
class SaveService:
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_cidade")
pass
# Cadastra o novo G_CIDADE
def execute(self, g_cidade_schema: GCidadeSaveSchema):
# Armazena possíveis erros
errors = []
# Verifica se o nome da cidade já está sendo utilizada
# Importação de service
nome_service = self.dynamic_import.service("g_cidade_get_nome_service", "GetByNomeService")
# Instanciamento da service
self.nome_service = nome_service()
# Verifica se o nome da cidade já está sendo utilizada
self.response = self.nome_service.execute(GCidadeNomeSchema(cidade_nome=g_cidade_schema.cidade_nome), False)
# Se houver retorno significa que o nome da cidade já está sendo utilizado
if self.response:
errors.append({'input': 'cidade_nome', 'message': 'O nome da cidade informado já está sendo utilizado.'})
# 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 g_cidade_schema.cidade_id:
# Crio um objeto de sequencia
sequencia_schema = GSequenciaSchema()
# Define os dados para atualizar a sequencia
sequencia_schema.tabela = 'G_CIDADE' # Nome da tabela para a sequência
# Busco a sequência atualizada
generate = GenerateService()
# Busco a sequência atualizada
sequencia = generate.execute(sequencia_schema)
# Atualiza os dados da chave primária
g_cidade_schema.cidade_id = sequencia.sequencia
# Instanciamento de ações
save_action = SaveAction()
# Retorna o resultado da operação
return save_action.execute(g_cidade_schema)

View file

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

View file

@ -0,0 +1,24 @@
from packages.v1.administrativo.schemas.g_cidade_schema import GCidadeUpdateSchema
from packages.v1.administrativo.actions.g_cidade.g_cidade_update_action import UpdateAction
class UpdateService:
"""
Serviço para a operação de atualização de um registro na tabela
G_CIDADE.
"""
def execute(self, cidade_id : int, g_cidade_schema: GCidadeUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
cidade_id (int): O ID (CIDADE_ID) do registro a ser atualizado.
g_cidade_schema (GCidadeUpdateSchema): O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# Instanciamento de ações
update_action = UpdateAction()
# Retorna o resultado da operação
return update_action.execute(cidade_id, g_cidade_schema)

View file

@ -19,6 +19,7 @@ from packages.v1.administrativo.endpoints import g_medida_tipo_endpoint
from packages.v1.administrativo.endpoints import t_minuta_endpoint
from packages.v1.administrativo.endpoints import g_tb_bairro_endpoint
from packages.v1.administrativo.endpoints import g_tb_tipologradouro_endpoint
from packages.v1.administrativo.endpoints import g_cidade_endpoint
# Cria uma instância do APIRouter que vai agregar todas as rotas da API
api_router = APIRouter()
@ -110,4 +111,10 @@ api_router.include_router(
# Inclui as rotas de g_tb_tipologradouro
api_router.include_router(
g_tb_tipologradouro_endpoint.router, prefix="/administrativo/g_tb_tipologradouro", tags=["Tipo logradouro"]
)
# Inclui as rotas de g_cidade
api_router.include_router(
g_cidade_endpoint.router, prefix="/administrativo/g_cidade", tags=["Cidades"]
)