Source
yaml
id: python-subflow-component
namespace: company.team
inputs:
- id: arg1
type: INT
defaults: 1
- id: arg2
type: INT
defaults: 2
tasks:
- id: python
type: io.kestra.plugin.scripts.python.Commands
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: ghcr.io/kestra-io/pydata:latest
inputFiles:
main.py: |
import argparse
from kestra import Kestra
def multiply_arguments(arg1, arg2):
return arg1 * arg2
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='Multiply two integers.')
parser.add_argument('--arg1', type=int, help='First integer argument')
parser.add_argument('--arg2', type=int, help='Second integer argument')
args = parser.parse_args()
result = multiply_arguments(args.arg1, args.arg2)
print(f"The product of {args.arg1} and {args.arg2} is: {result}")
Kestra.outputs({'result': result})
commands:
- python main.py --arg1 {{ inputs.arg1 }} --arg2 {{ inputs.arg2 }}
About this blueprint
Inputs Python Outputs
This flow shows how you can create a templated flow (subflow) to run a custom script.
This flow can be used in another flow, acting like a separated component. We can imagine having a complex flow, with many tasks but abstracted with inputs and outputs, so users can only deal with a simple interface.
Here is an example of calling this flow, giving inputs and retrieve the desired outputs.
text
id: call_python_component
namespace: company.team
tasks:
- id: python
type: io.kestra.plugin.core.flow.Subflow
namespace: company.team
flowId: python_subflow_component
wait: true
inputs:
arg1: 5
arg2: 3
outputs:
result: '{{ outputs.python.vars.result }}'