[MVPTN-18] feat(CRUD): Cria o CRUD do T_IMOVEL e T_IMOVEL_UNIDADE

This commit is contained in:
Keven Willian Pereira de Souza 2025-10-06 18:25:47 -03:00
parent aa9d1f57f3
commit a7978d385e
37 changed files with 2065 additions and 0 deletions

View file

@ -0,0 +1,24 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_imovel.t_imovel_delete_repository import TImovelDeleteRepository
from api.packages.v1.administrativo.schemas.t_imovel_schema import TImovelIdSchema
class TImovelDeleteAction(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 t_t_imovel_id.
"""
def execute(self, t_imovel_id_schema: TImovelIdSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
t_imovel_id_schema (TImovelSchema): O esquema com o ID a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# Instanciamento do repositório
t_imovel_delete_repository = TImovelDeleteRepository()
# Execução do repositório
return t_imovel_delete_repository.execute(t_imovel_id_schema)

View file

@ -0,0 +1,25 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_imovel.t_imovel_index_repository import TImovelIndexRepository
from api.packages.v1.administrativo.schemas.t_imovel_schema import TImovelIndexSchema
class TImovelIndexAction(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 t_censec_qualidade.
"""
def execute(self, t_imovel_index_schema: TImovelIndexSchema):
"""
Executa a operação de listagem no banco de dados.
Returns:
A lista de todos os registros.
"""
# Instanciamento do repositório
t_imovel_index_repositoy = TImovelIndexRepository()
# Execução do repositório
response = t_imovel_index_repositoy.execute(t_imovel_index_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_imovel.t_imovel_save_repository import TImovelSaveRepository
from api.packages.v1.administrativo.schemas.t_imovel_schema import TImovelSchema
class TImovelSaveAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvar um novo registro na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_schema: TImovelSchema):
"""
Executa a operação de salvamento.
Args:
t_imovel_schema (TCensecQualidadeSaveSchema): O esquema com os dados a serem salvos.
Returns:
O resultado da operação de salvamento.
"""
# Instanciamento do repositório
t_imovel_save_repository = TImovelSaveRepository()
# Execução do repositório
response = t_imovel_save_repository.execute(t_imovel_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_imovel.t_imovel_show_repository import TImovelShowRepository
from api.packages.v1.administrativo.schemas.t_imovel_schema import TImovelIdSchema
class TImovelShowAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_id_schema: TImovelIdSchema):
"""
Executa a operação de exibição.
Args:
t_imovel_schema (TCensecQualidadeIdSchema): 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_imovel_show_repository = TImovelShowRepository()
# Execução do repositório
response = t_imovel_show_repository.execute(t_imovel_id_schema)
# Retorno da informação
return response

View file

@ -0,0 +1,27 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_imovel.t_imovel_update_repository import TImovelUpdateRepository
from api.packages.v1.administrativo.schemas.t_imovel_schema import TImovelUpdateSchema
class TImovelUpdateAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_update_schema: TImovelUpdateSchema):
"""
Executa a operação de atualização.
Args:
t_imovel_unidade_id (int): O ID do registro a ser atualizado.
t_imovel_update_schema (TCensecQualidadeUpdateSchema): 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_imovel_update_repository = TImovelUpdateRepository()
# Chama o método de execução do repositório para realizar a atualização
return t_imovel_update_repository.execute(t_imovel_update_schema)

View file

@ -0,0 +1,26 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_imovel_unidade.t_imovel_unidade_delete_repository import TImovelUnidadeDeleteRepository
from packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeSchema
class TImovelUnidadeDeleteAction(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 t_t_imovel_unidade.
"""
def execute(self, t_imovel_unidade_schema: TImovelUnidadeSchema):
"""
Executa a operação de exclusão no banco de dados.
Args:
t_imovel_unidade_schema (TImovelUnidadeSchema): O esquema com o ID a ser excluído.
Returns:
O resultado da operação de exclusão.
"""
# Instanciamento do repositório
delete_repository = TImovelUnidadeDeleteRepository()
# Execução do repositório
return delete_repository.execute(t_imovel_unidade_schema)

View file

@ -0,0 +1,25 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_imovel_unidade.t_imovel_unidade_index_repository import TImovelUnidadeIndexRepository
from api.packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeIndexSchema
class TImovelUnidadeIndexAction(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 t_censec_qualidade.
"""
def execute(self, t_imovel_unidade_index_schema: TImovelUnidadeIndexSchema):
"""
Executa a operação de listagem no banco de dados.
Returns:
A lista de todos os registros.
"""
# Instanciamento do repositório
index_repository = TImovelUnidadeIndexRepository()
# Execução do repositório
response = index_repository.execute(t_imovel_unidade_index_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_imovel_unidade.t_imovel_unidade_save_repository import TImovelUnidadeSaveRepository
from api.packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeSchema
class TImovelUnidadeSaveAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de salvar um novo registro na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_unidade_schema: TImovelUnidadeSchema):
"""
Executa a operação de salvamento.
Args:
t_imovel_unidade_schema (TCensecQualidadeSaveSchema): O esquema com os dados a serem salvos.
Returns:
O resultado da operação de salvamento.
"""
# Instanciamento do repositório
t_imovel_unidade_save_repository = TImovelUnidadeSaveRepository()
# Execução do repositório
response = t_imovel_unidade_save_repository.execute(t_imovel_unidade_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_imovel_unidade.t_imovel_unidade_show_repository import TImovelUnidadeShowRepository
from api.packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeSchema
class TImovelUnidadeShowAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a exibição
de um registro na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_unidade_schema: TImovelUnidadeSchema):
"""
Executa a operação de exibição.
Args:
t_imovel_unidade_schema (TCensecQualidadeIdSchema): 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_imovel_unidade_show_repository = TImovelUnidadeShowRepository()
# Execução do repositório
response = t_imovel_unidade_show_repository.execute(t_imovel_unidade_schema)
# Retorno da informação
return response

View file

@ -0,0 +1,27 @@
from abstracts.action import BaseAction
from api.packages.v1.administrativo.repositories.t_imovel_unidade.t_imovel_unidade_update_repository import TImovelUnidadeUpdateRepository
from api.packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeUpdateSchema
class TImovelUnidadeUpdateAction(BaseAction):
"""
Serviço responsável por encapsular a lógica de negócio para a atualização
de um registro na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_unidade_schema: TImovelUnidadeUpdateSchema):
"""
Executa a operação de atualização.
Args:
t_imovel_unidade_id (int): O ID do registro a ser atualizado.
t_imovel_unidade_schema (TCensecQualidadeUpdateSchema): 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_imovelUnidade_update_repository = TImovelUnidadeUpdateRepository()
# Chama o método de execução do repositório para realizar a atualização
return t_imovelUnidade_update_repository.execute(t_imovel_unidade_schema)

View file

@ -0,0 +1,95 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.t_imovel_schema import (
TImovelIndexSchema,
TImovelSaveSchema,
TImovelUpdateSchema,
TImovelIdSchema
)
class TImovelController:
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_imovel")
pass
# Lista todos os registros de t_imovel
def index(self, t_imovel_index_schema: TImovelIndexSchema):
# Importação da classe desejada
t_imovel_index_service = self.dynamic_import.service("t_imovel_index_service", "TImovelIndexService")
# Instância da classe service
self.t_imovel_index_service = t_imovel_index_service()
# Lista todos os registros de t_imovel
return {
'message': 'Registros de t_imovel localizados com sucesso',
'data': self.t_imovel_index_service.execute(t_imovel_index_schema)
}
# Busca um registro de t_imovel específico pelo ID
def show(self, t_imovel_schema: TImovelIdSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('t_imovel_show_service', 'TImovelShowService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de t_imovel desejado
return {
'message': 'Registro de t_imovel localizado com sucesso',
'data': self.show_service.execute(t_imovel_schema)
}
# Cadastra um novo registro de t_imovel
def save(self, t_imovel_save_schema: TImovelSaveSchema):
#Importação da classe desejada
save_service = self.dynamic_import.service('t_imovel_save_service', 'TImovelSaveService')
# Instância da classe desejada
self.save_service = save_service()
# Busca e retorna o registro de t_imovel desejado
return {
'message': 'Registro de t_imovel salvo com sucesso',
'data': self.save_service.execute(t_imovel_save_schema)
}
# Atualiza os dados de um registro de t_imovel
def update(self, t_imovel_id: int, t_imovel_schema: TImovelUpdateSchema):
#Importação da classe desejada
update_service = self.dynamic_import.service('t_imovel_update_service', 'TImovelUpdateService')
# Instância da classe desejada
self.update_service = update_service()
# Busca e retorna o registro de t_imovel desejado
return {
'message': 'Registro de t_imovel atualizado com sucesso',
'data': self.update_service.execute(t_imovel_id, t_imovel_schema)
}
# Exclui um registro de t_imovel
def delete(self, t_imovel_schema: TImovelIdSchema):
#Importação da classe desejada
delete_service = self.dynamic_import.service('t_imovel_delete_service', 'TImovelDeleteService')
# Instância da classe desejada
self.delete_service = delete_service()
# Busca e retorna o registro de t_imovel desejado
return {
'message': 'Registro de t_imovel removido com sucesso',
'data': self.delete_service.execute(t_imovel_schema)
}

View file

@ -0,0 +1,95 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.t_imovel_unidade_schema import (
TImovelUnidadeIndexSchema,
TImovelUnidadeSaveSchema,
TImovelUnidadeUpdateSchema,
TImovelUnidadeIdSchema
)
class TImovelUnidadeController:
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_imovel_unidade")
pass
# Lista todos os registros de censec_qualidade
def index(self, t_imovel_unidade_index_schema: TImovelUnidadeIndexSchema):
# Importação da classe desejada
t_imovel_index_service = self.dynamic_import.service("t_imovel_unidade_index_service", "TImovelUnidadeIndexService")
# Instância da classe service
self.t_imovel_index_service = t_imovel_index_service()
# Lista todos os registros de censec_qualidade
return {
'message': 'Registros de t_imovel_unidade localizados com sucesso',
'data': self.t_imovel_index_service.execute(t_imovel_unidade_index_schema)
}
# Busca um registro de censec_qualidade específico pelo ID
def show(self, t_imovel_unidade_schema: TImovelUnidadeIdSchema):
#Importação da classe desejada
show_service = self.dynamic_import.service('t_imovel_unidade_show_service', 'ShowService')
# Instância da classe desejada
self.show_service = show_service()
# Busca e retorna o registro de censec_qualidade desejado
return {
'message': 'Registro de censec_qualidade localizado com sucesso',
'data': self.show_service.execute(t_imovel_unidade_schema)
}
# Cadastra um novo registro de censec_qualidade
def save(self, t_imovel_unidade_schema: TImovelUnidadeSaveSchema):
#Importação da classe desejada
save_service = self.dynamic_import.service('t_imovel_unidade_save_service', 'SaveService')
# Instância da classe desejada
self.save_service = save_service()
# Busca e retorna o registro de censec_qualidade desejado
return {
'message': 'Registro de censec_qualidade salvo com sucesso',
'data': self.save_service.execute(t_imovel_unidade_schema)
}
# Atualiza os dados de um registro de censec_qualidade
def update(self, t_imovel_unidade_id: int, t_imovel_unidade_schema: TImovelUnidadeUpdateSchema):
#Importação da classe desejada
update_service = self.dynamic_import.service('t_imovel_unidade_update_service', 'UpdateService')
# Instância da classe desejada
self.update_service = update_service()
# Busca e retorna o registro de censec_qualidade desejado
return {
'message': 'Registro de censec_qualidade atualizado com sucesso',
'data': self.update_service.execute(t_imovel_unidade_id, t_imovel_unidade_schema)
}
# Exclui um registro de censec_qualidade
def delete(self, t_imovel_unidade_schema: TImovelUnidadeIdSchema):
#Importação da classe desejada
delete_service = self.dynamic_import.service('t_imovel_unidade_delete_service', 'DeleteService')
# Instância da classe desejada
self.delete_service = delete_service()
# Busca e retorna o registro de censec_qualidade desejado
return {
'message': 'Registro de censec_qualidade removido com sucesso',
'data': self.delete_service.execute(t_imovel_unidade_schema)
}

View file

@ -0,0 +1,93 @@
# Importação de bibliotecas
from typing import Optional
from fastapi import APIRouter, Depends, status
from actions.jwt.get_current_user import get_current_user
from packages.v1.administrativo.controllers.t_imovel_controller import TImovelController
from packages.v1.administrativo.schemas.t_imovel_schema import (
TImovelIndexSchema,
TImovelSaveSchema,
TImovelUpdateSchema,
TImovelIdSchema
)
# Inicializa o roteador para as rotas do tipo de reconhecimento
router = APIRouter()
# Instanciamento do controller desejado
t_imovel_controller = TImovelController()
# Lista todos os registros de t_imovel
@router.get('/classe/{imovel_id}',
status_code=status.HTTP_200_OK,
summary='Lista todos os registros de t_imovel cadastrados',
response_description='Lista todos os registros de t_imovel cadastrados')
async def index(imovel_id: int, current_user: dict = Depends(get_current_user)):
t_imovel_index_schema = TImovelIndexSchema(imovel_id=imovel_id)
# Busca todos os registros de t_imovel cadastrados
response = t_imovel_controller.index(t_imovel_index_schema)
# Retorna os dados localizados
return response
# Localiza um registro de t_imovel pelo ID
@router.get('/{imovel_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de t_imovel em específico pelo ID',
response_description='Busca um registro de t_imovel em específico')
async def show(imovel_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
t_imovel_schema = TImovelIdSchema(imovel_id=imovel_id)
# Busca um registro de t_imovel específico pelo ID
response = t_imovel_controller.show(t_imovel_schema)
# Retorna os dados localizados
return response
# Cadastro de registro de t_imovel
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um registro de t_imovel',
response_description='Cadastra um registro de t_imovel')
async def save(t_imovel_schema: TImovelSaveSchema, current_user: dict = Depends(get_current_user)):
# Efetua o cadastro no banco de dados
response = t_imovel_controller.save(t_imovel_schema)
# Retorna os dados localizados
return response
# Atualiza os dados de um registro de t_imovel
@router.put('/{imovel_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza um registro de t_imovel',
response_description='Atualiza um registro de t_imovel')
async def update(imovel_id: int, t_imovel_schema: TImovelUpdateSchema, current_user: dict = Depends(get_current_user)):
# Efetua a atualização dos dados
response = t_imovel_controller.update(imovel_id, t_imovel_schema)
# Retorna os dados localizados
return response
# Exclui um determinado registro de t_imovel
@router.delete('/{imovel_id}',
status_code=status.HTTP_200_OK,
summary='Remove um registro de t_imovel',
response_description='Remove um registro de t_imovel')
async def delete(imovel_id: int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
t_imovel_schema = TImovelIdSchema(imovel_id=imovel_id)
# Efetua a exclusão do registro de t_imovel
response = t_imovel_controller.delete(t_imovel_schema)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,93 @@
# Importação de bibliotecas
from typing import Optional
from fastapi import APIRouter, Depends, status
from actions.jwt.get_current_user import get_current_user
from packages.v1.administrativo.controllers.t_imovel_unidade_controller import TImovelUnidadeController
from packages.v1.administrativo.schemas.t_imovel_unidade_schema import (
TImovelUnidadeIndexSchema,
TImovelUnidadeSaveSchema,
TImovelUnidadeUpdateSchema,
TImovelUnidadeIdSchema
)
# Inicializa o roteador para as rotas do tipo de reconhecimento
router = APIRouter()
# Instanciamento do controller desejado
t_imovel_unidade_controller = TImovelUnidadeController()
# Lista todos os registros de t_imovel_unidade
@router.get('/imovel/{imovel_id}',
status_code=status.HTTP_200_OK,
summary='Lista todos os registros de t_imovel_unidade cadastrados',
response_description='Lista todos os registros de t_imovel_unidade cadastrados')
async def index(imovel_id: int, current_user: dict = Depends(get_current_user)):
t_imovel_unidade_index_schema = TImovelUnidadeIndexSchema(imovel_id=imovel_id)
# Busca todos os registros de t_imovel_unidade cadastrados
response = t_imovel_unidade_controller.index(t_imovel_unidade_index_schema)
# Retorna os dados localizados
return response
# Localiza um registro de t_imovel_unidade pelo ID
@router.get('/{imovel_unidade_id}',
status_code=status.HTTP_200_OK,
summary='Busca um registro de t_imovel_unidade em específico pelo ID',
response_description='Busca um registro de t_imovel_unidade em específico')
async def show(imovel_unidade_id : int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
t_imovel_unidade_schema = TImovelUnidadeIdSchema(imovel_unidade_id=imovel_unidade_id)
# Busca um registro de t_imovel_unidade específico pelo ID
response = t_imovel_unidade_controller.show(t_imovel_unidade_schema)
# Retorna os dados localizados
return response
# Cadastro de registro de t_imovel_unidade
@router.post('/',
status_code=status.HTTP_201_CREATED,
summary='Cadastra um registro de t_imovel_unidade',
response_description='Cadastra um registro de t_imovel_unidade')
async def save(t_imovel_unidade_schema: TImovelUnidadeSaveSchema, current_user: dict = Depends(get_current_user)):
# Efetua o cadastro no banco de dados
response = t_imovel_unidade_controller.save(t_imovel_unidade_schema)
# Retorna os dados localizados
return response
# Atualiza os dados de um registro de t_imovel_unidade
@router.put('/{imovel_unidade_id}',
status_code=status.HTTP_200_OK,
summary='Atualiza um registro de t_imovel_unidade',
response_description='Atualiza um registro de t_imovel_unidade')
async def update(imovel_unidade_id: int, t_imovel_unidade_schema: TImovelUnidadeUpdateSchema, current_user: dict = Depends(get_current_user)):
# Efetua a atualização dos dados
response = t_imovel_unidade_controller.update(imovel_unidade_id, t_imovel_unidade_schema)
# Retorna os dados localizados
return response
# Exclui um determinado registro de t_imovel_unidade
@router.delete('/{imovel_unidade_id}',
status_code=status.HTTP_200_OK,
summary='Remove um registro de t_imovel_unidade',
response_description='Remove um registro de t_imovel_unidade')
async def delete(imovel_unidade_id: int, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
t_imovel_unidade_schema = TImovelUnidadeIdSchema(imovel_unidade_id=imovel_unidade_id)
# Efetua a exclusão do registro de t_imovel_unidade
response = t_imovel_unidade_controller.delete(t_imovel_unidade_schema)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,41 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_imovel_schema import TImovelIdSchema
from fastapi import HTTPException, status
class TImovelDeleteRepository(BaseRepository):
"""
Repositório para a operação de exclusão de um registro na tabela
t_censec_qualidade.
"""
def execute(self, t_imovel_id_schema: TImovelIdSchema):
"""
Executa a consulta SQL para remover um registro pelo ID.
Args:
t_imovel_schema (TImovelUnidadeIdSchema): 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_IMOVEL TI WHERE TI.IMOVEL_ID = :imovelId """
# Preenchimento de parâmetros
params = {
"imovelId": t_imovel_id_schema.imovel_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 IMÓVEL: {e}"
)

View file

@ -0,0 +1,30 @@
from abstracts.repository import BaseRepository
from api.packages.v1.administrativo.schemas.t_imovel_schema import TImovelIndexSchema
class TImovelIndexRepository(BaseRepository):
"""
Repositório para a operação de listagem de todos os registros
na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_index_schema: TImovelIndexSchema):
"""
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 TI.*, GTBB.DESCRICAO AS GTBB_DESCRICAO FROM T_IMOVEL TI
JOIN G_TB_BAIRRO GTBB ON TI.TB_BAIRRO_ID = GTBB.TB_BAIRRO_ID
WHERE TI.TIPO_CLASSE = :tipoClasse """
params = {
"tipoClasse": t_imovel_index_schema.tipo_classe
}
# Execução do sql
response = self.fetch_all(sql, params)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,90 @@
from api.packages.v1.administrativo.schemas.t_imovel_schema import TImovelSchema
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
class TImovelSaveRepository(BaseRepository):
"""
Repositório para a operação de salvamento de um novo registro na tabela t_t_imovel_unidade.
"""
def execute(self, t_imovel_schema: TImovelSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
t_imovel_schema (TImovelSchema): 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 T_IMOVEL (
IMOVEL_ID,
TIPO_CLASSE,
TIPO_REGISTRO,
DATA_REGISTRO,
NUMERO,
NUMERO_LETRA,
CIDADE,
CEP,
UF,
TB_BAIRRO_ID,
CARTORIO,
LIVRO,
CNS
) VALUES (
:imovel_id,
:tipo_classe,
:tipo_registro,
:data_registro,
:numero,
:numero_letra,
:cidade,
:cep,
:uf,
:tb_bairro_id,
:cartorio,
:livro,
:cns
)
RETURNING *;
"""
# ----------------------------------------------------
# Preenchimento dos parâmetros
# ----------------------------------------------------
params = {
"imovel_id": t_imovel_schema.imovel_id,
"tipo_classe": t_imovel_schema.tipo_classe,
"tipo_registro": t_imovel_schema.tipo_registro,
"data_registro": t_imovel_schema.data_registro,
"numero": t_imovel_schema.numero,
"numero_letra": t_imovel_schema.numero_letra,
"cidade": t_imovel_schema.cidade,
"cep": t_imovel_schema.cep,
"uf": t_imovel_schema.uf,
"tb_bairro_id": t_imovel_schema.tb_bairro_id,
"cartorio": t_imovel_schema.cartorio,
"livro": t_imovel_schema.livro,
"cns": t_imovel_schema.cns,
}
# 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 T_IMOVEL: {e}"
)

View file

@ -0,0 +1,46 @@
from abstracts.repository import BaseRepository
from api.packages.v1.administrativo.schemas.t_imovel_schema import TImovelIdSchema
from fastapi import HTTPException, status
class TImovelShowRepository(BaseRepository):
"""
Repositório para a operação de exibição de um registro na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_id_schema: TImovelIdSchema):
"""
Busca um registro específico de CENSEC_QUALIDADE pelo ID.
Args:
t_imovel_schema (TImovelIdSchema): 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_IMOVEL WHERE IMOVEL_ID = :imovel_id"
# Preenchimento de parâmetros
params = {
'imovel_id': t_imovel_id_schema.imovel_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_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao buscar registro: {str(e)}"
)

View file

@ -0,0 +1,84 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_imovel_schema import TImovelUpdateSchema
from fastapi import HTTPException, status
class TImovelUpdateRepository(BaseRepository):
"""
Repositório para atualização de registros na tabela T_IMOVEL.
"""
def execute(self, t_imovel_schema: TImovelUpdateSchema):
"""
Executa a atualização de um imóvel pelo ID.
"""
try:
# ----------------------------------------------------
# Mapeamento de atributos → colunas SQL
# ----------------------------------------------------
field_map = {
"tipo_classe": "TIPO_CLASSE",
"tipo_registro": "TIPO_REGISTRO",
"data_registro": "DATA_REGISTRO",
"numero": "NUMERO",
"numero_letra": "NUMERO_LETRA",
"cidade": "CIDADE",
"cep": "CEP",
"uf": "UF",
"tb_bairro_id": "TB_BAIRRO_ID",
"cartorio": "CARTORIO",
"livro": "LIVRO",
"cns": "CNS",
}
# ----------------------------------------------------
# Montagem dinâmica dos campos
# ----------------------------------------------------
updates = []
params = {}
for attr, column in field_map.items():
value = getattr(t_imovel_schema, attr, None)
if value is not None:
updates.append(f"{column} = :{attr}")
params[attr] = value
if not updates:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Nenhum campo válido foi enviado para atualização."
)
params["imovel_id"] = t_imovel_schema.imovel_id
# ----------------------------------------------------
# Montagem do SQL dinâmico
# ----------------------------------------------------
sql = f"""
UPDATE T_IMOVEL
SET {', '.join(updates)}
WHERE IMOVEL_ID = :imovel_id
RETURNING *;
"""
# ----------------------------------------------------
# Execução e retorno
# ----------------------------------------------------
result = self.run_and_return(sql, params)
if not result:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Nenhum imóvel encontrado para o ID informado."
)
return result
except HTTPException:
raise
except Exception as e:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao atualizar o imóvel: {str(e)}"
)

View file

@ -0,0 +1,41 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeIdSchema
from fastapi import HTTPException, status
class TImovelUnidadeDeleteRepository(BaseRepository):
"""
Repositório para a operação de exclusão de um registro na tabela
t_censec_qualidade.
"""
def execute(self, t_imovel_unidade_schema: TImovelUnidadeIdSchema):
"""
Executa a consulta SQL para remover um registro pelo ID.
Args:
t_imovel_unidade_schema (TImovelUnidadeIdSchema): 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_IMOVEL_UNIDADE WHERE IMOVEL_UNIDADE_ID = :imovel_unidade_id """
# Preenchimento de parâmetros
params = {
"imovel_unidade_id": t_imovel_unidade_schema.imovel_unidade_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 IMOVEL_UNIDADE: {e}"
)

View file

@ -0,0 +1,63 @@
from abstracts.repository import BaseRepository
from api.packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeIndexSchema
class TImovelUnidadeIndexRepository(BaseRepository):
"""
Repositório para a operação de listagem de todos os registros
na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_unidade_index_schema: TImovelUnidadeIndexSchema):
"""
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 TIU.IMOVEL_UNIDADE_ID,
TIU.IMOVEL_ID,
TIU.NUMERO_UNIDADE,
TIU.QUADRA,
TIU.AREA,
TIU.SUPERQUADRA,
TIU.CONJUNTO,
TIU.BLOCO,
TIU.AREA_DESCRITIVA,
TIU.CARACTERISTICA,
TIU.RESERVA_FLORESTAL,
TIU.GEO_REFERENCIAMENTO,
TIU.LOGRADOURO,
TIU.TB_TIPOLOGRADOURO_ID,
TIU.SELECIONADO,
TIU.COMPLEMENTO,
TIU.TIPO_IMOVEL,
TIU.TIPO_CONSTRUCAO,
TIU.NUMERO_EDIFICACAO,
TIU.IPTU,
TIU.CCIR,
TIU.NIRF,
TIU.LOTE,
TIU.TORRE,
TIU.NOMELOTEAMENTO,
TIU.NOMECONDOMINIO,
TIU.NUMERO,
TIU.CNM_NUMERO,
TIU.IMOVEL_PUBLICO_UNIAO,
TIU.SPU_RIP,
TIU.CAT,
TIU.INSCRICAO_MUNICIPAL,
TIU.CIB,
TIU.AREA_CONSTRUIDA
FROM T_IMOVEL_UNIDADE TIU
WHERE TIU.IMOVEL_ID = :imovelId """
params = {
"imovelId": t_imovel_unidade_index_schema.imovel_id
}
# Execução do sql
response = self.fetch_all(sql, params)
# Retorna os dados localizados
return response

