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

feat(): Criação da visualização do ato e suas dependências, montando o json conforme solicitação TJGO

Browse source
This commit is contained in:
Kenio 2025-11-04 15:00:11 -03:00
parent e020bfdd2d
commit 682cb61ab9
7 changed files with 435 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
packages/v1/administrativo/actions/ato_principal/ato_principal_show_atos_action.py Normal file
View file

@ -0,0 +1,21 @@
from abstracts.action import BaseAction
from packages.v1.administrativo.schemas.ato_principal_schema import (
AtoPrincipalCodigoAtoSchema,
)
from packages.v1.administrativo.repositories.ato_principal.ato_principal_show_atos_repository import (
ShowAtosRepository,
)
class ShowAtosAction(BaseAction):
def execute(self, ato_principal_schema: AtoPrincipalCodigoAtoSchema):
# Instânciamento do repositório sql
show_repository = ShowAtosRepository()
# Execução do sql
response = show_repository.execute(ato_principal_schema)
# Retorno da informação
return response

18
packages/v1/administrativo/controllers/ato_principal_controller.py
View file

@ -13,6 +13,9 @@ from packages.v1.administrativo.services.ato_principal.ato_principal_save_multip
from packages.v1.administrativo.services.ato_principal.ato_principal_show_service import (
ShowService,
)
from packages.v1.administrativo.services.ato_principal.ato_principal_show_atos_service import (
ShowAtosService,
)
from packages.v1.administrativo.services.ato_principal.ato_principal_update_service import (
UpdateService,
)
@ -26,6 +29,7 @@ from packages.v1.administrativo.schemas.ato_principal_schema import (
AtoPrincipalSaveSchema,
AtoPrincipalUpdateSchema,
AtoPrincipalIdSchema,
AtoPrincipalCodigoAtoSchema,
)
@ -59,8 +63,6 @@ class AtoPrincipalController:
}
# Busca um ato principal especifica pelo ID
# Note: O método 'showAtoPrincipal' foi removido por não se aplicar
# à entidade principal.
def show(self, ato_principal_schema: AtoPrincipalSchema):
# Instânciamento da classe desejada
@ -72,6 +74,18 @@ class AtoPrincipalController:
"data": show_service.execute(ato_principal_schema),
}
# Busca um ato principal especifico e suas dependências pelo codigo do ato
def showAtos(self, ato_principal_schema: AtoPrincipalCodigoAtoSchema):
# Instânciamento da classe desejada
show_service = ShowAtosService()
# Busca e retorna o ato principal desejado
return {
"message": "Ato Principal localizado com sucesso",
"data": show_service.execute(ato_principal_schema),
}
# Cadastra um novo ato principal
def save(self, ato_principal_schema: AtoPrincipalSaveSchema):

36
packages/v1/administrativo/endpoints/atos_endpoint.py Normal file
View file

@ -0,0 +1,36 @@
# Importação de bibliotecas
from typing import List
from typing import Optional
from fastapi import APIRouter, Body, Depends, status
from actions.jwt.get_current_user import get_current_user
from packages.v1.administrativo.controllers.ato_principal_controller import (
AtoPrincipalController,
)
from packages.v1.administrativo.schemas.ato_principal_schema import (
AtoPrincipalCodigoAtoSchema,
)
# Inicializa o roteador para as rotas de ato principal
router = APIRouter()
# Instânciamento do controller desejado
ato_principal_controller = AtoPrincipalController()
# Localiza um ato principal pelo ID
@router.get(
"/{codigo_ato}",
status_code=status.HTTP_200_OK,
summary="Busca um registro em especifico pelo código do ato",
response_description="Busca um registro em especifico e suas dependências",
)
async def showAtos(codigo_ato: str, current_user: dict = Depends(get_current_user)):
# Cria o schema com os dados recebidos
ato_principal_schema = AtoPrincipalCodigoAtoSchema(codigo_ato=codigo_ato)
# Busca um ato principal especifico pelo ID
response = ato_principal_controller.showAtos(ato_principal_schema)
# Retorna os dados localizados
return response

279
packages/v1/administrativo/repositories/ato_principal/ato_principal_show_atos_repository.py Normal file
View file

