dbnd

import os from dbnd import task, pipeline, dbnd_context # Configure DBND to log to console for demonstration # In a real scenario, you'd typically integrate with a Databand tracking server os.environ['DBND__CORE__TRACKER'] = 'console' # Ensure console logging is enabled os.environ['DBND__CORE__DATABAND_URL'] = os.environ.get('DATABAND_URL', 'http://localhost:8080') # Placeholder os.environ['DBND__CORE__DATABAND_ACCESS_TOKEN'] = os.environ.get('DATABAND_ACCESS_TOKEN', 'YOUR_API_KEY') # Placeholder @task def calculate_alpha(alpha: float) -> float: print(f"Alpha is: {alpha}") return alpha * 2 @pipeline def alpha_pipeline() -> float: val1 = calculate_alpha(alpha=1.0) val2 = calculate_alpha(alpha=val1) return val2 if __name__ == "__main__": # Using dbnd_context to ensure configuration is applied for programmatic runs with dbnd_context(conf={"core": {"tracker": ["console"]}}): result = alpha_pipeline() print(f"\nPipeline finished with result: {result}") print("\nTo run via CLI with full tracking (if configured): dbnd run alpha_pipeline")