Esta seção explica como a regra é aplicada no pipeline, por que cada etapa existe e o que observar ao operar em produção.
O fluxo foi desenhado para ser seguro, idempotente, escalável (chunks) e auditável, seguindo o padrão GSERV.
A função principal pipeline_atualizar_usuarios() organiza o workflow em quatro etapas:
inspecionar_tabelas)quantificar_antes)aplicar_regra)profiles_id).quantificar_depois)Fluxo lógico:
USER_DN_LIKE: filtro de DN dos usuários-alvo (ex.: «%OU=SEAS%»).
PROFILE_ID_SEAS: ID do perfil padrão a ser garantido (ex.: 147).
Tabelas envolvidas: glpi_users, glpi_profiles, glpi_profiles_users.
Conexão: função get_mysql_connection() (padronizada via Connection do Airflow).
Boa prática: parametrizar USER_DN_LIKE e PROFILE_ID_SEAS via Airflow Variables ou dag.params para reuso por órgão.
O que faz:
Obtém o schema ativo (SELECT DATABASE()).
Lê metadados do information_schema. COLUMNS para as três tabelas-alvo.
Retorna um objeto com:
Schema: colunas, tipos, nulabilidade, chaves, defaults.
Observacoes: notas úteis (ex.: nome do schema corrente).
Por que existe:
Evitar que a regra rode em estruturas inesperadas (coluna ausente, tipo divergente).
Fornecer um snapshot para auditoria/debug.
O que observar:
Chama _collect_metrics() e retorna um dicionário de métricas (via XCom).
Serve como linha de base para comparar efeitos da regra.
Métricas padrão (exemplo GSERV):
total_users
pc_users_like (ajustado pelo seu USER_DN_LIKE)
users_profiles_id
links_profiles_users
sample_pc_users (amostra de 10 linhas)
Use essas métricas depois para diferenças (delta) e relatórios.
Exemplo apenas a fim de documentação. Resumo da regra (SEAS)
Valida existência do perfil-alvo em glpi_profiles (PROFILE_ID_SEAS).
Seleciona usuários pelo USER_DN_LIKE.
Insere vínculos em glpi_profiles_users para quem ainda não tem (users_id, profiles_id=PROFILE_ID_SEAS, is_dynamic=1, is_recursive=0).
Atualiza glpi_users.profiles_id para PROFILE_ID_SEAS em quem precisa.
Tudo isso:
Em chunks de 1000 IDs (função chunked) para escalar sem sobrecarregar o banco.
Dentro de transação com commit() ao final e rollback() em erros.
Estrutura de retorno
O dicionário resumo documenta o que foi feito:
profile_exists: boolean (perfil existe?)
seas_candidates: total de usuários candidatos pelo DN
to_link_ids: lista de usuários que exigiam vínculo
inserted_links: total de vínculos inseridos
to_update_ids: lista de usuários que exigiam UPDATE
updated_users: total de updates em glpi_users
queries: log simplificado de operações executadas
Por que chunks?
Evita locks longos e excesso de placeholders.
Limita o impacto de reversões (rollback) e torna a operação mais previsível.
Facilita retentativas em caso de falhas parciais.
Idempotência prática
A seleção de quem precisa de vínculo/UPDATE é recalculada a cada execução.
Usuários já corretos não sofrem alterações → repetir a DAG não “dobra” vínculos/perfis.
Tratamento de erro
Se o perfil não existir → abort com mensagem clara.
Qualquer exceção → rollback() + re-lançar erro (para alertar e impedir estado inconsistente).
Executa _collect_metrics() novamente para capturar o estado final.
Com ANTES vs. DEPOIS:
Calcular deltas (quantos usuários passaram a ter o perfil correto; quantos vínculos foram criados).
Detectar inconsistências entre glpi_users e glpi_profiles_users.
Gerar relatórios/CSVs e HTML de evidência (IDs afetados, amostras, métricas).
Criando uma task para consolidar resultados:
from airflow.decorators import task
import json
@task
def comparar_e_gerar_relatorio(m_before: dict, m_after: dict, resumo: dict):
diff = {
"delta_users_profiles":
m_after["users_profiles_id_140"] - m_before["users_profiles_id_140"],
"delta_links_profiles":
m_after["links_profiles_users_140"] - m_before["links_profiles_users_140"],
"inserted_links": resumo["inserted_links"],
"updated_users": resumo["updated_users"],
"candidatos": resumo["seas_candidates"]
}
print("[ANTES]", json.dumps(m_before, ensure_ascii=False, indent=2))
print("[DEPOIS]", json.dumps(m_after, ensure_ascii=False, indent=2))
print("[RESUMO]", json.dumps(resumo, ensure_ascii=False, indent=2))
print("[DIFERENÇAS]", json.dumps(diff, ensure_ascii=False, indent=2))
# TODO: salvar CSV/HTML em /opt/airflow/output/<dag_run_ts>/...
return diff
E encadeie no DAG:
antes = quantificar_antes()
resumo = aplicar_regra(antes)
depois = quantificar_depois()
comparar_e_gerar_relatorio(antes, depois, resumo)
Índices recomendados:
glpi_users(profiles_id), glpi_profiles_users(profiles_id)
Avaliar índice para user_dn conforme padrão LIKE (prefixo fixo ajuda).
Janela de execução:
Permissões mínimas:
Observabilidade:
✅ PROFILE_ID_SEAS existe em glpi_profiles.
✅ USER_DN_LIKE reflete corretamente o órgão/OU.
✅ Conexão do Airflow válida e com permissão adequada.
✅ Espaço/permite escrita em /opt/airflow/output (se gerar relatórios).
✅ Índices avaliados para consultas predominantes.
✅ Métricas ANTES e DEPOIS armazenadas (XCom/artefatos).
Rastreável: números e amostras que provam o efeito.
Reutilizável: trocar DN/perfil e replicar a outros órgãos.
Confiável: transações + chunks + idempotência.
Decisório: fornece insumos para KPIs, auditoria e priorização de novas DAGs.
Esse padrão sustenta a evolução do nosso catálogo de DAGs, reduz riscos e acelera operações consistentes entre sistemas institucionais.