View file

@ -0,0 +1,156 @@
from fastapi import HTTPException, status
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_imovel_unidade_schema import TUnidadeImovelUnidadeSchema
class TImovelUnidadeSaveRepository(BaseRepository):
"""
Repositório para a operação de salvamento de um novo registro na tabela t_t_imovel_unidade.
"""
def execute(self, t_imovel_unidade_schema: TUnidadeImovelUnidadeSchema):
"""
Executa a operação de salvamento no banco de dados.
Args:
t_imovel_unidade_schema (TUnidadeImovelUnidadeSchema): 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 T_IMOVEL_UNIDADE (
IMOVEL_UNIDADE_ID,
IMOVEL_ID,
NUMERO_UNIDADE,
QUADRA,
AREA,
SUPERQUADRA,
CONJUNTO,
BLOCO,
AREA_DESCRITIVA,
CARACTERISTICA,
RESERVA_FLORESTAL,
GEO_REFERENCIAMENTO,
LOGRADOURO,
TB_TIPOLOGRADOURO_ID,
SELECIONADO,
COMPLEMENTO,
TIPO_IMOVEL,
TIPO_CONSTRUCAO,
TEXTO,
NUMERO_EDIFICACAO,
IPTU,
CCIR,
NIRF,
LOTE,
TORRE,
NOMELOTEAMENTO,
NOMECONDOMINIO,
NUMERO,
CNM_NUMERO,
IMOVEL_PUBLICO_UNIAO,
SPU_RIP,
CAT,
INSCRICAO_MUNICIPAL,
CIB,
AREA_CONSTRUIDA
) VALUES (
:imovel_unidade_id,
:imovel_id,
:numero_unidade,
:quadra,
:area,
:superquadra,
:conjunto,
:bloco,
:area_descritiva,
:caracteristica,
:reserva_florestal,
:geo_referenciamento,
:logradouro,
:tb_tipologradouro_id,
:selecionado,
:complemento,
:tipo_imovel,
:tipo_construcao,
:texto,
:numero_edificacao,
:iptu,
:ccir,
:nirf,
:lote,
:torre,
:nomeloteamento,
:nomecondominio,
:numero,
:cnm_numero,
:imovel_publico_uniao,
:spu_rip,
:cat,
:inscricao_municipal,
:cib,
:area_construida
)
RETURNING *;
"""
# ----------------------------------------------------
# Preenchimento dos parâmetros
# ----------------------------------------------------
params = {
'imovel_unidade_id': t_imovel_unidade_schema.imovel_unidade_id,
'imovel_id': t_imovel_unidade_schema.imovel_id,
'numero_unidade': t_imovel_unidade_schema.numero_unidade,
'quadra': t_imovel_unidade_schema.quadra,
'area': t_imovel_unidade_schema.area,
'superquadra': t_imovel_unidade_schema.superquadra,
'conjunto': t_imovel_unidade_schema.conjunto,
'bloco': t_imovel_unidade_schema.bloco,
'area_descritiva': t_imovel_unidade_schema.area_descritiva,
'caracteristica': t_imovel_unidade_schema.caracteristica,
'reserva_florestal': t_imovel_unidade_schema.reserva_florestal,
'geo_referenciamento': t_imovel_unidade_schema.geo_referenciamento,
'logradouro': t_imovel_unidade_schema.logradouro,
'tb_tipologradouro_id': t_imovel_unidade_schema.tb_tipologradouro_id,
'selecionado': t_imovel_unidade_schema.selecionado,
'complemento': t_imovel_unidade_schema.complemento,
'tipo_imovel': t_imovel_unidade_schema.tipo_imovel,
'tipo_construcao': t_imovel_unidade_schema.tipo_construcao,
'texto': t_imovel_unidade_schema.texto,
'numero_edificacao': t_imovel_unidade_schema.numero_edificacao,
'iptu': t_imovel_unidade_schema.iptu,
'ccir': t_imovel_unidade_schema.ccir,
'nirf': t_imovel_unidade_schema.nirf,
'lote': t_imovel_unidade_schema.lote,
'torre': t_imovel_unidade_schema.torre,
'nomeloteamento': t_imovel_unidade_schema.nomeloteamento,
'nomecondominio': t_imovel_unidade_schema.nomecondominio,
'numero': t_imovel_unidade_schema.numero,
'cnm_numero': t_imovel_unidade_schema.cnm_numero,
'imovel_publico_uniao': t_imovel_unidade_schema.imovel_publico_uniao,
'spu_rip': t_imovel_unidade_schema.spu_rip,
'cat': t_imovel_unidade_schema.cat,
'inscricao_municipal': t_imovel_unidade_schema.inscricao_municipal,
'cib': t_imovel_unidade_schema.cib,
'area_construida': t_imovel_unidade_schema.area_construida,
}
# 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 T_IMOVEL_UNIDADE: {e}"
)

View file

@ -0,0 +1,46 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeIdSchema
from fastapi import HTTPException, status
class TImovelUnidadeShowRepository(BaseRepository):
"""
Repositório para a operação de exibição de um registro na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_unidade_schema: TImovelUnidadeIdSchema):
"""
Busca um registro específico de CENSEC_QUALIDADE pelo ID.
Args:
t_imovel_unidade_schema (TImovelUnidadeIdSchema): 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_IMOVEL_UNIDADE WHERE IMOVEL_UNIDADE_ID = :imovel_unidade_Id"
# Preenchimento de parâmetros
params = {
'imovel_unidade_Id': t_imovel_unidade_schema.imovel_unidade_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_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao buscar registro: {str(e)}"
)

View file

@ -0,0 +1,110 @@
from abstracts.repository import BaseRepository
from packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeUpdateSchema
from fastapi import HTTPException, status
class TImovelUnidadeUpdateRepository(BaseRepository):
"""
Repositório para a operação de atualização na tabela T_CENSEC_QUALIDADE.
"""
def execute(self, t_imovel_unidade_id : int, t_imovel_unidade_schema: TImovelUnidadeUpdateSchema):
"""
Executa a atualização de um registro na tabela.
Args:
t_imovel_unidade_id (int): O ID do registro a ser atualizado.
t_imovel_unidade_schema (TImovelUnidadeUpdateSchema): 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:
# Dicionário de mapeamento: atributo → coluna SQL
field_map = {
"imovel_id": "IMOVEL_ID",
"numero_unidade": "NUMERO_UNIDADE",
"quadra": "QUADRA",
"area": "AREA",
"superquadra": "SUPERQUADRA",
"conjunto": "CONJUNTO",
"bloco": "BLOCO",
"area_descritiva": "AREA_DESCRITIVA",
"caracteristica": "CARACTERISTICA",
"reserva_florestal": "RESERVA_FLORESTAL",
"geo_referenciamento": "GEO_REFERENCIAMENTO",
"logradouro": "LOGRADOURO",
"tb_tipologradouro_id": "TB_TIPOLOGRADOURO_ID",
"selecionado": "SELECIONADO",
"complemento": "COMPLEMENTO",
"tipo_imovel": "TIPO_IMOVEL",
"tipo_construcao": "TIPO_CONSTRUCAO",
"texto": "TEXTO",
"numero_edificacao": "NUMERO_EDIFICACAO",
"iptu": "IPTU",
"ccir": "CCIR",
"nirf": "NIRF",
"lote": "LOTE",
"torre": "TORRE",
"nomeloteamento": "NOMELOTEAMENTO",
"nomecondominio": "NOMECONDOMINIO",
"numero": "NUMERO",
"cnm_numero": "CNM_NUMERO",
"imovel_publico_uniao": "IMOVEL_PUBLICO_UNIAO",
"spu_rip": "SPU_RIP",
"cat": "CAT",
"inscricao_municipal": "INSCRICAO_MUNICIPAL",
"cib": "CIB",
"area_construida": "AREA_CONSTRUIDA"
}
# Lista de campos que serão atualizados
updates = []
params = {}
# Monta dinamicamente os campos presentes no schema
for attr, column in field_map.items():
value = getattr(t_imovel_unidade_schema, attr, None)
if value is not None:
updates.append(f"{column} = :{attr}")
params[attr] = value
# Nenhum campo a atualizar?
if not updates:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Nenhum campo válido foi enviado para atualização."
)
# ID obrigatório
params["t_imovel_unidade_id"] = t_imovel_unidade_id
# Monta o SQL dinâmico
sql = f"""
UPDATE T_IMOVEL_UNIDADE
SET {', '.join(updates)}
WHERE IMOVEL_UNIDADE_ID = :t_imovel_unidade_id
RETURNING *;
"""
# Executa a query
result = self.run_and_return(sql, params)
if not result:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Nenhuma unidade de imóvel encontrada para o ID informado."
)
return result
except HTTPException:
# Deixa HTTPException original passar direto
raise
except Exception as e:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Erro ao atualizar o imóvel unidade: {str(e)}"
)

View file

@ -0,0 +1,135 @@
from pydantic import BaseModel, field_validator, model_validator
from fastapi import HTTPException, status
from typing import Optional
from datetime import datetime
# Função de sanitização
from actions.validations.text import Text
# ----------------------------------------------------
# Schema base
# ----------------------------------------------------
class TImovelSchema(BaseModel):
imovel_id: Optional[int] = None
tipo_classe: Optional[str] = None
tipo_registro: Optional[str] = None
data_registro: Optional[datetime] = None
numero: Optional[float] = None
numero_letra: Optional[str] = None
cidade: Optional[str] = None
cep: Optional[float] = None
uf: Optional[str] = None
tb_bairro_id: Optional[int] = None
cartorio: Optional[str] = None
livro: Optional[str] = None
cns: Optional[float] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema base
# ----------------------------------------------------
class TImovelIndexSchema(BaseModel):
tipo_classe: str = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar um IMÓVEL pelo ID (GET)
# ----------------------------------------------------
class TImovelIdSchema(BaseModel):
imovel_id: int
# ----------------------------------------------------
# Schema para criação de novo IMÓVEL (POST)
# ----------------------------------------------------
class TImovelSaveSchema(BaseModel):
imovel_id: Optional[int] = None
tipo_classe: str
tipo_registro: str
data_registro: datetime
numero: float
numero_letra: Optional[str] = None
cidade: str
cep: Optional[float] = None
uf: str
tb_bairro_id: Optional[int] = None
cartorio: Optional[str] = None
livro: Optional[str] = None
cns: Optional[float] = None
# Sanitiza os campos textuais
@field_validator('tipo_classe', 'tipo_registro', 'numero_letra', 'cidade', 'uf', 'cartorio', 'livro')
def sanitize_fields(cls, v):
if v:
return Text.sanitize_input(v)
return v
# Validações adicionais
@model_validator(mode='after')
def validate_fields(self):
errors = []
if not self.tipo_classe or len(self.tipo_classe.strip()) == 0:
errors.append({'input': 'tipo_classe', 'message': 'O tipo de classe é obrigatório.'})
if not self.tipo_registro or len(self.tipo_registro.strip()) == 0:
errors.append({'input': 'tipo_registro', 'message': 'O tipo de registro é obrigatório.'})
if not self.data_registro:
errors.append({'input': 'data_registro', 'message': 'A data de registro é obrigatória.'})
if not self.numero:
errors.append({'input': 'numero', 'message': 'O número é obrigatório.'})
if not self.cidade or len(self.cidade.strip()) == 0:
errors.append({'input': 'cidade', 'message': 'A cidade é obrigatória.'})
if not self.uf or len(self.uf.strip()) == 0:
errors.append({'input': 'uf', 'message': 'A UF é obrigatória.'})
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=errors
)
return self
# ----------------------------------------------------
# Schema para atualização de IMÓVEL (PUT)
# ----------------------------------------------------
class TImovelUpdateSchema(BaseModel):
tipo_classe: Optional[str] = None
tipo_registro: Optional[str] = None
data_registro: Optional[datetime] = None
numero: Optional[float] = None
numero_letra: Optional[str] = None
cidade: Optional[str] = None
cep: Optional[float] = None
uf: Optional[str] = None
tb_bairro_id: Optional[int] = None
cartorio: Optional[str] = None
livro: Optional[str] = None
cns: Optional[float] = None
# Sanitiza os campos textuais
@field_validator('tipo_classe', 'tipo_registro', 'numero_letra', 'cidade', 'uf', 'cartorio', 'livro')
def sanitize_fields(cls, v):
if v:
return Text.sanitize_input(v)
return v
# Garante que ao menos um campo seja enviado
@model_validator(mode='after')
def validate_partial_update(self):
if not any([
self.tipo_classe, self.tipo_registro, self.data_registro, self.numero,
self.numero_letra, self.cidade, self.cep, self.uf, self.tb_bairro_id,
self.cartorio, self.livro, self.cns
]):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="Pelo menos um campo deve ser fornecido para a atualização."
)
return self

View file

@ -0,0 +1,230 @@
from pydantic import BaseModel, field_validator, model_validator
from fastapi import HTTPException, status
from typing import Optional
from datetime import datetime
# Função de sanitização
from actions.validations.text import Text
# ----------------------------------------------------
# Schema base para TImovelUnidade
# ----------------------------------------------------
class TImovelUnidadeSchema(BaseModel):
imovel_unidade_id: Optional[int] = None
imovel_id: Optional[int] = None
numero_unidade: Optional[str] = None
quadra: Optional[str] = None
area: Optional[float] = None
superquadra: Optional[str] = None
conjunto: Optional[str] = None
bloco: Optional[str] = None
area_descritiva: Optional[str] = None
caracteristica: Optional[str] = None
reserva_florestal: Optional[str] = None
geo_referenciamento: Optional[str] = None
logradouro: Optional[str] = None
tb_tipologradouro_id: Optional[int] = None
selecionado: Optional[str] = None
complemento: Optional[str] = None
tipo_imovel: Optional[int] = None
tipo_construcao: Optional[int] = None
texto: Optional[bytes] = None
numero_edificacao: Optional[str] = None
iptu: Optional[str] = None
ccir: Optional[str] = None
nirf: Optional[str] = None
lote: Optional[str] = None
torre: Optional[str] = None
nomeloteamento: Optional[str] = None
nomecondominio: Optional[str] = None
numero: Optional[float] = None
cnm_numero: Optional[str] = None
imovel_publico_uniao: Optional[str] = None
spu_rip: Optional[str] = None
cat: Optional[str] = None
inscricao_municipal: Optional[str] = None
cib: Optional[str] = None
area_construida: Optional[float] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema base para TImovelUnidadeIndex
# ----------------------------------------------------
class TImovelUnidadeIndexSchema(BaseModel):
imovel_id: Optional[int] = None
class Config:
from_attributes = True
# ----------------------------------------------------
# Schema para localizar uma unidade pelo ID (GET)
# ----------------------------------------------------
class TImovelUnidadeIdSchema(BaseModel):
imovel_unidade_id: int
# ----------------------------------------------------
# Schema para criação de nova unidade (POST)
# ----------------------------------------------------
class TImovelUnidadeSaveSchema(BaseModel):
imovel_id: int
numero_unidade: str
area: float
logradouro: str
tb_tipologradouro_id: int
texto: Optional[bytes] = None
# Outros campos opcionais
quadra: Optional[str] = None
superquadra: Optional[str] = None
conjunto: Optional[str] = None
bloco: Optional[str] = None
area_descritiva: Optional[str] = None
caracteristica: Optional[str] = None
reserva_florestal: Optional[str] = None
geo_referenciamento: Optional[str] = None
selecionado: Optional[str] = None
complemento: Optional[str] = None
tipo_imovel: Optional[int] = None
tipo_construcao: Optional[int] = None
numero_edificacao: Optional[str] = None
iptu: Optional[str] = None
ccir: Optional[str] = None
nirf: Optional[str] = None
lote: Optional[str] = None
torre: Optional[str] = None
nomeloteamento: Optional[str] = None
nomecondominio: Optional[str] = None
numero: Optional[float] = None
cnm_numero: Optional[str] = None
imovel_publico_uniao: Optional[str] = None
spu_rip: Optional[str] = None
cat: Optional[str] = None
inscricao_municipal: Optional[str] = None
cib: Optional[str] = None
area_construida: Optional[float] = None
# Sanitiza campos textuais
@field_validator(
'numero_unidade', 'quadra', 'superquadra', 'conjunto', 'bloco',
'area_descritiva', 'caracteristica', 'reserva_florestal', 'geo_referenciamento',
'logradouro', 'selecionado', 'complemento', 'numero_edificacao',
'iptu', 'ccir', 'nirf', 'lote', 'torre', 'nomeloteamento', 'nomecondominio',
'cnm_numero', 'imovel_publico_uniao', 'spu_rip', 'cat', 'inscricao_municipal', 'cib'
)
def sanitize_fields(cls, v):
if v:
return Text.sanitize_input(v)
return v
# Valida campo BLOB
@field_validator('texto')
def validate_blob(cls, v):
if v is not None and not isinstance(v, bytes):
raise ValueError('O campo TEXTO deve ser do tipo bytes.')
return v
# Valida campos obrigatórios
@model_validator(mode='after')
def validate_required_fields(self):
errors = []
if not self.imovel_id:
errors.append({'input': 'imovel_id', 'message': 'O ID do imóvel é obrigatório.'})
if not self.numero_unidade or len(self.numero_unidade.strip()) == 0:
errors.append({'input': 'numero_unidade', 'message': 'O número da unidade é obrigatório.'})
if not self.area:
errors.append({'input': 'area', 'message': 'A área é obrigatória.'})
if not self.logradouro or len(self.logradouro.strip()) == 0:
errors.append({'input': 'logradouro', 'message': 'O logradouro é obrigatório.'})
if not self.tb_tipologradouro_id:
errors.append({'input': 'tb_tipologradouro_id', 'message': 'O tipo de logradouro é obrigatório.'})
if errors:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=errors)
return self
# ----------------------------------------------------
# Schema para atualização de unidade (PUT)
# ----------------------------------------------------
class TImovelUnidadeUpdateSchema(BaseModel):
imovel_id: Optional[int] = None
numero_unidade: Optional[str] = None
area: Optional[float] = None
logradouro: Optional[str] = None
tb_tipologradouro_id: Optional[int] = None
texto: Optional[bytes] = None
# Outros campos opcionais
quadra: Optional[str] = None
superquadra: Optional[str] = None
conjunto: Optional[str] = None
bloco: Optional[str] = None
area_descritiva: Optional[str] = None
caracteristica: Optional[str] = None
reserva_florestal: Optional[str] = None
geo_referenciamento: Optional[str] = None
selecionado: Optional[str] = None
complemento: Optional[str] = None
tipo_imovel: Optional[int] = None
tipo_construcao: Optional[int] = None
numero_edificacao: Optional[str] = None
iptu: Optional[str] = None
ccir: Optional[str] = None
nirf: Optional[str] = None
lote: Optional[str] = None
torre: Optional[str] = None
nomeloteamento: Optional[str] = None
nomecondominio: Optional[str] = None
numero: Optional[float] = None
cnm_numero: Optional[str] = None
imovel_publico_uniao: Optional[str] = None
spu_rip: Optional[str] = None
cat: Optional[str] = None
inscricao_municipal: Optional[str] = None
cib: Optional[str] = None
area_construida: Optional[float] = None
# Sanitiza campos textuais
@field_validator(
'numero_unidade', 'quadra', 'superquadra', 'conjunto', 'bloco',
'area_descritiva', 'caracteristica', 'reserva_florestal', 'geo_referenciamento',
'logradouro', 'selecionado', 'complemento', 'numero_edificacao',
'iptu', 'ccir', 'nirf', 'lote', 'torre', 'nomeloteamento', 'nomecondominio',
'cnm_numero', 'imovel_publico_uniao', 'spu_rip', 'cat', 'inscricao_municipal', 'cib'
)
def sanitize_fields(cls, v):
if v:
return Text.sanitize_input(v)
return v
# Valida campo BLOB
@field_validator('texto')
def validate_blob(cls, v):
if v is not None and not isinstance(v, bytes):
raise ValueError('O campo TEXTO deve ser do tipo bytes.')
return v
# Garante que ao menos um campo seja enviado
@model_validator(mode='after')
def validate_partial_update(self):
if not any([
self.imovel_id, self.numero_unidade, self.area, self.logradouro,
self.tb_tipologradouro_id, self.texto, self.quadra, self.superquadra,
self.conjunto, self.bloco, self.area_descritiva, self.caracteristica,
self.reserva_florestal, self.geo_referenciamento, self.selecionado,
self.complemento, self.tipo_imovel, self.tipo_construcao, self.numero_edificacao,
self.iptu, self.ccir, self.nirf, self.lote, self.torre, self.nomeloteamento,
self.nomecondominio, self.numero, self.cnm_numero, self.imovel_publico_uniao,
self.spu_rip, self.cat, self.inscricao_municipal, self.cib, self.area_construida
]):
return self
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="Pelo menos um campo deve ser fornecido para a atualização."
)

View file

@ -0,0 +1,30 @@
from api.packages.v1.administrativo.actions.t_imovel.t_imovel_delete_action import TImovelDeleteAction
from api.packages.v1.administrativo.schemas.t_imovel_schema import TImovelIdSchema
class TImovelDeleteService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_id_schema: TImovelIdSchema):
"""
Executa a operação de exclusão do registro no banco de dados.
Args:
t_imovel_schema (TCensecQualidadeIdSchema): 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_imovel_delete_action = TImovelDeleteAction()
# Executa a ação em questão
data = t_imovel_delete_action.execute(t_imovel_id_schema)
# Retorno da informação
return data

