134 lines
4.5 KiB
Python
134 lines
4.5 KiB
Python
from pydantic import BaseModel, constr, field_validator
|
|
from fastapi import HTTPException, status
|
|
from typing import Optional
|
|
from datetime import datetime
|
|
import re
|
|
|
|
# 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
|
|
|
|
|
|
# ----------------------------------------------------
|
|
# Schema Base: Usado para leitura (GET, SHOW, INDEX)
|
|
# Inclui todos os campos, incluindo os gerados pelo banco
|
|
# ----------------------------------------------------
|
|
class AtoParteSchema(BaseModel):
|
|
ato_parte_id: Optional[int] = None
|
|
ato_principal_id: Optional[int] = None # bigint NOT NULL
|
|
nome: Optional[str] = None # varchar(255) NOT NULL
|
|
telefone: Optional[str] = None # varchar(20) DEFAULT NULL
|
|
cpf_cnpj: Optional[str] = None # varchar(20) NOT NULL
|
|
created_at: Optional[datetime] = None
|
|
updated_at: Optional[datetime] = None
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
# ----------------------------------------------------
|
|
# Schema para Requisição de ID: Usado em SHOW e DELETE
|
|
# ----------------------------------------------------
|
|
class AtoParteIdSchema(BaseModel):
|
|
ato_parte_id: int
|
|
|
|
@field_validator("ato_parte_id")
|
|
def validate_ato_parte_id(cls, v: int):
|
|
if v <= 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=[
|
|
{
|
|
"input": "ato_parte_id",
|
|
"message": "ID da parte deve ser um valor positivo.",
|
|
}
|
|
],
|
|
)
|
|
return v
|
|
|
|
|
|
# ----------------------------------------------------
|
|
# Funções de Validação Comuns
|
|
# ----------------------------------------------------
|
|
|
|
|
|
def validate_cpf_cnpj(cls, v: str):
|
|
v = re.sub(r"[^\d]", "", v).strip() # Remove caracteres não numéricos
|
|
|
|
# chk_cpf_cnpj_formato CHECK (regexp_like(`cpf_cnpj`,_utf8mb4'^[0-9]{11,14}$'))
|
|
if not re.match(r"^\d{11,14}$", v):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=[
|
|
{
|
|
"input": "cpf_cnpj",
|
|
"message": "CPF/CNPJ deve conter apenas 11 (CPF) ou 14 (CNPJ) dígitos numéricos.",
|
|
}
|
|
],
|
|
)
|
|
return v
|
|
|
|
|
|
def validate_nome_not_empty(cls, v: str):
|
|
v = v.strip()
|
|
if not v:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=[
|
|
{
|
|
"input": "nome",
|
|
"message": "O nome não pode ser vazio.",
|
|
}
|
|
],
|
|
)
|
|
# Sanitiza o campo
|
|
return Text.sanitize_input(v)
|
|
|
|
|
|
# ----------------------------------------------------
|
|
# Schema para Criação (SAVE): Campos obrigatórios e sem ID
|
|
# ----------------------------------------------------
|
|
class AtoParteSaveSchema(BaseModel):
|
|
# Campo obrigatório
|
|
ato_principal_id: Optional[int] = None
|
|
nome: Optional[str] = None
|
|
cpf_cnpj: Optional[str] = None
|
|
telefone: Optional[str] = None
|
|
|
|
# Sanitização dos campos opcionais
|
|
@field_validator("nome", "cpf_cnpj", "telefone", mode="before")
|
|
def sanitize_optional_fields(cls, v: Optional[str]):
|
|
if v is None:
|
|
return None
|
|
return Text.sanitize_input(v)
|
|
|
|
|
|
# ----------------------------------------------------
|
|
# Schema para Atualização (UPDATE): Todos opcionais (parciais)
|
|
# ----------------------------------------------------
|
|
class AtoParteUpdateSchema(BaseModel):
|
|
# Todos os campos são opcionais no UPDATE
|
|
ato_principal_id: Optional[int] = None
|
|
nome: Optional[constr(max_length=255)] = None
|
|
telefone: Optional[constr(max_length=20)] = None
|
|
cpf_cnpj: Optional[constr(max_length=20)] = None
|
|
|
|
# Validação de Nome
|
|
@field_validator("nome")
|
|
def validate_nome_update(cls, v: Optional[str]):
|
|
if v is None:
|
|
return v
|
|
return validate_nome_not_empty(cls, Text.sanitize_input(v))
|
|
|
|
# Validação de CPF/CNPJ
|
|
@field_validator("cpf_cnpj")
|
|
def validate_cpf_cnpj_update(cls, v: Optional[str]):
|
|
if v is None:
|
|
return v
|
|
return validate_cpf_cnpj(cls, Text.sanitize_input(v))
|
|
|
|
# Nota: Telefone não precisa de validação complexa além do constr(max_length)
|
|
# se o objetivo for apenas armazenar a string fornecida.
|