@ -0,0 +1,279 @@
from typing import Optional, Dict, Any, List
from sqlalchemy import func
from fastapi import HTTPException, status
from database.mysql import SessionLocal, get_database_settings
from packages.v1.administrativo.models.ato_principal_model import AtoPrincipal
from packages.v1.administrativo.models.ato_parte_model import AtoParte
from packages.v1.administrativo.models.ato_documento_model import AtoDocumento
from packages.v1.administrativo.schemas.ato_principal_schema import (
AtoPrincipalCodigoAtoSchema,
)
from datetime import datetime
# === Recupera configurações e chave AES ===
DB_SETTINGS = get_database_settings()
AES_KEY = getattr(DB_SETTINGS, "aeskey", None)
class ShowAtosRepository:
"""
Repositório responsável por retornar:
- o ato principal (buscado por codigo_ato)
- suas partes e documentos
- seus atos vinculados (e respectivos documentos e partes)
"""
def execute(
self, codigo_ato_schema: AtoPrincipalCodigoAtoSchema
) -> Optional[Dict[str, Any]]:
db = SessionLocal()
try:
codigo_ato = codigo_ato_schema.codigo_ato
# === 1. Buscar ato principal ===
ato = (
db.query(
AtoPrincipal.ato_principal_id,
AtoPrincipal.origem_ato_principal_id,
AtoPrincipal.identificacao_pedido_cgj,
AtoPrincipal.tipo_ato,
AtoPrincipal.codigo_selo,
AtoPrincipal.codigo_ato,
func.AES_DECRYPT(AtoPrincipal.nome_civil_ato, AES_KEY).label(
"nome_civil_ato"
),
func.AES_DECRYPT(
AtoPrincipal.nome_serventuario_praticou_ato, AES_KEY
).label("nome_serventuario_praticou_ato"),
func.AES_DECRYPT(AtoPrincipal.inteiro_teor, AES_KEY).label(
"inteiro_teor"
),
AtoPrincipal.data_solicitacao,
AtoPrincipal.ip_maquina,
AtoPrincipal.valor_entrada,
AtoPrincipal.emolumento,
AtoPrincipal.taxa_judiciaria,
AtoPrincipal.fundos_estaduais,
AtoPrincipal.protocolo_protesto,
AtoPrincipal.protocolo_imovel,
)
.filter(AtoPrincipal.codigo_ato == codigo_ato)
.first()
)
if not ato:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Ato com codigo_ato '{codigo_ato}' não encontrado.",
)
# === 2. Buscar partes do ato principal ===
partes = (
db.query(
func.AES_DECRYPT(AtoParte.nome, AES_KEY).label("nome"),
func.AES_DECRYPT(AtoParte.telefone, AES_KEY).label("telefone"),
func.AES_DECRYPT(AtoParte.cpf_cnpj, AES_KEY).label("cpf_cnpj"),
)
.filter(AtoParte.ato_principal_id == ato.ato_principal_id)
.all()
)
partes_list = [
{
"nome": p.nome.decode("utf-8") if p.nome else None,
"telefone": p.telefone.decode("utf-8") if p.telefone else None,
"cpf_cnpj": p.cpf_cnpj.decode("utf-8") if p.cpf_cnpj else None,
}
for p in partes
]
# === 3. Buscar documentos do ato principal ===
documentos = (
db.query(
func.AES_DECRYPT(AtoDocumento.url, AES_KEY).label("url"),
func.AES_DECRYPT(AtoDocumento.nome_documento, AES_KEY).label(
"nome_documento"
),
func.AES_DECRYPT(AtoDocumento.tipo_documento, AES_KEY).label(
"tipo_documento"
),
)
.filter(AtoDocumento.ato_principal_id == ato.ato_principal_id)
.all()
)
documentos_list = [
{
"url": d.url.decode("utf-8") if d.url else None,
"nome_documento": (
d.nome_documento.decode("utf-8") if d.nome_documento else None
),
"tipo_documento": (
d.tipo_documento.decode("utf-8") if d.tipo_documento else None
),
}
for d in documentos
]
# === 4. Buscar atos vinculados ===
atos_vinculados = (
db.query(
AtoPrincipal.ato_principal_id,
AtoPrincipal.identificacao_pedido_cgj,
AtoPrincipal.tipo_ato,
AtoPrincipal.codigo_selo,
AtoPrincipal.codigo_ato,
func.AES_DECRYPT(AtoPrincipal.nome_civil_ato, AES_KEY).label(
"nome_civil_ato"
),
func.AES_DECRYPT(
AtoPrincipal.nome_serventuario_praticou_ato, AES_KEY
).label("nome_serventuario_praticou_ato"),
func.AES_DECRYPT(AtoPrincipal.inteiro_teor, AES_KEY).label(
"inteiro_teor"
),
AtoPrincipal.data_solicitacao,
AtoPrincipal.ip_maquina,
AtoPrincipal.valor_entrada,
AtoPrincipal.emolumento,
AtoPrincipal.taxa_judiciaria,
AtoPrincipal.fundos_estaduais,
AtoPrincipal.protocolo_protesto,
AtoPrincipal.protocolo_imovel,
)
.filter(AtoPrincipal.origem_ato_principal_id == ato.ato_principal_id)
.all()
)
atos_vinculados_list = []
for av in atos_vinculados:
# === Documentos do ato vinculado ===
documentos_vinc = (
db.query(
func.AES_DECRYPT(AtoDocumento.url, AES_KEY).label("url"),
func.AES_DECRYPT(AtoDocumento.nome_documento, AES_KEY).label(
"nome_documento"
),
func.AES_DECRYPT(AtoDocumento.tipo_documento, AES_KEY).label(
"tipo_documento"
),
)
.filter(AtoDocumento.ato_principal_id == av.ato_principal_id)
.all()
)
documentos_vinc_list = [
{
"url": d.url.decode("utf-8") if d.url else None,
"nome_documento": (
d.nome_documento.decode("utf-8")
if d.nome_documento
else None
),
"tipo_documento": (
d.tipo_documento.decode("utf-8")
if d.tipo_documento
else None
),
}
for d in documentos_vinc
]
# === Partes do ato vinculado ===
partes_vinc = (
db.query(
func.AES_DECRYPT(AtoParte.nome, AES_KEY).label("nome"),
func.AES_DECRYPT(AtoParte.telefone, AES_KEY).label("telefone"),
func.AES_DECRYPT(AtoParte.cpf_cnpj, AES_KEY).label("cpf_cnpj"),
)
.filter(AtoParte.ato_principal_id == av.ato_principal_id)
.all()
)
partes_vinc_list = [
{
"nome": p.nome.decode("utf-8") if p.nome else None,
"telefone": p.telefone.decode("utf-8") if p.telefone else None,
"cpf_cnpj": p.cpf_cnpj.decode("utf-8") if p.cpf_cnpj else None,
}
for p in partes_vinc
]
atos_vinculados_list.append(
{
"inteiro_teor": (
av.inteiro_teor.decode("utf-8") if av.inteiro_teor else None
),
"identificacao_pedido_na_cgj": av.identificacao_pedido_cgj,
"tipo_de_ato": av.tipo_ato,
"codigo_do_selo": av.codigo_selo,
"codigo_do_ato": av.codigo_ato,
"nome_do_civil_do_ato": (
av.nome_civil_ato.decode("utf-8")
if av.nome_civil_ato
else None
),
"nome_do_serventuario_que_praticou_ato": (
av.nome_serventuario_praticou_ato.decode("utf-8")
if av.nome_serventuario_praticou_ato
else None
),
"data_hora_da_solicitacao": (
av.data_solicitacao.isoformat()
if av.data_solicitacao
else None
),
"ip_da_maquina_que_praticou_ato": av.ip_maquina,
"valor_de_entrada_do_ato": av.valor_entrada,
"emolumento_do_ato": av.emolumento,
"taxa_judiciaria_do_ato": av.taxa_judiciaria,
"fundos_estaduais_do_ato": av.fundos_estaduais,
"protocolo_do_protesto": av.protocolo_protesto,
"protocolo_do_imovel": av.protocolo_imovel,
"documentos": documentos_vinc_list,
"partes": partes_vinc_list,
}
)
# === 5. Montar JSON final ===
return {
"inteiro_teor": (
ato.inteiro_teor.decode("utf-8") if ato.inteiro_teor else None
),
"identificacao_pedido_na_cgj": ato.identificacao_pedido_cgj,
"tipo_de_ato": ato.tipo_ato,
"codigo_do_selo": ato.codigo_selo,
"codigo_do_ato": ato.codigo_ato,
"nome_do_civil_do_ato": (
ato.nome_civil_ato.decode("utf-8") if ato.nome_civil_ato else None
),
"nome_do_serventuario_que_praticou_ato": (
ato.nome_serventuario_praticou_ato.decode("utf-8")
if ato.nome_serventuario_praticou_ato
else None
),
"data_hora_da_solicitacao": (
ato.data_solicitacao.isoformat() if ato.data_solicitacao else None
),
"ip_da_maquina_que_praticou_ato": ato.ip_maquina,
"valor_de_entrada_do_ato": ato.valor_entrada,
"emolumento_do_ato": ato.emolumento,
"taxa_judiciaria_do_ato": ato.taxa_judiciaria,
"fundos_estaduais_do_ato": ato.fundos_estaduais,
"protocolo_do_protesto": ato.protocolo_protesto,
"protocolo_do_imovel": ato.protocolo_imovel,
"documentos": documentos_list,
"partes": partes_list,
"atos_vinculados": atos_vinculados_list,
}
except HTTPException:
raise
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Erro ao buscar ato principal com vínculos: {str(e)}",
)
finally:
db.close()