View file

@ -0,0 +1,33 @@
from api.packages.v1.administrativo.actions.t_imovel.t_imovel_index_action import TImovelIndexAction
from api.packages.v1.administrativo.schemas.t_imovel_schema import TImovelIndexSchema
from fastapi import HTTPException, status
class TImovelIndexService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de registros na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_index_schema: TImovelIndexSchema):
"""
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_imovel_index_action = TImovelIndexAction()
# Executa a busca de todas as ações
data = t_imovel_index_action.execute(t_imovel_index_schema)
# Verifica se foram localizados registros
if not data:
# Retorna uma exceção
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail='Não foi possível localizar os registros de T_IMOVEL_UNIDADE'
)
# Retorna as informações localizadas
return data

View file

@ -0,0 +1,46 @@
from api.actions.dynamic_import.dynamic_import import DynamicImport
from api.packages.v1.administrativo.actions.t_imovel.t_imovel_save_action import TImovelSaveAction
from api.packages.v1.sequencia.schemas.g_sequencia import GSequenciaSchema
from api.packages.v1.administrativo.schemas.t_imovel_schema import TImovelSaveSchema
from api.packages.v1.sequencia.services.g_sequencia.checkout_service import GenerateService
from fastapi import HTTPException, status
class TImovelSaveService:
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_imovel")
pass
# Cadastra o novo CENSEC_QUALIDADE
def execute(self, t_imovel_save_schema: TImovelSaveSchema):
# Verifica se precisa gerar o ID de sequência
if not t_imovel_save_schema.imovel_id:
# Crio um objeto de sequencia
sequencia_schema = GSequenciaSchema()
# Define os dados para atualizar a sequencia
sequencia_schema.tabela = 'T_IMOVEL'
# Busco a sequência atualizada
generate = GenerateService()
# Busco a sequência atualizada
sequencia = generate.execute(sequencia_schema)
# Atualiza os dados da chave primária
t_imovel_save_schema.imovel_id = sequencia.sequencia
# Instanciamento de ações
t_imovel_save_action = TImovelSaveAction()
# Retorna o resultado da operação
return t_imovel_save_action.execute(t_imovel_save_schema)

View file

@ -0,0 +1,35 @@
from api.packages.v1.administrativo.actions.t_imovel_unidade.t_imovel_unidade_show_action import TImovelUnidadeShowAction
from api.packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeIdSchema
from fastapi import HTTPException, status
class TImovelUnidadeShowService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_unidade_id_schema: TImovelUnidadeIdSchema):
"""
Executa a operação de busca no banco de dados.
Args:
t_imovel_unidade_schema (TCensecQualidadeIdSchema): O esquema com o ID a ser buscado.
Returns:
O resultado da busca.
"""
# Instanciamento da ação
t_imovel_unidade_show_action = TImovelUnidadeShowAction()
# Executa a ação em questão
data = t_imovel_unidade_show_action.execute(t_imovel_unidade_id_schema)
if not data:
# Retorna uma exceção
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail='Não foi possível localizar o registro de T_IMOVEL_UNIDADE'
)
# Retorno da informação
return data

View file

@ -0,0 +1,25 @@
from api.packages.v1.administrativo.actions.t_imovel_unidade.t_imovel_unidade_update_action import TImovelUnidadeUpdateAction
from api.packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeUpdateSchema
class TImovelUnidadeUpdateService:
"""
Serviço para a operação de atualização de um registro na tabela
t_censec_qualidade.
"""
def execute(self, t_imovel_unidade_update_schema: TImovelUnidadeUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
t_imovel_unidade_schema (TCensecQualidadeUpdateSchema): O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# Instanciamento de ações
t_imovel_unidade_update_action = TImovelUnidadeUpdateAction()
# Retorna o resultado da operação
return t_imovel_unidade_update_action.execute(t_imovel_unidade_update_schema)

View file

@ -0,0 +1,28 @@
from api.packages.v1.administrativo.actions.t_imovel_unidade.t_imovel_unidade_delete_action import TImovelUnidadeDeleteAction
from api.packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeIdSchema
class TImovelUnidadeDeleteService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de exclusão de um registro na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_unidade_schema: TImovelUnidadeIdSchema):
"""
Executa a operação de exclusão do registro no banco de dados.
Args:
t_imovel_unidade_schema (TCensecQualidadeIdSchema): 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_imovel_unidade_delete = TImovelUnidadeDeleteAction()
# Executa a ação em questão
data = t_imovel_unidade_delete.execute(t_imovel_unidade_schema)
# Retorno da informação
return data

View file

@ -0,0 +1,33 @@
from api.packages.v1.administrativo.actions.t_imovel_unidade.t_imovel_unidade_index_action import TImovelUnidadeIndexAction
from api.packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeIndexSchema
from fastapi import HTTPException, status
class TImovelUnidadeIndexService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de listagem de registros na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_unidade_index_schema: TImovelUnidadeIndexSchema):
"""
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_imovel_unidade_index_action = TImovelUnidadeIndexAction()
# Executa a busca de todas as ações
data = t_imovel_unidade_index_action.execute(t_imovel_unidade_index_schema)
# Verifica se foram localizados registros
if not data:
# Retorna uma exceção
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail='Não foi possível localizar os registros de T_IMOVEL_UNIDADE'
)
# Retorna as informações localizadas
return data

View file

@ -0,0 +1,49 @@
from actions.dynamic_import.dynamic_import import DynamicImport
from api.packages.v1.administrativo.actions.t_imovel_unidade.t_imovel_unidade_save_action import TImovelUnidadeSaveAction
from api.packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeSchema
from packages.v1.sequencia.schemas.g_sequencia import GSequenciaSchema
from packages.v1.sequencia.services.g_sequencia.generate_service import GenerateService
from fastapi import HTTPException, status
class TImovelUnidadeSaveService:
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_imovel_unidade")
pass
# Cadastra o novo CENSEC_QUALIDADE
def execute(self, t_imovel_unidade_schema: TImovelUnidadeSchema):
# Armazena possíveis erros
errors = []
# Verifica se precisa gerar o ID de sequência
if not t_imovel_unidade_schema.imovel_unidade_id:
# Crio um objeto de sequencia
sequencia_schema = GSequenciaSchema()
# Define os dados para atualizar a sequencia
sequencia_schema.tabela = 'T_CENSEC_QUALIDADE'
# Busco a sequência atualizada
generate = GenerateService()
# Busco a sequência atualizada
sequencia = generate.execute(sequencia_schema)
# Atualiza os dados da chave primária
t_imovel_unidade_schema.imovel_unidade_id = sequencia.sequencia
# Instanciamento de ações
t_imovel_unidade_save_action = TImovelUnidadeSaveAction()
# Retorna o resultado da operação
return t_imovel_unidade_save_action.execute(t_imovel_unidade_schema)

View file

@ -0,0 +1,35 @@
from api.packages.v1.administrativo.actions.t_imovel_unidade.t_imovel_unidade_show_action import TImovelUnidadeShowAction
from api.packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeIdSchema
from fastapi import HTTPException, status
class TImovelUnidadeShowService:
"""
Serviço responsável por encapsular a lógica de negócio para a operação
de busca de um registro na tabela t_censec_qualidade.
"""
def execute(self, t_imovel_unidade_id_schema: TImovelUnidadeIdSchema):
"""
Executa a operação de busca no banco de dados.
Args:
t_imovel_unidade_schema (TCensecQualidadeIdSchema): O esquema com o ID a ser buscado.
Returns:
O resultado da busca.
"""
# Instanciamento da ação
t_imovel_unidade_show_action = TImovelUnidadeShowAction()
# Executa a ação em questão
data = t_imovel_unidade_show_action.execute(t_imovel_unidade_id_schema)
if not data:
# Retorna uma exceção
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail='Não foi possível localizar o registro de T_IMOVEL_UNIDADE'
)
# Retorno da informação
return data

View file

@ -0,0 +1,25 @@
from api.packages.v1.administrativo.actions.t_imovel_unidade.t_imovel_unidade_update_action import TImovelUnidadeUpdateAction
from api.packages.v1.administrativo.schemas.t_imovel_unidade_schema import TImovelUnidadeUpdateSchema
class TImovelUnidadeUpdateService:
"""
Serviço para a operação de atualização de um registro na tabela
t_censec_qualidade.
"""
def execute(self, t_imovel_unidade_update_schema: TImovelUnidadeUpdateSchema):
"""
Executa a operação de atualização no banco de dados.
Args:
t_imovel_unidade_schema (TCensecQualidadeUpdateSchema): O esquema com os dados a serem atualizados.
Returns:
O resultado da operação de atualização.
"""
# Instanciamento de ações
t_imovel_unidade_update_action = TImovelUnidadeUpdateAction()
# Retorna o resultado da operação
return t_imovel_unidade_update_action.execute(t_imovel_unidade_update_schema)

View file

@ -1,4 +1,5 @@
# 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
@ -26,6 +27,7 @@ from packages.v1.administrativo.endpoints import t_servico_tipo_endpoint
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
# Cria uma instância do APIRouter que vai agregar todas as rotas da API
api_router = APIRouter()
@ -152,4 +154,14 @@ api_router.include_router(
# Inclui as rotas de g_uf
api_router.include_router(
g_uf_endpoint.router, prefix="/administrativo/g_uf", tags=["Estados"]
)
# Inclui as rotas de t_imovel
api_router.include_router(
t_imovel_endpoint.router, prefix="/administrativo/t_imovel", tags=["Imóveis"]
)
# 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"]
)