76 lines
2.4 KiB
Python
76 lines
2.4 KiB
Python
"""
|
|
CRUD de CLIENTE_CONTATO — relacionado a CLIENTE por CLIENTE_ID (FK real,
|
|
ON DELETE RESTRICT ON UPDATE CASCADE). Substitui CLIENTE_CONTATO_CHAVE
|
|
(superseded — ver licenciamento-n8n/docs/CANDIDATO_CLIENTE_CONTATO.md).
|
|
"""
|
|
from .connection import get_connection
|
|
|
|
|
|
def listar_contatos_por_cliente(cliente_id: int) -> list[dict]:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT ID, NOME, TELEFONE, EMAIL FROM CLIENTE_CONTATO "
|
|
"WHERE CLIENTE_ID = %s ORDER BY ID",
|
|
(cliente_id,),
|
|
)
|
|
return list(cur.fetchall())
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def contato_duplicado(cliente_id: int, telefone: str, ignorar_id: int | None = None) -> bool:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
sql = "SELECT ID FROM CLIENTE_CONTATO WHERE CLIENTE_ID = %s AND TELEFONE = %s"
|
|
params = [cliente_id, telefone]
|
|
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 inserir_contato(cliente_id: int, nome: str | None, telefone: str, email: str | None = None) -> int:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"INSERT INTO CLIENTE_CONTATO (CLIENTE_ID, NOME, TELEFONE, EMAIL) VALUES (%s, %s, %s, %s)",
|
|
(cliente_id, nome, telefone, email),
|
|
)
|
|
return cur.lastrowid
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def atualizar_contato(
|
|
contato_id: int, cliente_id: int, nome: str | None, telefone: str, email: str | None = None
|
|
) -> None:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"UPDATE CLIENTE_CONTATO SET NOME = %s, TELEFONE = %s, EMAIL = %s "
|
|
"WHERE ID = %s AND CLIENTE_ID = %s",
|
|
(nome, telefone, email, contato_id, cliente_id),
|
|
)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def excluir_contato(contato_id: int, cliente_id: int) -> None:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"DELETE FROM CLIENTE_CONTATO WHERE ID = %s AND CLIENTE_ID = %s",
|
|
(contato_id, cliente_id),
|
|
)
|
|
finally:
|
|
conn.close()
|