38
packages/v1/administrativo/schemas/ato_principal_schema.py
View file

@ -89,6 +89,44 @@ class AtoPrincipalIdSchema(BaseModel):
return v
# ----------------------------------------------------
# Schema para Requisição de Código: Usado em SHOW e DELETE
# ----------------------------------------------------
class AtoPrincipalCodigoAtoSchema(BaseModel):
codigo_ato: str # Campo string equivalente ao varchar(50)
@field_validator("codigo_ato")
def validate_codigo_ato(cls, v: str):
# Remove espaços extras
v = v.strip()
# Verifica se está vazio
if not v:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=[
{
"input": "codigo_ato",
"message": "O campo 'código do ato' é obrigatório.",
}
],
)
# Verifica o tamanho máximo (equivalente a varchar(50))
if len(v) > 50:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=[
{
"input": "codigo_ato",
"message": "O campo 'código do ato' deve ter no máximo 50 caracteres.",
}
],
)
return v
# ----------------------------------------------------
# Schema para Criação (SAVE): Campos obrigatórios e sem ID
# ----------------------------------------------------

39
packages/v1/administrativo/services/ato_principal/ato_principal_show_atos_service.py Normal file
View file

@ -0,0 +1,39 @@
from fastapi import HTTPException, status
# Assumindo que o novo schema se chama AtoPrincipalSchema
from packages.v1.administrativo.schemas.ato_principal_schema import (
AtoPrincipalCodigoAtoSchema,
)
# Assumindo que a nova action se chama ShowAction
from packages.v1.administrativo.actions.ato_principal.ato_principal_show_atos_action import (
ShowAtosAction,
)
# Mantendo o padrão de nome de classe
class ShowAtosService:
"""
Serviço responsável por buscar os dados de um Ato Principal
utilizando o seu ID como parâmetro de busca.
"""
# O parâmetro do execute deve usar o novo Schema
def execute(self, ato_principal_schema: AtoPrincipalCodigoAtoSchema):
# Instânciamento da Action com o novo nome
show_action = ShowAtosAction()
# Executa a ação em questão (busca pelo ID)
# O action.execute receberá o schema com o ID a ser buscado
data = show_action.execute(ato_principal_schema)
if not data:
# Retorna uma exceção 404 se não encontrar
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Não foi possível localizar o Ato e suas dependências.",
)
# Retorno da informação (dados do Ato Principal)
return data

6
packages/v1/api.py
View file

@ -6,6 +6,7 @@ from packages.v1.administrativo.endpoints import ato_principal_endpoint
from packages.v1.administrativo.endpoints import usuario_endpoint
from packages.v1.administrativo.endpoints import ato_documento_endpoint
from packages.v1.administrativo.endpoints import ato_parte_endpoint
from packages.v1.administrativo.endpoints import atos_endpoint
# Cria uma instância do APIRouter que vai agregar todas as rotas da API
api_router = APIRouter()
@ -35,3 +36,8 @@ api_router.include_router(
api_router.include_router(
ato_principal_endpoint.router, prefix="/ato", tags=["Dados do Ato"]
)
# Inclui as rotas de ato_principal
api_router.include_router(
atos_endpoint.router, prefix="/atos", tags=["Retorna o ato e suas dependências"]
)