Gustavo
bb8fd596b1
CSRF: novo app/csrf.py + middleware em app/main.py garantem um cookie csrf_token por navegador; todo formulario que muda estado (criar/editar/ excluir cliente, adicionar/editar/excluir contato) embute esse valor como campo oculto, e Depends(exigir_csrf) confere no POST que o valor enviado bate com o cookie antes de deixar a rota executar. Verificado ponta a ponta com o app real rodando (curl): token errado -> 403, token correto -> 303. Paginacao: db_clientes.listar_clientes ganha offset, nova contar_clientes(); GET /clientes aceita ?pagina=N (100 clientes por pagina), template mostra Anterior/Proxima e o total. 53/53 testes passando (8 novos).
125 lines
4.2 KiB
Python
125 lines
4.2 KiB
Python
"""
|
|
CRUD completo de CLIENTE. Única tabela cadastral (ID autoincrement é a PK
|
|
técnica; CODIGO é o identificador de negócio usado pelo resto do sistema —
|
|
ver licenciamento-n8n/docs/MAPEAMENTO_BANCO.md).
|
|
"""
|
|
import pymysql
|
|
|
|
from .connection import get_connection
|
|
|
|
CAMPOS_CLIENTE = [
|
|
"CODIGO", "CNPJ", "RAZAO", "FANTASIA", "IE", "ENDERECO", "CIDADE",
|
|
"BAIRRO", "UF", "CONTATO", "TELEFONE", "CELULAR", "CEP", "IMPLANTACAO",
|
|
"MENSALIDADE", "JUROS_MORA", "JUROS_MULTA", "DATA_CONTRATO", "EMAIL",
|
|
"OBS", "VENCIMENTO_CONTRATO", "DIA_VENCIMENTO",
|
|
]
|
|
|
|
_COLUNAS_SELECT = ", ".join(["ID", *CAMPOS_CLIENTE])
|
|
|
|
|
|
class ClienteComDependenciasError(Exception):
|
|
"""Levantada ao tentar excluir um CLIENTE que ainda tem CLIENTE_CONTATO."""
|
|
|
|
|
|
def listar_clientes(busca: str | None = None, limite: int = 100, offset: int = 0) -> list[dict]:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
if busca:
|
|
sql = (
|
|
f"SELECT {_COLUNAS_SELECT} FROM CLIENTE "
|
|
"WHERE CODIGO LIKE %s OR CNPJ LIKE %s OR RAZAO LIKE %s OR FANTASIA LIKE %s "
|
|
"ORDER BY RAZAO LIMIT %s OFFSET %s"
|
|
)
|
|
termo = f"%{busca}%"
|
|
cur.execute(sql, (termo, termo, termo, termo, limite, offset))
|
|
else:
|
|
cur.execute(
|
|
f"SELECT {_COLUNAS_SELECT} FROM CLIENTE ORDER BY RAZAO LIMIT %s OFFSET %s", (limite, offset)
|
|
)
|
|
return cur.fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def contar_clientes(busca: str | None = None) -> int:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
if busca:
|
|
sql = (
|
|
"SELECT COUNT(*) AS total FROM CLIENTE "
|
|
"WHERE CODIGO LIKE %s OR CNPJ LIKE %s OR RAZAO LIKE %s OR FANTASIA LIKE %s"
|
|
)
|
|
termo = f"%{busca}%"
|
|
cur.execute(sql, (termo, termo, termo, termo))
|
|
else:
|
|
cur.execute("SELECT COUNT(*) AS total FROM CLIENTE")
|
|
return cur.fetchone()["total"]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def obter_cliente(cliente_id: int) -> dict | None:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(f"SELECT {_COLUNAS_SELECT} FROM CLIENTE WHERE ID = %s", (cliente_id,))
|
|
return cur.fetchone()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def codigo_em_uso(codigo: str, ignorar_id: int | None = None) -> bool:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
sql = "SELECT ID FROM CLIENTE WHERE CODIGO = %s"
|
|
params = [codigo]
|
|
if ignorar_id is not None:
|
|
sql += " AND ID != %s"
|
|
params.append(ignorar_id)
|
|
cur.execute(sql, params)
|
|
return cur.fetchone() is not None
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def criar_cliente(dados: dict) -> int:
|
|
colunas = ", ".join(CAMPOS_CLIENTE)
|
|
marcadores = ", ".join(["%s"] * len(CAMPOS_CLIENTE))
|
|
valores = [dados[campo] for campo in CAMPOS_CLIENTE]
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(f"INSERT INTO CLIENTE ({colunas}) VALUES ({marcadores})", valores)
|
|
return cur.lastrowid
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def atualizar_cliente(cliente_id: int, dados: dict) -> None:
|
|
atribuicoes = ", ".join(f"{campo} = %s" for campo in CAMPOS_CLIENTE)
|
|
valores = [dados[campo] for campo in CAMPOS_CLIENTE]
|
|
valores.append(cliente_id)
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(f"UPDATE CLIENTE SET {atribuicoes} WHERE ID = %s", valores)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def excluir_cliente(cliente_id: int) -> None:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
try:
|
|
cur.execute("DELETE FROM CLIENTE WHERE ID = %s", (cliente_id,))
|
|
except pymysql.err.IntegrityError as exc:
|
|
raise ClienteComDependenciasError(
|
|
"Este cliente tem contatos cadastrados (CLIENTE_CONTATO) — remova-os antes de excluir."
|
|
) from exc
|
|
finally:
|
|
conn.close()
|