[MVPTN-18] feat(CRUD): Cria os recursos para manipular o CRUD de Pessoa

This commit is contained in:
Keven Willian Pereira de Souza 2025-10-07 16:17:31 -03:00
parent 588a8dc156
commit 4a48001fef
22 changed files with 1197 additions and 7 deletions

View file

@ -1,20 +1,24 @@
import importlib
from actions.config.config import Config
from typing import Optional, Any, Type
class DynamicImport:
def __init__(self):
self.config = Config.get('app.json')
self.base = 'packages.v1'
def __init__(self) -> None:
self.config: dict[str, Any] = Config.get("app.json")
self.base: str = "packages.v1"
self.package: Optional[str] = None
self.table: Optional[str] = None
def set_package(self, name):
def set_package(self, name: str) -> None:
self.package = name
def set_table(self, table):
def set_table(self, table: str):
self.table = table
def service(self, name: str, class_name : str):
def service(self, name: str, class_name : str) -> Type[Any]:
try:
# Define o nome do Módulo
module_file = f"{name}"

View file

@ -0,0 +1,26 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_pessoa.t_pessoa_delete_repository import TPessoaDeleteRepository
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaIdSchema
class TPessoaDeleteAction(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_regimebens.
"""
def execute(self, t_pessoa_id_schema: TPessoaIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
regimebens_schema (GTbRegimebensIdSchema): O esquema com o ID a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# Instanciamento do repositório
t_pessoa_delete_repository = TPessoaDeleteRepository()
# Execução do repositório
return t_pessoa_delete_repository.execute(t_pessoa_id_schema)

View file

@ -0,0 +1,27 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaNameSchema
class TPessoaGetByNameAction(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_regimebens por descrição.
"""
def execute(self, t_pessoa_name_schema: TPessoaNameSchema):
"""
Executa a operação de busca no banco de dados.
Args:
regimebens_schema (GTbRegimebensDescricaoSchema): O esquema com a descrição a ser buscada.
Returns:
O registro encontrado ou None.
"""
# Instanciamento do repositório
t_pessoa_get_by_name = TPessoaGetByNameAction()
# Execução do repositório
response = t_pessoa_get_by_name.execute(t_pessoa_name_schema)
# Retorno da informação
return response

View file

@ -0,0 +1,24 @@
from api.packages.v1.administrativo.repositories.t_pessoa.t_pessoa_index_repository import TPessoaIndexRepository
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaTipoSchema
class TPessoaIndexAction():
"""
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_regimebens.
"""
def execute(self, t_pessoa_tipo_schema: TPessoaTipoSchema):
"""
Executa a operação de listagem no banco de dados.
Returns:
A lista de todos os registros.
"""
# Instanciamento do repositório
t_pessoa_index_repository = TPessoaIndexRepository()
# Execução do repositório
response = t_pessoa_index_repository.execute(t_pessoa_tipo_schema)
# Retorno da informação
return response

View file

@ -0,0 +1,28 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_pessoa.t_pessoa_save_repository import TPessoaSaveRepository
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaSaveSchema
class TPessoaSaveAction(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_regimebens.
"""
def execute(self, t_pessoa_save_schema: TPessoaSaveSchema):
"""
Executa a operação de salvamento.
Args:
regimebens_schema (GTbRegimebensSaveSchema): O esquema com os dados a serem salvos.
Returns:
O resultado da operação de salvamento.
"""
# Instânciamento do repositório
t_pessoa_save_repository = TPessoaSaveRepository()
# Execução do repositório
response = t_pessoa_save_repository.execute(t_pessoa_save_schema)
# Retorno da informação
return response

View file

@ -0,0 +1,28 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_pessoa.t_pessoa_show_repository import TPessoaShowRepository
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaIdSchema
class TPessoaShowAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela g_tb_regimebens.
"""
def execute(self, t_pessoa_id_schema: TPessoaIdSchema):
"""
Executa a operação de exibição.
Args:
regimebens_schema (GTbRegimebensIdSchema): O esquema com o ID do registro a ser exibido.
Returns:
O resultado da operação de exibição.
"""
# Instânciamento do repositório
t_pessoa_show_repository = TPessoaShowRepository()
# Execução do repositório
response = t_pessoa_show_repository.execute(t_pessoa_id_schema)
# Retorno da informação
return response

View file

@ -0,0 +1,25 @@
from api.packages.v1.administrativo.repositories.t_pessoa.t_pessoa_update_repository import TPessoaUpdateRepository
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaUpdateSchema
class TPessoaUpdateAction:
"""
Service responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela g_tb_regimebens.
"""
def execute(self, t_pessoa_update_schema: TPessoaUpdateSchema):
"""
Executa a operação de atualização.
Args:
regimebens_schema (GTbRegimebensUpdateSchema): 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
t_pessoa_update_repository = TPessoaUpdateRepository()
# Chama o método de execução do repositório para realizar a atualização
return t_pessoa_update_repository.execute(t_pessoa_update_schema)

View file

@ -0,0 +1,107 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaIdSchema, TPessoaNameSchema, TPessoaSaveSchema, TPessoaTipoSchema, TPessoaUpdateSchema
class TPessoaController:
def __init__(self):
# Action responsável por carregar as services de acordo com o estado
self.dynamic_import = DynamicImport()
# Define o pacote que deve ser carregado
self.dynamic_import.set_package("administrativo")
# Define a tabela que o pacote pertence
self.dynamic_import.set_table("t_pessoa")
pass
# Lista todos os regimes de bens
def index(self, t_pessoa_tipo_schema : TPessoaTipoSchema):
# Importação da classe desejada
t_pessoa_index_service = self.dynamic_import.service("t_pessoa_index_service", "TPessoaIndexService")
# Instância da classe service
self.t_pessoa_index_service = t_pessoa_index_service()
# Lista todos os regimes de bens
return {
'message': 'Pessoas localizados com sucesso',
'data': self.t_pessoa_index_service.execute(t_pessoa_tipo_schema)
}
# Busca um regime de bens específico pelo ID
def show(self, t_pessoa_id_schema : TPessoaIdSchema):
#Importação da classe desejada
t_pessoa_show_service = self.dynamic_import.service('t_pessoa_show_service', 'TPessoaShowService')
# Instância da classe desejada
self.t_pessoa_show_service = t_pessoa_show_service()
# Busca e retorna o regime de bens desejado
return {
'message' : 'Pessoa localizado com sucesso',
'data': self.t_pessoa_show_service.execute(t_pessoa_id_schema)
}
# Busca um regime de bens pela descrição
def get_by_nome(self, t_pessoa_name_schema : TPessoaNameSchema):
#Importação da classe desejada
t_pessoa_show_service = self.dynamic_import.service('t_pessoa_show_service', 'TPessoaShowService')
# Instância da classe desejada
self.t_pessoa_show_service = t_pessoa_show_service()
# Busca e retorna o regime de bens desejado
return {
'message' : 'Pessoa localizado com sucesso',
'data': self.t_pessoa_show_service.execute(regimebens_schema, True)
}
# Cadastra um novo regime de bens
def save(self, t_pessoa_save_schema : TPessoaSaveSchema):
#Importação da classe desejada
t_pessoa_save_service = self.dynamic_import.service('t_pessoa_save_service', 'TPessoaSaveService')
# Instância da classe desejada
self.t_pessoa_save_service = t_pessoa_save_service()
# Busca e retorna o regime de bens desejado
return {
'message' : 'Pessoa salvo com sucesso',
'data': self.t_pessoa_save_service.execute(t_pessoa_save_schema)
}
# Atualiza os dados de um regime de bens
def update(self, t_pessoa_update_schame : TPessoaUpdateSchema):
#Importação da classe desejada
t_pessoa_update_service = self.dynamic_import.service('t_pessoa_update_service', 'TPessoaUpdateService')
# Instância da classe desejada
self.t_pessoa_update_service = t_pessoa_update_service()
# Busca e retorna o regime de bens desejado
return {
'message' : 'Pessoa atualizado com sucesso',
'data': self.t_pessoa_update_service.execute(t_pessoa_update_schame)
}
# Exclui um regime de bens
def delete(self, t_pessoa_id_schema : TPessoaIdSchema):
#Importação da classe desejada
t_pessoa_delete_service = self.dynamic_import.service('t_pessoa_delete_service', 'TPessoaDeleteService')
# Instância da classe desejada
self.t_pessoa_delete_service = t_pessoa_delete_service()
# Busca e retorna o regime de bens desejado
return {
'message' : 'Pessoa removido com sucesso',
'data': self.t_pessoa_delete_service.execute(t_pessoa_id_schema)
}

View file

@ -0,0 +1,108 @@
# Importação de bibliotecas
from fastapi import APIRouter, Depends, status
from actions.jwt.get_current_user import get_current_user
from api.packages.v1.administrativo.controllers.t_pessoa_controller import TPessoaController
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaIdSchema, TPessoaNameSchema, TPessoaSaveSchema, TPessoaUpdateSchema, TPessoaTipoSchema
# Inicializa o roteador para as rotas do regime de bens
router = APIRouter()
# Instânciamento do controller desejado
t_pessoa_controller = TPessoaController()
# Lista todos as pessoas
@router.get('/tipo/{pessoa_tipo}',
status_code=status.HTTP_200_OK,
summary='Lista todos as pessoas cadastrados',
response_description='Lista todos as pessoas cadastrados')
async def index(pessoa_tipo : str, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
t_pessoa_tipo_schema = TPessoaTipoSchema(pessoa_tipo=pessoa_tipo)
# Busca todos as pessoas cadastrados
response = t_pessoa_controller.index(t_pessoa_tipo_schema)
# Retorna os dados localizados
return response
# Localiza um regime de bens pela descrição
@router.get('/nome',
status_code=status.HTTP_200_OK,
summary='Busca um registro em específico pelo nome',
response_description='Busca um registro em específico')
async def get_by_nome(nome : str, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
t_pessoa_nome_schema = TPessoaNameSchema(nome=nome)
# Busca um regime de bens específico pela descrição
response = t_pessoa_controller.get_by_nome(t_pessoa_nome_schema)
# Retorna os dados localizados
return response
# Localiza um regime de bens pelo ID
@router.get('/{pessoa_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro em específico pelo ID da pessoa',
response_description='Busca um registro em específico')
async def show(pessoa_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
t_pessoa_id_schema = TPessoaIdSchema(pessoa_id=pessoa_id)
# Busca um regime de bens específico pelo ID
response = t_pessoa_controller.show(t_pessoa_id_schema)
# Retorna os dados localizados
return response
# Cadastro de regime de bens
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um regime de bens',
response_description='Cadastra um regime de bens')
async def save(t_pessoa_save_schema : TPessoaSaveSchema, current_user: dict = Depends(get_current_user)):
# Efetua o cadastro no banco de dados
response = t_pessoa_controller.save(t_pessoa_save_schema)
# Retorna os dados localizados
return response
# Atualiza os dados de um regime de bens
@router.put('/{pessoa_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza um regime de bens',
response_description='Atualiza um regime de bens')
async def update(pessoa_id : int, t_pessoa_update_schema : TPessoaUpdateSchema, current_user: dict = Depends(get_current_user)):
# Adiciona o ID do registro ao schema
t_pessoa_update_schema.pessoa_id = pessoa_id
# Efetua a atualização dos dados
response = t_pessoa_controller.update(t_pessoa_update_schema)
# Retorna os dados localizados
return response
# Exclui um determinado regime de bens
@router.delete('/{pessoa_id}',
status_code=status.HTTP_200_OK,
summary='Remove um regime de bens',
response_description='Remove um regime de bens')
async def delete(pessoa_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
t_pessoa_id_schema = TPessoaIdSchema(pessoa_id=pessoa_id)
# Efetua a exclusão do regime de bens
response = t_pessoa_controller.delete(t_pessoa_id_schema)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,41 @@
from abstracts.repository import BaseRepository
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaIdSchema
from fastapi import HTTPException, status
class TPessoaDeleteRepository(BaseRepository):
"""
Repositório para a operação de exclusão de um registro na tabela
g_tb_regimebens.
"""
def execute(self, t_pessoa_id_schema: TPessoaIdSchema):
"""
Executa a consulta SQL para remover um registro pelo ID.
Args:
regimebens_schema (GTbRegimebensIdSchema): O esquema com o ID a ser removido.
Returns:
O resultado da operação de exclusão.
"""
try:
# Montagem do sql
sql = """ DELETE FROM T_PESSOA WHERE PESSOA_ID = :pessoa_id """
# Preenchimento de parâmetros
params = {
"pessoa_id": t_pessoa_id_schema.pessoa_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 regime de bens: {e}"
)

View file

@ -0,0 +1,46 @@
from abstracts.repository import BaseRepository
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaTipoSchema
class TPessoaIndexRepository(BaseRepository):
"""
Repositório para a operação de listagem de todos os registros
na tabela g_tb_regimebens.
"""
def execute(self, t_pessoa_tipo_schema : TPessoaTipoSchema):
"""
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
TP.PESSOA_ID,
TP.PESSOA_TIPO,
TP.NOME,
TP.CPF_CNPJ,
TP.DATA_NASCIMENTO,
TP.SEXO,
TP.NACIONALIDADE,
TP.NATURALIDADE,
TP.EMAIL,
TP.TELEFONE,
TP.ENDERECO,
TP.NUMERO_END,
TP.BAIRRO,
TP.CIDADE,
TP.UF,
TP.CEP
FROM T_PESSOA TP
WHERE TP.PESSOA_TIPO = :pessoa_tipo """
params = {
"pessoa_tipo": t_pessoa_tipo_schema.pessoa_tipo
}
# Execução do sql
response = self.fetch_all(sql, params)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,226 @@
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaSaveSchema
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
class TPessoaSaveRepository(BaseRepository):
"""
Repositório para a operação de salvamento de um novo registro na tabela g_tb_regimebens.
"""
def execute(self, t_pessoa_save_schema: TPessoaSaveSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
regimebens_schema (GTbRegimebensSaveSchema): 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_PESSOA(
PESSOA_ID,
PESSOA_TIPO,
NOME,
NACIONALIDADE,
DOCUMENTO,
TB_DOCUMENTOTIPO_ID,
TB_PROFISSAO_ID,
TB_ESTADOCIVIL_ID,
NOME_PAI,
NOME_MAE,
DATA_CADASTRO,
NATURALIDADE,
TELEFONE,
ENDERECO,
CIDADE,
UF,
DATA_NASCIMENTO,
SEXO,
TB_REGIMECOMUNHAO_ID,
PESSOA_CONJUGE_ID,
EMAIL,
DOCUMENTO_NUMERO,
BAIRRO,
CEP,
DOCUMENTO_EXPEDICAO,
DOCUMENTO_VALIDADE,
OBSERVACAO,
CPF_CNPJ,
CPF_TERCEIRO,
NOME_FANTASIA,
TEXTO,
DDD,
CERT_CASAMENTO_NUMERO,
CERT_CASAMENTO_FOLHA,
CERT_CASAMENTO_LIVRO,
CERT_CASAMENTO_CARTORIO,
CERT_CASAMENTO_DATA,
CERT_CASAMENTO_LEI,
PESSOA_CONJUGE_NOME,
ESTRANGEIRO_NAT,
ESTRANGEIRO_NAT_TB_PAIS_ID,
ESTRANGEIRO_RES_TB_PAIS_ID,
ESTRANGEIRO_RES,
MUNICIPIO_ID,
DOCUMENTO_ORGAO,
DOCUMENTO_UF,
UF_RESIDENCIA,
INSCRICAO_MUNICIPAL,
ENVIADO_CNNCNB,
DATA_AUTERACAO,
DATA_ENVIOCCN,
CCNREGISTROS_ID,
OBSERVACAO_ENVIOCCN,
OBSERVACAO_ENVIO_CCN,
DEFICIENCIAS,
GRAU_INSTRUCAO,
CIDADE_NAT_ID,
TB_TIPOLOGRADOURO_ID,
UNIDADE,
NUMERO_END
) VALUES (
:pessoa_id,
:pessoa_tipo,
:nome,
:nacionalidade,
:documento,
:tb_documentotipo_id,
:tb_profissao_id,
:tb_estadocivil_id,
:nome_pai,
:nome_mae,
:data_cadastro,
:naturalidade,
:telefone,
:endereco,
:cidade,
:uf,
:data_nascimento,
:sexo,
:tb_regimecomunhao_id,
:pessoa_conjuge_id,
:email,
:documento_numero,
:bairro,
:cep,
:documento_expedicao,
:documento_validade,
:observacao,
:cpf_cnpj,
:cpf_terceiro,
:nome_fantasia,
:texto,
:ddd,
:cert_casamento_numero,
:cert_casamento_folha,
:cert_casamento_livro,
:cert_casamento_cartorio,
:cert_casamento_data,
:cert_casamento_lei,
:pessoa_conjuge_nome,
:estrangeiro_nat,
:estrangeiro_nat_tb_pais_id,
:estrangeiro_res_tb_pais_id,
:estrangeiro_res,
:municipio_id,
:documento_orgao,
:documento_uf,
:uf_residencia,
:inscricao_municipal,
:enviado_cncncnb,
:data_auteracao,
:data_envioccn,
:ccnregistros_id,
:observacao_envioccn,
:observacao_envio_ccn,
:deficiencias,
:grau_instrucao,
:cidade_nat_id,
:tb_tipologradouro_id,
:unidade,
:numero_end
) RETURNING *;
"""
# Preenchimento de parâmetros
params = {
'pessoa_id': t_pessoa_save_schema.pessoa_id,
'pessoa_tipo': t_pessoa_save_schema.pessoa_tipo,
'nome': t_pessoa_save_schema.nome,
'nacionalidade': t_pessoa_save_schema.nacionalidade,
'documento': t_pessoa_save_schema.documento,
'tb_documentotipo_id': t_pessoa_save_schema.tb_documentotipo_id,
'tb_profissao_id': t_pessoa_save_schema.tb_profissao_id,
'tb_estadocivil_id': t_pessoa_save_schema.tb_estadocivil_id,
'nome_pai': t_pessoa_save_schema.nome_pai,
'nome_mae': t_pessoa_save_schema.nome_mae,
'data_cadastro': t_pessoa_save_schema.data_cadastro,
'naturalidade': t_pessoa_save_schema.naturalidade,
'telefone': t_pessoa_save_schema.telefone,
'endereco': t_pessoa_save_schema.endereco,
'cidade': t_pessoa_save_schema.cidade,
'uf': t_pessoa_save_schema.uf,
'data_nascimento': t_pessoa_save_schema.data_nascimento,
'sexo': t_pessoa_save_schema.sexo,
'tb_regimecomunhao_id': t_pessoa_save_schema.tb_regimecomunhao_id,
'pessoa_conjuge_id': t_pessoa_save_schema.pessoa_conjuge_id,
'email': t_pessoa_save_schema.email,
'documento_numero': t_pessoa_save_schema.documento_numero,
'bairro': t_pessoa_save_schema.bairro,
'cep': t_pessoa_save_schema.cep,
'documento_expedicao': t_pessoa_save_schema.documento_expedicao,
'documento_validade': t_pessoa_save_schema.documento_validade,
'observacao': t_pessoa_save_schema.observacao,
'cpf_cnpj': t_pessoa_save_schema.cpf_cnpj,
'cpf_terceiro': t_pessoa_save_schema.cpf_terceiro,
'nome_fantasia': t_pessoa_save_schema.nome_fantasia,
'texto': t_pessoa_save_schema.texto,
'ddd': t_pessoa_save_schema.ddd,
'cert_casamento_numero': t_pessoa_save_schema.cert_casamento_numero,
'cert_casamento_folha': t_pessoa_save_schema.cert_casamento_folha,
'cert_casamento_livro': t_pessoa_save_schema.cert_casamento_livro,
'cert_casamento_cartorio': t_pessoa_save_schema.cert_casamento_cartorio,
'cert_casamento_data': t_pessoa_save_schema.cert_casamento_data,
'cert_casamento_lei': t_pessoa_save_schema.cert_casamento_lei,
'pessoa_conjuge_nome': t_pessoa_save_schema.pessoa_conjuge_nome,
'estrangeiro_nat': t_pessoa_save_schema.estrangeiro_nat,
'estrangeiro_nat_tb_pais_id': t_pessoa_save_schema.estrangeiro_nat_tb_pais_id,
'estrangeiro_res_tb_pais_id': t_pessoa_save_schema.estrangeiro_res_tb_pais_id,
'estrangeiro_res': t_pessoa_save_schema.estrangeiro_res,
'municipio_id': t_pessoa_save_schema.municipio_id,
'documento_orgao': t_pessoa_save_schema.documento_orgao,
'documento_uf': t_pessoa_save_schema.documento_uf,
'uf_residencia': t_pessoa_save_schema.uf_residencia,
'inscricao_municipal': t_pessoa_save_schema.inscricao_municipal,
'enviado_cncncnb': t_pessoa_save_schema.enviado_cncncnb,
'data_auteracao': t_pessoa_save_schema.data_auteracao,
'data_envioccn': t_pessoa_save_schema.data_envioccn,
'ccnregistros_id': t_pessoa_save_schema.ccnregistros_id,
'observacao_envioccn': t_pessoa_save_schema.observacao_envioccn,
'observacao_envio_ccn': t_pessoa_save_schema.observacao_envio_ccn,
'deficiencias': t_pessoa_save_schema.deficiencias,
'grau_instrucao': t_pessoa_save_schema.grau_instrucao,
'cidade_nat_id': t_pessoa_save_schema.cidade_nat_id,
'tb_tipologradouro_id': t_pessoa_save_schema.tb_tipologradouro_id,
'unidade': t_pessoa_save_schema.unidade,
'numero_end': t_pessoa_save_schema.numero_end
}
# 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 registro: {e}"
)

View file

@ -0,0 +1,46 @@
from abstracts.repository import BaseRepository
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaIdSchema
from fastapi import HTTPException, status
class TPessoaShowRepository(BaseRepository):
"""
Repositório para a operação de exibição de um registro na tabela g_tb_regimebens.
"""
def execute(self, t_pessoa_id_schema: TPessoaIdSchema):
"""
Busca um tipo de regime de bens específico pelo ID.
Args:
regimebens_schema (GTbRegimebensIdSchema): 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 T_PESSOA WHERE PESSOA_ID = :pessoa_id"
# Preenchimento de parâmetros
params = {
'pessoa_id': t_pessoa_id_schema.pessoa_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,77 @@
from abstracts.repository import BaseRepository
from fastapi import HTTPException, status
from datetime import datetime
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaUpdateSchema
class TPessoaUpdateRepository(BaseRepository):
"""
Repositório responsável por atualizar parcialmente registros na tabela T_PESSOA.
"""
def execute(self, t_pessoa_update_schema: TPessoaUpdateSchema):
"""
Executa a atualização de um registro da tabela T_PESSOA.
Args:
pessoa_id (int): ID da pessoa que será atualizada.
t_pessoa_update_schema (TPessoaUpdateSchema): Dados enviados para atualização.
Returns:
dict: Registro atualizado.
Raises:
HTTPException: Caso o registro não exista ou ocorra erro de execução.
"""
try:
# Extrai os campos do schema
data = t_pessoa_update_schema.model_dump(exclude_unset=True)
updates = []
params = {"pessoa_id": t_pessoa_update_schema.pessoa_id}
# Monta dinamicamente os campos que serão atualizados
for field, value in data.items():
if value is not None:
updates.append(f"{field.upper()} = :{field}")
params[field] = value
# Caso nenhum campo tenha sido informado
if not updates:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Nenhum campo informado para atualização."
)
# Atualiza a data de alteração automaticamente
updates.append("DATA_AUTERACAO = :data_auteracao")
params["data_auteracao"] = datetime.now()
# Montagem do SQL dinâmico
sql = f"""
UPDATE T_PESSOA
SET {', '.join(updates)}
WHERE PESSOA_ID = :pessoa_id
RETURNING *;
"""
# Executa o comando e retorna o registro atualizado
result = self.run_and_return(sql, params)
if not result:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Nenhuma pessoa encontrada com o ID {t_pessoa_update_schema.pessoa_id}."
)
return result
except HTTPException:
raise # Repassa exceções controladas
except Exception as e:
# Captura falhas inesperadas
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao atualizar a pessoa: {e}"
)

View file

@ -0,0 +1,167 @@
from pydantic import BaseModel, field_validator, model_validator
from fastapi import HTTPException, status
from typing import Optional
from datetime import datetime
# Funções para sanitização de entradas
from actions.validations.text import Text
# ----------------------------------------------------
# Schema base
# ----------------------------------------------------
class TPessoaSchema(BaseModel):
pessoa_id: Optional[float] = None
pessoa_tipo: Optional[str] = None
nome: Optional[str] = None
nacionalidade: Optional[str] = None
documento: Optional[str] = None
tb_documentotipo_id: Optional[float] = None
tb_profissao_id: Optional[float] = None
tb_estadocivil_id: Optional[float] = None
nome_pai: Optional[str] = None
nome_mae: Optional[str] = None
data_cadastro: Optional[datetime] = None
naturalidade: Optional[str] = None
telefone: Optional[str] = None
endereco: Optional[str] = None
cidade: Optional[str] = None
uf: Optional[str] = None
data_nascimento: Optional[datetime] = None
sexo: Optional[str] = None
tb_regimecomunhao_id: Optional[float] = None
pessoa_conjuge_id: Optional[float] = None
email: Optional[str] = None
documento_numero: Optional[str] = None
bairro: Optional[str] = None
cep: Optional[str] = None
documento_expedicao: Optional[datetime] = None
documento_validade: Optional[datetime] = None
observacao: Optional[str] = None
cpf_cnpj: Optional[str] = None
cpf_terceiro: Optional[str] = None
nome_fantasia: Optional[str] = None
texto: Optional[bytes] = None
ddd: Optional[str] = None
cert_casamento_numero: Optional[float] = None
cert_casamento_folha: Optional[str] = None
cert_casamento_livro: Optional[str] = None
cert_casamento_cartorio: Optional[str] = None
cert_casamento_data: Optional[datetime] = None
cert_casamento_lei: Optional[str] = None
pessoa_conjuge_nome: Optional[str] = None
estrangeiro_nat: Optional[str] = None
estrangeiro_nat_tb_pais_id: Optional[float] = None
estrangeiro_res_tb_pais_id: Optional[float] = None
estrangeiro_res: Optional[str] = None
municipio_id: Optional[float] = None
documento_orgao: Optional[str] = None
documento_uf: Optional[str] = None
uf_residencia: Optional[str] = None
inscricao_municipal: Optional[str] = None
enviado_cncncnb: Optional[str] = None
data_auteracao: Optional[datetime] = None
data_envio_ccn: Optional[datetime] = None
ccnregistros_id: Optional[float] = None
observacao_envioccn: Optional[str] = None
observacao_envio_ccn: Optional[str] = None
deficiencias: Optional[str] = None
grau_instrucao: Optional[float] = None
cidade_nat_id: Optional[float] = None
tb_tipologradouro_id: Optional[float] = None
unidade: Optional[str] = None
numero_end: Optional[float] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar um tipo especifico pelo ID (GET)
# ----------------------------------------------------
class TPessoaIdSchema(BaseModel):
pessoa_id: int
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar um tipo especifico pelo ID (GET)
# ----------------------------------------------------
class TPessoaNameSchema(BaseModel):
nome: str
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar um tipo especifico pelo ID (GET)
# ----------------------------------------------------
class TPessoaTipoSchema(BaseModel):
pessoa_tipo: str
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para criação de pessoa (POST)
# ----------------------------------------------------
class TPessoaSaveSchema(TPessoaSchema):
# Campos obrigatórios podem ser definidos aqui
nome: str
pessoa_tipo: str
@field_validator(
'nome', 'pessoa_tipo', 'nacionalidade', 'documento', 'nome_pai', 'nome_mae',
'endereco', 'cidade', 'uf', 'sexo', 'email', 'bairro', 'cep'
)
def sanitize_fields(cls, v):
if v:
return Text.sanitize_input(v)
return v
@model_validator(mode='after')
def validate_required_fields(self):
errors = []
if not self.nome or len(self.nome.strip()) == 0:
errors.append({'input': 'nome', 'message': 'O nome é obrigatório.'})
if not self.pessoa_tipo or len(self.pessoa_tipo.strip()) == 0:
errors.append({'input': 'pessoa_tipo', 'message': 'O tipo de pessoa é obrigatório.'})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self
# ----------------------------------------------------
# Schema para atualização de pessoa (PUT)
# ----------------------------------------------------
class TPessoaUpdateSchema(TPessoaSchema):
@field_validator(
'nome', 'pessoa_tipo', 'nacionalidade', 'documento', 'nome_pai', 'nome_mae',
'endereco', 'cidade', 'uf', 'sexo', 'email', 'bairro', 'cep'
)
def sanitize_fields(cls, v):
if v:
return Text.sanitize_input(v)
return v
@model_validator(mode='after')
def validate_required_fields(self):
# Em update, podemos permitir que campos sejam opcionais, mas se enviados, não podem ser vazios
errors = []
for field in ['nome', 'pessoa_tipo']:
value = getattr(self, field)
if value is not None and len(str(value).strip()) == 0:
errors.append({'input': field, 'message': f'O campo {field} não pode estar vazio.'})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self

View file

@ -0,0 +1,27 @@
from api.packages.v1.administrativo.actions.t_pessoa.t_pessoa_delete_action import TPessoaDeleteAction
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaIdSchema
class TPessoaDeleteService:
"""
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_regimebens.
"""
def execute(self, t_pessoa_id_schema: TPessoaIdSchema):
"""
Executa a operação de exclusão do registro no banco de dados.
Args:
regimebens_schema (GTbRegimebensIdSchema): O esquema com o ID do registro a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# Instanciamento da ação
t_pessoa_delete_action = TPessoaDeleteAction()
# Executa a ação em questão
data = t_pessoa_delete_action.execute(t_pessoa_id_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_regimebens_schema import GTbRegimebensDescricaoSchema
from packages.v1.administrativo.actions.g_tb_regimebens.g_tb_regimebens_get_by_descricao_action import GetByDescricaoAction
class GetByDescricaoService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela g_tb_regimebens pela sua descrição.
"""
def execute(self, regimebens_schema: GTbRegimebensDescricaoSchema, messageValidate: bool):
"""
Executa a operação de busca no banco de dados.
Args:
regimebens_schema (GTbRegimebensDescricaoSchema): O esquema com a descrição a ser buscada.
messageValidate (bool): Se True, lança uma exceção HTTP caso o registro não seja encontrado.
Returns:
O registro encontrado ou None.
"""
# Instanciamento da ação
show_action = GetByDescricaoAction()
# Executa a ação em questão
data = show_action.execute(regimebens_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 regime de bens'
)
# Retorno da informação
return data

View file

@ -0,0 +1,34 @@
from api.packages.v1.administrativo.actions.t_pessoa.t_pessoa_index_action import TPessoaIndexAction
from fastapi import HTTPException, status
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaTipoSchema
class TPessoaIndexService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de registros na tabela g_tb_regimebens.
"""
def execute(self, t_pessoa_tipo_schema: TPessoaTipoSchema):
"""
Executa a operação de busca de todos os registros no banco de dados.
Returns:
A lista de registros encontrados.
"""
# Instanciamento da ação
t_pessoa_index_action = TPessoaIndexAction()
# Executa a busca de todas as ações
data = t_pessoa_index_action.execute(t_pessoa_tipo_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 as pessoas'
)
# Retorna as informações localizadas
return data

View file

@ -0,0 +1,45 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from api.packages.v1.administrativo.actions.t_pessoa.t_pessoa_save_action import TPessoaSaveAction
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaSaveSchema
from api.packages.v1.sequencia.schemas.g_sequencia import GSequenciaSchema
from api.packages.v1.sequencia.services.g_sequencia.checkout_service import GenerateService
class TPessoaSaveService:
def __init__(self):
# Ação responsável por carregar as services de acordo com o estado
self.dynamic_import = DynamicImport()
# Define o pacote que deve ser carregado
self.dynamic_import.set_package("administrativo")
# Define a tabela que o pacote pertence
self.dynamic_import.set_table("t_pessoa")
pass
# Cadastra o novo regime de bens
def execute(self, t_pessoa_save_schema: TPessoaSaveSchema):
# Verifica se precisa gerar o ID de sequência
if not t_pessoa_save_schema.pessoa_id:
# Crio um objeto de sequencia
sequencia_schema = GSequenciaSchema()
# Define os dados para atualizar a sequencia
sequencia_schema.tabela = 'T_PESSOA'
# Busco a sequência atualizada
generate = GenerateService()
# Busco a sequência atualizada
sequencia = generate.execute(sequencia_schema)
# Atualiza os dados da chave primária
regimebens_schema.tb_regimebens_id = sequencia.sequencia
# Instanciamento de ações
t_pessoa_save_action = TPessoaSaveAction()
# Retorna o resultado da operação
return t_pessoa_save_action.execute(t_pessoa_save_schema)

View file

@ -0,0 +1,36 @@
from fastapi import HTTPException, status
from api.packages.v1.administrativo.actions.t_pessoa.t_pessoa_show_action import TPessoaShowAction
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaIdSchema
class TPessoaShowService:
"""
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_regimebens.
"""
def execute(self, t_pessoa_id_schema: TPessoaIdSchema):
"""
Executa a operação de busca no banco de dados.
Args:
regimebens_schema (GTbRegimebensIdSchema): O esquema com o ID a ser buscado.
Returns:
O resultado da busca.
"""
# Instanciamento da ação
t_pessoa_show_action = TPessoaShowAction()
# Executa a ação em questão
data = t_pessoa_show_action.execute(t_pessoa_id_schema)
if not data:
# Retorna uma exceção
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail='Não foi possivel localizar a pessoa desejada'
)
# Retorno da informação
return data

View file

@ -0,0 +1,24 @@
from api.packages.v1.administrativo.actions.t_pessoa.t_pessoa_update_action import TPessoaUpdateAction
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaUpdateSchema
class TPessoaUpdateService:
"""
Serviço para a operação de atualização de um registro na tabela
g_tb_regimebens.
"""
def execute(self, t_pessoa_update_schema: TPessoaUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
regimebens_schema (GTbRegimebensUpdateSchema): O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# Instanciamento de ações
t_pessoa_update_action = TPessoaUpdateAction()
# Retorna o resultado da operação
return t_pessoa_update_action.execute(t_pessoa_update_schema)

View file

@ -1,5 +1,4 @@
# Importa o gerenciador de rotas do FastAPI
from api.packages.v1.administrativo.endpoints import t_imovel_unidade_endpoint
from fastapi import APIRouter
# Importa os módulos de rotas específicos
@ -28,6 +27,8 @@ from packages.v1.administrativo.endpoints import g_marcacao_tipo_endpoint
from packages.v1.administrativo.endpoints import t_servico_etiqueta_endpoint
from packages.v1.administrativo.endpoints import g_uf_endpoint
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
# Cria uma instância do APIRouter que vai agregar todas as rotas da API
api_router = APIRouter()
@ -164,4 +165,9 @@ api_router.include_router(
# Inclui as rotas de t_imovel_unidade
api_router.include_router(
t_imovel_unidade_endpoint.router, prefix="/administrativo/t_imovel_unidade", tags=["Imóveis Unidades"]
)
# Inclui as rotas de t_imovel_unidade
api_router.include_router(
t_pessoa_endpoint.router, prefix="/administrativo/t_pessoa", tags=["Pessoas Fisicas e Jurídicas"]
)