Nivel 102: demo_level102.pyο
Este es el nivel 102 del tour de aprendizaje.
CΓ³digo Fuenteο
"""
DEMO LEVEL 102: get_top_slow_steps
---------------------------------
Adds: Obtener pasos mΓ‘s lentos.
Continues: L101.
DIAGRAM:
analysis.get_top_slow_steps(limit=3)
"""
import time
from wpipe import Pipeline, step
@step(name="rapido")
def rapido(data: dict) -> None:
"""Rapido step.
Args:
data: Input data for the step.
Returns:
dict: Result of the step.
"""
return {"ok": True}
@step(name="lento")
def lento(data: dict) -> None:
"""Lento step.
Args:
data: Input data for the step.
Returns:
dict: Result of the step.
"""
time.sleep(0.05)
return {"ok": True}
if __name__ == "__main__":
print(">>> Pasos mΓ‘s lentos...")
for i in range(3):
pipe = Pipeline(
pipeline_name=f"Viaje_L102_{i}",
verbose=False,
tracking_db="output/slow102.db",
)
pipe.set_steps([rapido, lento])
pipe.run({})
slow = pipe.tracker.analysis.get_top_slow_steps(limit=3)
print(f"\nπ’ Pasos lentos: {len(slow)}")
for s in slow:
print(f" - {s.get('step_name')}: {s.get('avg_duration_ms')}ms")
Resultado de EjecuciΓ³nο
>>> Pasos mΓ‘s lentos... Viaje_L102_0 ββββββββββββββββββββββββββββββββββββββββ 100% 0:00:00 Viaje_L102_0 ββββββββββββββββββββββββββββββββββββββββ 100% 0:00:00 Viaje_L102_1 ββββββββββββββββββββββββββββββββββββββββ 100% 0:00:00 Viaje_L102_0 ββββββββββββββββββββββββββββββββββββββββ 100% 0:00:00 Viaje_L102_1 ββββββββββββββββββββββββββββββββββββββββ 100% 0:00:00 Viaje_L102_2 ββββββββββββββββββββββββββββββββββββββββ 100% 0:00:00π’ Pasos lentos: 0