feat: coluna ATIVO em CLIENTE e filtro na camada de banco
This commit is contained in:
+47
-20
@@ -2,6 +2,21 @@
|
||||
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).
|
||||
|
||||
CLIENTE.ATIVO (TINYINT(1) NOT NULL DEFAULT 1) controla se o cliente
|
||||
aparece no filtro padrão da listagem ("Ativos"). Se CLIENTE já existir de
|
||||
uma instalação anterior sem essa coluna, rode:
|
||||
|
||||
ALTER TABLE CLIENTE
|
||||
ADD COLUMN ATIVO TINYINT(1) NOT NULL DEFAULT 1
|
||||
AFTER DIA_VENCIMENTO;
|
||||
|
||||
DEFAULT 1 é intencional: todo cliente cadastrado antes da migração nasce
|
||||
como Ativo — decisão do projeto (quem administra desativa manualmente os
|
||||
que sabe que não são mais, ver
|
||||
docs/superpowers/specs/2026-08-20-status-ativo-clientes-design.md).
|
||||
Cadastros/edições novos sempre informam ATIVO explicitamente (vindo do
|
||||
formulário), nunca dependem desse default.
|
||||
"""
|
||||
import pymysql
|
||||
|
||||
@@ -11,7 +26,7 @@ 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",
|
||||
"OBS", "VENCIMENTO_CONTRATO", "DIA_VENCIMENTO", "ATIVO",
|
||||
]
|
||||
|
||||
_COLUNAS_SELECT = ", ".join(["ID", *CAMPOS_CLIENTE])
|
||||
@@ -42,26 +57,31 @@ def listar_clientes(
|
||||
offset: int = 0,
|
||||
ordenar_por: str = ORDENACAO_PADRAO,
|
||||
direcao: str = DIRECAO_PADRAO,
|
||||
filtro_ativo: int | None = 1,
|
||||
) -> list[dict]:
|
||||
coluna = ORDENACOES_VALIDAS.get(ordenar_por, ORDENACOES_VALIDAS[ORDENACAO_PADRAO])
|
||||
direcao_sql = DIRECOES_VALIDAS.get(direcao, DIRECOES_VALIDAS[DIRECAO_PADRAO])
|
||||
# A busca é restrita à mesma coluna escolhida em "ordenar_por" — selecionar
|
||||
# "Código" busca só em CODIGO, não em todos os campos de uma vez.
|
||||
# filtro_ativo: 1 = só ativos, 0 = só inativos, None = sem filtro (todos).
|
||||
condicoes = []
|
||||
params: list = []
|
||||
if busca:
|
||||
condicoes.append(f"{coluna} LIKE %s")
|
||||
params.append(f"%{busca}%")
|
||||
if filtro_ativo is not None:
|
||||
condicoes.append("ATIVO = %s")
|
||||
params.append(filtro_ativo)
|
||||
where_sql = f"WHERE {' AND '.join(condicoes)} " if condicoes else ""
|
||||
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
if busca:
|
||||
# A busca é restrita à mesma coluna escolhida em "ordenar_por"
|
||||
# — selecionar "Código" busca só em CODIGO, não em todos os
|
||||
# campos de uma vez.
|
||||
sql = (
|
||||
f"SELECT {_COLUNAS_SELECT} FROM CLIENTE "
|
||||
f"WHERE {coluna} LIKE %s ORDER BY {coluna} {direcao_sql} LIMIT %s OFFSET %s"
|
||||
)
|
||||
cur.execute(sql, (f"%{busca}%", limite, offset))
|
||||
else:
|
||||
cur.execute(
|
||||
f"SELECT {_COLUNAS_SELECT} FROM CLIENTE ORDER BY {coluna} {direcao_sql} LIMIT %s OFFSET %s",
|
||||
(limite, offset),
|
||||
)
|
||||
sql = (
|
||||
f"SELECT {_COLUNAS_SELECT} FROM CLIENTE {where_sql}"
|
||||
f"ORDER BY {coluna} {direcao_sql} LIMIT %s OFFSET %s"
|
||||
)
|
||||
cur.execute(sql, [*params, limite, offset])
|
||||
return cur.fetchall()
|
||||
finally:
|
||||
conn.close()
|
||||
@@ -91,15 +111,22 @@ def proximo_codigo_sugerido() -> str:
|
||||
return str(proximo).zfill(6)
|
||||
|
||||
|
||||
def contar_clientes(busca: str | None = None, ordenar_por: str = ORDENACAO_PADRAO) -> int:
|
||||
def contar_clientes(busca: str | None = None, ordenar_por: str = ORDENACAO_PADRAO, filtro_ativo: int | None = 1) -> int:
|
||||
coluna = ORDENACOES_VALIDAS.get(ordenar_por, ORDENACOES_VALIDAS[ORDENACAO_PADRAO])
|
||||
condicoes = []
|
||||
params: list = []
|
||||
if busca:
|
||||
condicoes.append(f"{coluna} LIKE %s")
|
||||
params.append(f"%{busca}%")
|
||||
if filtro_ativo is not None:
|
||||
condicoes.append("ATIVO = %s")
|
||||
params.append(filtro_ativo)
|
||||
where_sql = f"WHERE {' AND '.join(condicoes)}" if condicoes else ""
|
||||
|
||||
conn = get_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
if busca:
|
||||
cur.execute(f"SELECT COUNT(*) AS total FROM CLIENTE WHERE {coluna} LIKE %s", (f"%{busca}%",))
|
||||
else:
|
||||
cur.execute("SELECT COUNT(*) AS total FROM CLIENTE")
|
||||
cur.execute(f"SELECT COUNT(*) AS total FROM CLIENTE {where_sql}", params)
|
||||
return cur.fetchone()["total"]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
Reference in New Issue
Block a user