[MVPTN] feat(CRUD): Finalização do CRUD De pessoas Fisicas e Jurídicas
This commit is contained in:
parent
355dda6f53
commit
765d36a5f6
7 changed files with 186 additions and 186 deletions
31
actions/data/prepare_update_data.py
Normal file
31
actions/data/prepare_update_data.py
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
from typing import Tuple, Dict, Any
|
||||
from pydantic import BaseModel
|
||||
|
||||
def prepare_update_data(schema: BaseModel, exclude_fields: list[str] = None, id_field: str = "id") -> Tuple[Dict[str, Any], str]:
|
||||
"""
|
||||
Gera dinamicamente os dados e SQL para update com base em um schema Pydantic.
|
||||
|
||||
Args:
|
||||
schema: Instância do schema (ex: t_pessoa_save_schema)
|
||||
exclude_fields: Lista de campos a serem ignorados no SET (ex: ['pessoa_id'])
|
||||
id_field: Nome do campo identificador primário (ex: 'pessoa_id')
|
||||
|
||||
Returns:
|
||||
Tuple contendo:
|
||||
- data_dict: dicionário com apenas campos preenchidos (sem unset)
|
||||
- update_sql: string SQL do tipo "campo1 = :campo1, campo2 = :campo2"
|
||||
"""
|
||||
exclude_fields = exclude_fields or [id_field]
|
||||
|
||||
# Cria o dicionário apenas com os campos enviados
|
||||
data_dict = schema.model_dump(exclude_unset=True)
|
||||
|
||||
# Monta lista dinâmica de campos para o SET
|
||||
update_fields = [
|
||||
f"{k} = :{k}"
|
||||
for k in data_dict.keys()
|
||||
if k not in exclude_fields
|
||||
]
|
||||
|
||||
update_sql = ", ".join(update_fields)
|
||||
return data_dict, update_sql
|
||||
|
|
@ -16,6 +16,7 @@ class TPessoaIndexRepository(BaseRepository):
|
|||
"""
|
||||
# Montagem do SQL
|
||||
sql = """ SELECT
|
||||
FIRST 200
|
||||
TP.PESSOA_ID,
|
||||
TP.PESSOA_TIPO,
|
||||
TP.NOME,
|
||||
|
|
@ -33,8 +34,8 @@ class TPessoaIndexRepository(BaseRepository):
|
|||
TP.UF,
|
||||
TP.CEP
|
||||
FROM T_PESSOA TP
|
||||
WHERE TP.PESSOA_TIPO = :pessoa_tipo """
|
||||
|
||||
WHERE TP.PESSOA_TIPO = :pessoa_tipo
|
||||
ORDER BY TP.PESSOA_ID DESC"""
|
||||
params = {
|
||||
"pessoa_tipo": t_pessoa_tipo_schema.pessoa_tipo
|
||||
}
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ class TPessoaSaveRepository(BaseRepository):
|
|||
|
||||
# Montagem do SQL
|
||||
sql = """
|
||||
INSERT INTO G_PESSOA(
|
||||
INSERT INTO T_PESSOA(
|
||||
PESSOA_ID,
|
||||
PESSOA_TIPO,
|
||||
NOME,
|
||||
|
|
|
|||
|
|
@ -1,77 +1,35 @@
|
|||
from abstracts.repository import BaseRepository
|
||||
from api.actions.data.prepare_update_data import prepare_update_data
|
||||
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaSaveSchema
|
||||
from fastapi import HTTPException, status
|
||||
from datetime import datetime
|
||||
|
||||
from api.packages.v1.administrativo.schemas.t_pessoa_schema import TPessoaUpdateSchema
|
||||
from abstracts.repository import BaseRepository
|
||||
|
||||
|
||||
class TPessoaUpdateRepository(BaseRepository):
|
||||
"""
|
||||
Repositório responsável por atualizar parcialmente registros na tabela T_PESSOA.
|
||||
"""
|
||||
|
||||
def execute(self, t_pessoa_update_schema: TPessoaUpdateSchema):
|
||||
"""
|
||||
Executa a atualização de um registro da tabela T_PESSOA.
|
||||
def execute(self, t_pessoa_save_schema: TPessoaSaveSchema):
|
||||
|
||||
Args:
|
||||
pessoa_id (int): ID da pessoa que será atualizada.
|
||||
t_pessoa_update_schema (TPessoaUpdateSchema): Dados enviados para atualização.
|
||||
|
||||
Returns:
|
||||
dict: Registro atualizado.
|
||||
|
||||
Raises:
|
||||
HTTPException: Caso o registro não exista ou ocorra erro de execução.
|
||||
"""
|
||||
try:
|
||||
# Extrai os campos do schema
|
||||
data = t_pessoa_update_schema.model_dump(exclude_unset=True)
|
||||
updates = []
|
||||
params = {"pessoa_id": t_pessoa_update_schema.pessoa_id}
|
||||
|
||||
# Monta dinamicamente os campos que serão atualizados
|
||||
for field, value in data.items():
|
||||
if value is not None:
|
||||
updates.append(f"{field.upper()} = :{field}")
|
||||
params[field] = value
|
||||
params, update_columns = prepare_update_data(
|
||||
t_pessoa_save_schema,
|
||||
exclude_fields=["pessoa_id"],
|
||||
id_field="pessoa_id"
|
||||
)
|
||||
|
||||
# Caso nenhum campo tenha sido informado
|
||||
if not updates:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Nenhum campo informado para atualização."
|
||||
)
|
||||
|
||||
# Atualiza a data de alteração automaticamente
|
||||
updates.append("DATA_AUTERACAO = :data_auteracao")
|
||||
params["data_auteracao"] = datetime.now()
|
||||
|
||||
# Montagem do SQL dinâmico
|
||||
sql = f"""
|
||||
UPDATE T_PESSOA
|
||||
SET {', '.join(updates)}
|
||||
WHERE PESSOA_ID = :pessoa_id
|
||||
RETURNING *;
|
||||
"""
|
||||
UPDATE T_PESSOA
|
||||
SET {update_columns}
|
||||
WHERE PESSOA_ID = :pessoa_id
|
||||
RETURNING pessoa_id
|
||||
"""
|
||||
|
||||
# Executa o comando e retorna o registro atualizado
|
||||
result = self.run_and_return(sql, params)
|
||||
# Executa o update
|
||||
response = self.run_and_return(sql, params)
|
||||
|
||||
if not result:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Nenhuma pessoa encontrada com o ID {t_pessoa_update_schema.pessoa_id}."
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
except HTTPException:
|
||||
raise # Repassa exceções controladas
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
# Captura falhas inesperadas
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail=f"Erro ao atualizar a pessoa: {e}"
|
||||
detail=f"Erro ao atualizar registro: {e}"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,8 +1,5 @@
|
|||
from pydantic import BaseModel, field_validator, model_validator
|
||||
from fastapi import HTTPException, status
|
||||
from pydantic import BaseModel
|
||||
from typing import Optional
|
||||
from actions.validations.text import Text
|
||||
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Schema base
|
||||
|
|
@ -19,7 +16,6 @@ class TPessoaRepresentanteSchema(BaseModel):
|
|||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Schema para localizar (GET)
|
||||
# ----------------------------------------------------
|
||||
|
|
@ -32,51 +28,20 @@ class TPessoaRepresentanteIdSchema(BaseModel):
|
|||
class TPessoaRepresentantePessoaIdSchema(BaseModel):
|
||||
pessoa_id: int
|
||||
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Schema para criação (POST)
|
||||
# ----------------------------------------------------
|
||||
class TPessoaRepresentanteSaveSchema(BaseModel):
|
||||
representante_id: Optional[int] = None
|
||||
pessoa_representante_id: Optional[int] = None
|
||||
pessoa_id: Optional[int]
|
||||
pessoa_auxiliar_id: Optional[int]
|
||||
marcacao_tipo_id: Optional[int]
|
||||
ato_partetipo_id: Optional[int]
|
||||
assinatura_tipo: Optional[str]
|
||||
|
||||
# Sanitiza o campo assinatura_tipo (texto)
|
||||
@field_validator('assinatura_tipo')
|
||||
def sanitize_fields(cls, v):
|
||||
if v:
|
||||
return Text.sanitize_input(v)
|
||||
return v
|
||||
|
||||
# Valida campos obrigatórios
|
||||
@model_validator(mode='after')
|
||||
def validate_required_fields(self):
|
||||
errors = []
|
||||
|
||||
if not self.pessoa_id:
|
||||
errors.append({'input': 'pessoa_id', 'message': 'O campo pessoa_id é obrigatório.'})
|
||||
|
||||
if not self.marcacao_tipo_id:
|
||||
errors.append({'input': 'marcacao_tipo_id', 'message': 'O campo marcacao_tipo_id é obrigatório.'})
|
||||
|
||||
if not self.ato_partetipo_id:
|
||||
errors.append({'input': 'ato_partetipo_id', 'message': 'O campo ato_partetipo_id é obrigatório.'})
|
||||
|
||||
if not self.assinatura_tipo or len(self.assinatura_tipo.strip()) == 0:
|
||||
errors.append({'input': 'assinatura_tipo', 'message': 'O campo assinatura_tipo é obrigatório.'})
|
||||
|
||||
if errors:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail=errors
|
||||
)
|
||||
|
||||
return self
|
||||
pessoa_id: Optional[int] = None
|
||||
pessoa_auxiliar_id: Optional[int] = None
|
||||
marcacao_tipo_id: Optional[int] = None
|
||||
ato_partetipo_id: Optional[int] = None
|
||||
assinatura_tipo: Optional[str] = None
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Schema para atualização (PUT)
|
||||
|
|
@ -89,25 +54,5 @@ class TPessoaRepresentanteUpdateSchema(BaseModel):
|
|||
ato_partetipo_id: Optional[int] = None
|
||||
assinatura_tipo: Optional[str] = None
|
||||
|
||||
# Sanitiza
|
||||
@field_validator('assinatura_tipo')
|
||||
def sanitize_fields(cls, v):
|
||||
if v:
|
||||
return Text.sanitize_input(v)
|
||||
return v
|
||||
|
||||
# Valida se o campo texto não está vazio
|
||||
@model_validator(mode='after')
|
||||
def validate_fields(self):
|
||||
errors = []
|
||||
|
||||
if self.assinatura_tipo is not None and len(self.assinatura_tipo.strip()) == 0:
|
||||
errors.append({'input': 'assinatura_tipo', 'message': 'O campo assinatura_tipo não pode estar vazio.'})
|
||||
|
||||
if errors:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail=errors
|
||||
)
|
||||
|
||||
return self
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
|
@ -1,10 +1,7 @@
|
|||
from pydantic import BaseModel, field_validator, model_validator
|
||||
from fastapi import HTTPException, status
|
||||
from pydantic import BaseModel
|
||||
from typing import Optional
|
||||
from datetime import datetime
|
||||
|
||||
# Funções para sanitização de entradas
|
||||
from actions.validations.text import Text
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Schema base
|
||||
|
|
@ -60,7 +57,7 @@ class TPessoaSchema(BaseModel):
|
|||
inscricao_municipal: Optional[str] = None
|
||||
enviado_cncncnb: Optional[str] = None
|
||||
data_auteracao: Optional[datetime] = None
|
||||
data_envio_ccn: Optional[datetime] = None
|
||||
data_envioccn: Optional[datetime] = None
|
||||
ccnregistros_id: Optional[float] = None
|
||||
observacao_envioccn: Optional[str] = None
|
||||
observacao_envio_ccn: Optional[str] = None
|
||||
|
|
@ -106,62 +103,130 @@ class TPessoaTipoSchema(BaseModel):
|
|||
# Schema para criação de pessoa (POST)
|
||||
# ----------------------------------------------------
|
||||
class TPessoaSaveSchema(TPessoaSchema):
|
||||
|
||||
# Campos obrigatórios podem ser definidos aqui
|
||||
nome: str
|
||||
pessoa_tipo: str
|
||||
|
||||
@field_validator(
|
||||
'nome', 'pessoa_tipo', 'nacionalidade', 'documento', 'nome_pai', 'nome_mae',
|
||||
'endereco', 'cidade', 'uf', 'sexo', 'email', 'bairro', 'cep'
|
||||
)
|
||||
def sanitize_fields(cls, v):
|
||||
if v:
|
||||
return Text.sanitize_input(v)
|
||||
return v
|
||||
|
||||
@model_validator(mode='after')
|
||||
def validate_required_fields(self):
|
||||
errors = []
|
||||
|
||||
if not self.nome or len(self.nome.strip()) == 0:
|
||||
errors.append({'input': 'nome', 'message': 'O nome é obrigatório.'})
|
||||
if not self.pessoa_tipo or len(self.pessoa_tipo.strip()) == 0:
|
||||
errors.append({'input': 'pessoa_tipo', 'message': 'O tipo de pessoa é obrigatório.'})
|
||||
|
||||
if errors:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail=errors
|
||||
)
|
||||
return self
|
||||
pessoa_id: Optional[float] = None
|
||||
pessoa_tipo: Optional[str] = None
|
||||
nome: Optional[str] = None
|
||||
nacionalidade: Optional[str] = None
|
||||
documento: Optional[str] = None
|
||||
tb_documentotipo_id: Optional[float] = None
|
||||
tb_profissao_id: Optional[float] = None
|
||||
tb_estadocivil_id: Optional[float] = None
|
||||
nome_pai: Optional[str] = None
|
||||
nome_mae: Optional[str] = None
|
||||
data_cadastro: Optional[datetime] = None
|
||||
naturalidade: Optional[str] = None
|
||||
telefone: Optional[str] = None
|
||||
endereco: Optional[str] = None
|
||||
cidade: Optional[str] = None
|
||||
uf: Optional[str] = None
|
||||
data_nascimento: Optional[datetime] = None
|
||||
sexo: Optional[str] = None
|
||||
tb_regimecomunhao_id: Optional[float] = None
|
||||
pessoa_conjuge_id: Optional[float] = None
|
||||
email: Optional[str] = None
|
||||
documento_numero: Optional[str] = None
|
||||
bairro: Optional[str] = None
|
||||
cep: Optional[str] = None
|
||||
documento_expedicao: Optional[datetime] = None
|
||||
documento_validade: Optional[datetime] = None
|
||||
observacao: Optional[str] = None
|
||||
cpf_cnpj: Optional[str] = None
|
||||
cpf_terceiro: Optional[str] = None
|
||||
nome_fantasia: Optional[str] = None
|
||||
texto: Optional[bytes] = None
|
||||
ddd: Optional[str] = None
|
||||
cert_casamento_numero: Optional[float] = None
|
||||
cert_casamento_folha: Optional[str] = None
|
||||
cert_casamento_livro: Optional[str] = None
|
||||
cert_casamento_cartorio: Optional[str] = None
|
||||
cert_casamento_data: Optional[datetime] = None
|
||||
cert_casamento_lei: Optional[str] = None
|
||||
pessoa_conjuge_nome: Optional[str] = None
|
||||
estrangeiro_nat: Optional[str] = None
|
||||
estrangeiro_nat_tb_pais_id: Optional[float] = None
|
||||
estrangeiro_res_tb_pais_id: Optional[float] = None
|
||||
estrangeiro_res: Optional[str] = None
|
||||
municipio_id: Optional[float] = None
|
||||
documento_orgao: Optional[str] = None
|
||||
documento_uf: Optional[str] = None
|
||||
uf_residencia: Optional[str] = None
|
||||
inscricao_municipal: Optional[str] = None
|
||||
enviado_cncncnb: Optional[str] = None
|
||||
data_auteracao: Optional[datetime] = None
|
||||
data_envioccn: Optional[datetime] = None
|
||||
ccnregistros_id: Optional[float] = None
|
||||
observacao_envioccn: Optional[str] = None
|
||||
observacao_envio_ccn: Optional[str] = None
|
||||
deficiencias: Optional[str] = None
|
||||
grau_instrucao: Optional[float] = None
|
||||
cidade_nat_id: Optional[float] = None
|
||||
tb_tipologradouro_id: Optional[float] = None
|
||||
unidade: Optional[str] = None
|
||||
numero_end: Optional[float] = None
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Schema para atualização de pessoa (PUT)
|
||||
# ----------------------------------------------------
|
||||
class TPessoaUpdateSchema(TPessoaSchema):
|
||||
|
||||
@field_validator(
|
||||
'nome', 'pessoa_tipo', 'nacionalidade', 'documento', 'nome_pai', 'nome_mae',
|
||||
'endereco', 'cidade', 'uf', 'sexo', 'email', 'bairro', 'cep'
|
||||
)
|
||||
def sanitize_fields(cls, v):
|
||||
if v:
|
||||
return Text.sanitize_input(v)
|
||||
return v
|
||||
|
||||
@model_validator(mode='after')
|
||||
def validate_required_fields(self):
|
||||
# Em update, podemos permitir que campos sejam opcionais, mas se enviados, não podem ser vazios
|
||||
errors = []
|
||||
for field in ['nome', 'pessoa_tipo']:
|
||||
value = getattr(self, field)
|
||||
if value is not None and len(str(value).strip()) == 0:
|
||||
errors.append({'input': field, 'message': f'O campo {field} não pode estar vazio.'})
|
||||
|
||||
if errors:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail=errors
|
||||
)
|
||||
return self
|
||||
pessoa_id: Optional[float] = None
|
||||
pessoa_tipo: Optional[str] = None
|
||||
nome: Optional[str] = None
|
||||
nacionalidade: Optional[str] = None
|
||||
documento: Optional[str] = None
|
||||
tb_documentotipo_id: Optional[float] = None
|
||||
tb_profissao_id: Optional[float] = None
|
||||
tb_estadocivil_id: Optional[float] = None
|
||||
nome_pai: Optional[str] = None
|
||||
nome_mae: Optional[str] = None
|
||||
data_cadastro: Optional[datetime] = None
|
||||
naturalidade: Optional[str] = None
|
||||
telefone: Optional[str] = None
|
||||
endereco: Optional[str] = None
|
||||
cidade: Optional[str] = None
|
||||
uf: Optional[str] = None
|
||||
data_nascimento: Optional[datetime] = None
|
||||
sexo: Optional[str] = None
|
||||
tb_regimecomunhao_id: Optional[float] = None
|
||||
pessoa_conjuge_id: Optional[float] = None
|
||||
email: Optional[str] = None
|
||||
documento_numero: Optional[str] = None
|
||||
bairro: Optional[str] = None
|
||||
cep: Optional[str] = None
|
||||
documento_expedicao: Optional[datetime] = None
|
||||
documento_validade: Optional[datetime] = None
|
||||
observacao: Optional[str] = None
|
||||
cpf_cnpj: Optional[str] = None
|
||||
cpf_terceiro: Optional[str] = None
|
||||
nome_fantasia: Optional[str] = None
|
||||
texto: Optional[bytes] = None
|
||||
ddd: Optional[str] = None
|
||||
cert_casamento_numero: Optional[float] = None
|
||||
cert_casamento_folha: Optional[str] = None
|
||||
cert_casamento_livro: Optional[str] = None
|
||||
cert_casamento_cartorio: Optional[str] = None
|
||||
cert_casamento_data: Optional[datetime] = None
|
||||
cert_casamento_lei: Optional[str] = None
|
||||
pessoa_conjuge_nome: Optional[str] = None
|
||||
estrangeiro_nat: Optional[str] = None
|
||||
estrangeiro_nat_tb_pais_id: Optional[float] = None
|
||||
estrangeiro_res_tb_pais_id: Optional[float] = None
|
||||
estrangeiro_res: Optional[str] = None
|
||||
municipio_id: Optional[float] = None
|
||||
documento_orgao: Optional[str] = None
|
||||
documento_uf: Optional[str] = None
|
||||
uf_residencia: Optional[str] = None
|
||||
inscricao_municipal: Optional[str] = None
|
||||
enviado_cncncnb: Optional[str] = None
|
||||
data_auteracao: Optional[datetime] = None
|
||||
data_envioccn: Optional[datetime] = None
|
||||
ccnregistros_id: Optional[float] = None
|
||||
observacao_envioccn: Optional[str] = None
|
||||
observacao_envio_ccn: Optional[str] = None
|
||||
deficiencias: Optional[str] = None
|
||||
grau_instrucao: Optional[float] = None
|
||||
cidade_nat_id: Optional[float] = None
|
||||
tb_tipologradouro_id: Optional[float] = None
|
||||
unidade: Optional[str] = None
|
||||
numero_end: Optional[float] = None
|
||||
|
|
|
|||
|
|
@ -36,7 +36,7 @@ class TPessoaSaveService:
|
|||
sequencia = generate.execute(sequencia_schema)
|
||||
|
||||
# Atualiza os dados da chave primária
|
||||
regimebens_schema.tb_regimebens_id = sequencia.sequencia
|
||||
t_pessoa_save_schema.pessoa_id = sequencia.sequencia
|
||||
|
||||
# Instanciamento de ações
|
||||
t_pessoa_save_action = TPessoaSaveAction()
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue