318 lines
12 KiB
Python
318 lines
12 KiB
Python
from pydantic import BaseModel, constr, field_validator
|
|
from fastapi import HTTPException, status
|
|
from typing import List
|
|
from typing import Optional
|
|
from datetime import datetime
|
|
from decimal import Decimal # Importar Decimal para campos monetários
|
|
|
|
# Funções para sanitização de entradas (evitar XSS, SQLi etc.)
|
|
from actions.validations.text import Text
|
|
|
|
# Funções para sanitização de entradas (evitar XSS, SQLi etc.)
|
|
# É importante que esta função seja mantida/implementada no seu ambiente
|
|
# from actions.validations.text import Text # Descomentar se for usar
|
|
|
|
# Schemas dos itens
|
|
from packages.v1.administrativo.schemas.ato_parte_schema import AtoParteSaveSchema
|
|
from packages.v1.administrativo.schemas.ato_documento_schema import (
|
|
AtoDocumentoSaveSchema,
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------
|
|
# Schema Base: Usado para leitura (GET, SHOW, INDEX)
|
|
# Inclui todos os campos, incluindo os gerados pelo banco
|
|
# ----------------------------------------------------
|
|
class AtoPrincipalSchema(BaseModel):
|
|
# DDL: bigint NOT NULL AUTO_INCREMENT
|
|
ato_principal_id: Optional[int] = None
|
|
# DDL: bigint DEFAULT NULL
|
|
origem_ato_principal_id: Optional[int] = None
|
|
# DDL: bigint NOT NULL
|
|
identificacao_pedido_cgj: int
|
|
# DDL: int NOT NULL
|
|
tipo_ato: int
|
|
# DDL: varchar(50) NOT NULL (UNIQUE)
|
|
codigo_selo: constr(max_length=50)
|
|
# DDL: varchar(50) NOT NULL (UNIQUE)
|
|
codigo_ato: constr(max_length=50)
|
|
# DDL: varchar(255) NOT NULL (mas o banco pode conter valores NULL em registros antigos)
|
|
nome_civil_ato: Optional[constr(max_length=255)] = None
|
|
# DDL: varchar(255) NOT NULL (mas o banco pode conter valores NULL em registros antigos)
|
|
nome_serventuario_praticou_ato: Optional[constr(max_length=255)] = None
|
|
# DDL: datetime NOT NULL
|
|
data_solicitacao: datetime
|
|
# DDL: varchar(45) DEFAULT NULL
|
|
ip_maquina: Optional[constr(max_length=45)] = None
|
|
# DDL: text NOT NULL
|
|
inteiro_teor: Optional[str] = None # permite valor nulo
|
|
# DDL: decimal(12,2) DEFAULT NULL
|
|
valor_entrada: Optional[Decimal] = None
|
|
# DDL: decimal(12,2) NOT NULL
|
|
emolumento: Decimal
|
|
# DDL: decimal(12,2) NOT NULL
|
|
taxa_judiciaria: Decimal
|
|
# DDL: decimal(12,2) NOT NULL
|
|
fundos_estaduais: Decimal
|
|
# DDL: varchar(50) DEFAULT NULL
|
|
protocolo_protesto: Optional[constr(max_length=50)] = None
|
|
# DDL: varchar(50) DEFAULT NULL
|
|
protocolo_imovel: Optional[constr(max_length=50)] = None
|
|
# DDL: datetime NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
created_at: Optional[datetime] = None
|
|
# DDL: datetime NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
updated_at: Optional[datetime] = None
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
# Permite a conversão de float/int para Decimal
|
|
json_encoders = {Decimal: lambda v: float(v)}
|
|
# Define a precisão para Decimal
|
|
decimal_places = 2
|
|
|
|
|
|
# ----------------------------------------------------
|
|
# Schema para Requisição de ID: Usado em SHOW e DELETE
|
|
# ----------------------------------------------------
|
|
class AtoPrincipalIdSchema(BaseModel):
|
|
ato_principal_id: int
|
|
|
|
@field_validator("ato_principal_id")
|
|
def validate_ato_principal_id(cls, v: int):
|
|
if v <= 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=[
|
|
{
|
|
"input": "ato_principal_id",
|
|
"message": "ID do Ato Principal deve ser um valor positivo.",
|
|
}
|
|
],
|
|
)
|
|
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 Text.sanitize_input(v)
|
|
|
|
|
|
# ----------------------------------------------------
|
|
# Schema para Criação (SAVE): Campos obrigatórios e sem ID
|
|
# ----------------------------------------------------
|
|
class AtoPrincipalSaveSchema(BaseModel):
|
|
ato_principal_id: Optional[int] = None # <<< tornar opcional
|
|
|
|
# Campos obrigatórios baseados na DDL (NOT NULL)
|
|
origem_ato_principal_id: Optional[int] = None # bigint DEFAULT NULL
|
|
identificacao_pedido_cgj: int # bigint NOT NULL
|
|
tipo_ato: int # int NOT NULL
|
|
codigo_selo: constr(max_length=50) # varchar(50) NOT NULL
|
|
codigo_ato: constr(max_length=50) # varchar(50) NOT NULL
|
|
nome_civil_ato: constr(max_length=255) # varchar(255) NOT NULL
|
|
nome_serventuario_praticou_ato: constr(max_length=255) # varchar(255) NOT NULL
|
|
data_solicitacao: datetime # datetime NOT NULL
|
|
ip_maquina: Optional[constr(max_length=45)] = None # varchar(45) DEFAULT NULL
|
|
inteiro_teor: str # text NOT NULL
|
|
valor_entrada: Optional[Decimal] = None # decimal(12,2) DEFAULT NULL
|
|
emolumento: Decimal # decimal(12,2) NOT NULL
|
|
taxa_judiciaria: Decimal # decimal(12,2) NOT NULL
|
|
fundos_estaduais: Decimal # decimal(12,2) NOT NULL
|
|
protocolo_protesto: Optional[constr(max_length=50)] = (
|
|
None # varchar(50) DEFAULT NULL
|
|
)
|
|
protocolo_imovel: Optional[constr(max_length=50)] = None # varchar(50) DEFAULT NULL
|
|
ato_partes: List[AtoParteSaveSchema] = [] # precisa existir
|
|
ato_documentos: List[AtoDocumentoSaveSchema] = [] # precisa existir
|
|
|
|
# NOVO CAMPO: Referência recursiva a outros atos principais
|
|
# Usando string para forward reference
|
|
atos_vinculados: Optional[List["AtoPrincipalSaveSchema"]] = None
|
|
|
|
# Validação e Sanitização de campos de texto obrigatórios (adaptado do código fonte)
|
|
@field_validator(
|
|
"codigo_selo",
|
|
"codigo_ato",
|
|
"nome_civil_ato",
|
|
"nome_serventuario_praticou_ato",
|
|
"inteiro_teor",
|
|
)
|
|
def validate_required_strings(cls, v: str):
|
|
v = v.strip()
|
|
if not v:
|
|
# Identifica o campo que falhou a validação
|
|
input_name = (
|
|
"Erro de campo" # Será substituído se a função for mais complexa
|
|
)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=[
|
|
{
|
|
"input": input_name,
|
|
"message": "Este campo obrigatório não pode ser vazio.",
|
|
}
|
|
],
|
|
)
|
|
# Adicionar aqui a sanitização de texto (ex: Text.sanitize(v)) se disponível
|
|
return Text.sanitize_input(v)
|
|
|
|
# Validação dos campos monetários (baseado na CHECK CONSTRAINT da DDL)
|
|
@field_validator(
|
|
"valor_entrada", "emolumento", "taxa_judiciaria", "fundos_estaduais"
|
|
)
|
|
def validate_positive_values(cls, v: Optional[Decimal]):
|
|
if v is None and (
|
|
"valor_entrada" in cls.__annotations__
|
|
and cls.__annotations__["valor_entrada"] == Optional[Decimal]
|
|
):
|
|
# valor_entrada é o único campo DEFAULT NULL, aceita None
|
|
return v
|
|
if v is not None and v < 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=[
|
|
{
|
|
"input": "campo_monetario",
|
|
"message": "Os valores monetários não podem ser negativos.",
|
|
}
|
|
],
|
|
)
|
|
return v
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
json_encoders = {Decimal: lambda v: float(v)}
|
|
decimal_places = 2
|
|
|
|
|
|
# ----------------------------------------------------
|
|
# Schema para Atualização (UPDATE): Todos opcionais (parciais)
|
|
# ----------------------------------------------------
|
|
class AtoPrincipalUpdateSchema(BaseModel):
|
|
# Todos os campos que podem ser atualizados são opcionais
|
|
origem_ato_principal_id: Optional[int] = None
|
|
identificacao_pedido_cgj: Optional[int] = None
|
|
tipo_ato: Optional[int] = None
|
|
codigo_selo: Optional[constr(max_length=50)] = None
|
|
codigo_ato: Optional[constr(max_length=50)] = None
|
|
nome_civil_ato: Optional[constr(max_length=255)] = None
|
|
nome_serventuario_praticou_ato: Optional[constr(max_length=255)] = None
|
|
data_solicitacao: Optional[datetime] = None
|
|
ip_maquina: Optional[constr(max_length=45)] = None
|
|
inteiro_teor: Optional[str] = None
|
|
valor_entrada: Optional[Decimal] = None
|
|
emolumento: Optional[Decimal] = None
|
|
taxa_judiciaria: Optional[Decimal] = None
|
|
fundos_estaduais: Optional[Decimal] = None
|
|
protocolo_protesto: Optional[constr(max_length=50)] = None
|
|
protocolo_imovel: Optional[constr(max_length=50)] = None
|
|
|
|
# Reutiliza a validação de strings obrigatórias (só valida se o campo for fornecido)
|
|
@field_validator(
|
|
"codigo_selo",
|
|
"codigo_ato",
|
|
"nome_civil_ato",
|
|
"nome_serventuario_praticou_ato",
|
|
"inteiro_teor",
|
|
)
|
|
def validate_non_empty_strings_on_update(cls, v: Optional[str]):
|
|
if v is not None:
|
|
v_stripped = v.strip()
|
|
if not v_stripped:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=[
|
|
{
|
|
"input": "campo_de_texto",
|
|
"message": "Este campo não pode ser atualizado para um valor vazio.",
|
|
}
|
|
],
|
|
)
|
|
return Text.sanitize_input(v)
|
|
|
|
# Reutiliza a validação de valores positivos
|
|
@field_validator(
|
|
"valor_entrada", "emolumento", "taxa_judiciaria", "fundos_estaduais"
|
|
)
|
|
def validate_positive_values_on_update(cls, v: Optional[Decimal]):
|
|
if v is not None and v < 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=[
|
|
{
|
|
"input": "campo_monetario",
|
|
"message": "Os valores monetários não podem ser negativos.",
|
|
}
|
|
],
|
|
)
|
|
return Text.sanitize_input(v)
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
json_encoders = {Decimal: lambda v: float(v)}
|
|
decimal_places = 2
|
|
|
|
|
|
# ----------------------------------------------------
|
|
# Schema auxiliar para buscar por campos únicos (opcional, mas útil para o Save Service)
|
|
# ----------------------------------------------------
|
|
class AtoPrincipalCodigoSchema(BaseModel):
|
|
codigo: constr(max_length=50) # Pode ser codigo_selo ou codigo_ato
|
|
|
|
@field_validator("codigo")
|
|
def validate_codigo(cls, v: str):
|
|
v = v.strip()
|
|
if not v:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=[
|
|
{
|
|
"input": "codigo",
|
|
"message": "O código de busca não pode ser vazio.",
|
|
}
|
|
],
|
|
)
|
|
return v
|
|
|
|
|
|
# --- ADIÇÃO CRÍTICA PARA RECURSIVIDADE (Resolução da forward reference) ---
|
|
# Necessário para resolver a referência da classe dentro de si mesma.
|
|
# Tenta Pydantic v2 (model_rebuild) e, se falhar, Pydantic v1 (update_forward_refs).
|
|
try:
|
|
AtoPrincipalSaveSchema.model_rebuild()
|
|
except AttributeError:
|
|
# Se estiver usando Pydantic v1
|
|
AtoPrincipalSaveSchema.update_forward_refs()
|