flyteidl2

import asyncio import flyte import os env = flyte.TaskEnvironment( name="hello_world_idl_example", image=flyte.Image.from_debian_base(python_version=(3, 11)), ) @env.task def calculate(x: int) -> int: # Flyteidl2 defines the underlying types and messages used by Flyte, # but users primarily interact with the high-level 'flyte' SDK. # For instance, the 'int' type here is automatically mapped to Flyte's # internal Literal type defined in flyteidl2.core.literals_pb2. return x * 2 + 5 @env.task async def main_workflow(numbers: list[int]) -> float: results = await asyncio.gather(*[calculate.aio(num) for num in numbers]) return sum(results) / len(results) if __name__ == "__main__": # For local execution, you need to initialize Flyte. # This assumes a local Flyte setup (e.g., Flyte Sandbox via 'flytectl start') # or a configuration file (.flyte/config.yaml). # Ensure Docker is running if using sandbox. # For a quick local run without a full cluster, 'flyte.init()' is for local debugging. flyte.init() # To run this on a local Flyte sandbox, you'd typically use the CLI: # flyte run --local example_script.py main_workflow --numbers '[1,2,3]' # For a simple local Python execution: print("Running locally...") # The 'run' method creates a local execution environment. run_result = flyte.run(main_workflow, numbers=list(range(10))) print(f"Result: {run_result.result}") # To explicitly show flyteidl2 is a dependency and available: try: from flyteidl2.core import literals_pb2 print(f"Successfully imported literals_pb2 from flyteidl2: {literals_pb2.__file__}") except ImportError: print("Could not import literals_pb2 from flyteidl2. Is it installed?")