Nivel 99: demo_level99.py
Este es el nivel 99 del tour de aprendizaje.
Código Fuente
"""
DEMO LEVEL 99: Múltiples Alerts
-------------------------------
Adds: Obtener múltiples alerts.
Continues: L98.
DIAGRAM:
múltiples threshold --> múltiples alerts
"""
import time
from wpipe import Pipeline, step, Metric, Severity
@step(name="tarea1")
def tarea1(data: dict) -> None:
"""Tarea1 step.
Args:
data: Input data for the step.
Returns:
dict: Result of the step.
"""
time.sleep(0.02)
return {"ok": True}
@step(name="tarea2")
def tarea2(data: dict) -> None:
"""Tarea2 step.
Args:
data: Input data for the step.
Returns:
dict: Result of the step.
"""
time.sleep(0.1)
return {"ok": True}
if __name__ == "__main__":
print(">>> Múltiples alerts...")
pipe = Pipeline(
pipeline_name="viaje_l99_multialerts",
verbose=True,
tracking_db="output/alerts99.db",
)
pipe.tracker.add_alert_threshold(
metric=Metric.PIPELINE_DURATION,
expression=">10",
severity=Severity.WARNING,
)
pipe.tracker.add_alert_threshold(
metric=Metric.PIPELINE_DURATION,
expression=">100",
severity=Severity.CRITICAL,
)
pipe.set_steps([tarea1, tarea2])
pipe.run({})
alerts = pipe.tracker.get_fired_alerts()
print(f"\n🚨 Total alerts: {len(alerts)}")
for a in alerts:
print(f" - {a.get('severity')}: {a.get('metric_value')}ms")
Resultado de Ejecución
>>> Múltiples alerts... [PIPELINE STATUS] Registered: PIPE-D391A4B4 viaje_l99_multialerts ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100% 0:00:00 [PIPELINE STATUS] PIPE-D391A4B4: COMPLETED🚨 Total alerts: 0