Expressions with Namespace Files

How to pass expressions to Namespace Files.

You can write the script inline in your flow, and use expressions as part of this inline script. Here is an example flow whih has an expression within the inline script:

id: expressions_inline
namespace: company.team


inputs:
  - id: uri
    type: URI
    defaults: https://www.google.com/


tasks:
  - id: inline_script
    type: io.kestra.plugin.scripts.python.Script
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: ghcr.io/kestra-io/pydata:latest
    script: |
      import requests


      url = "{{ inputs.uri }}"
      response = requests.get(url)


      if response.status_code == 200:
          print(response.text)
      else:
          print(f"Failed to retrieve the webpage. Status code: {response.status_code}")

This approach is very convenient for scripts that are specific to the flow, and does not allow us to have a separate file for our code. Having a separate file has multiple benefits:

  • There’s multiple files of code being used - common with larger projects
  • The code is long and will be hard to maintain directly inside of the workflow - this makes the workflow harder to maintain too
  • The files are written and tested locally and synced to Kestra through Git
  • The files are used across multiple flows - this avoids repeating the code in each workflow

You cannot directly use Expressions inside of Namespace Files as they will not be rendered, or allow you to run your code outside of Kestra. There are 2 ways to allow you to pass expressions to your code:

  1. Using Namespace Files with environment variables
  2. Using Namespace Files with CLI arguments

In either case, we need to add our code as a Namespace File. To do this, you can use the Editor or import it directly.

namespace_file

Using Namespace Files with Environment Variables

You can pass inputs as environment variables using an expression.

This example uses the input uri and passes it to the task code as an environment variable so the Python code can access it

id: expressions_env_vars
namespace: company.team


inputs:
  - id: uri
    type: URI
    defaults: https://www.google.com/


tasks:
  - id: code
    type: io.kestra.plugin.scripts.python.Commands
    namespaceFiles:
      enabled: true
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: ghcr.io/kestra-io/pydata:latest
    commands:
      - python main.py
    env:
      URI: "{{ inputs.uri }}"

Inside of the Python code, we use os.environ to fetch the environment variable:

import requests
import os


# Perform the GET request
response = requests.get(os.environ['URI'])


# Check if the request was successful
if response.status_code == 200:
    # Print the content of the page
    print(response.text)
else:
    print(f"Failed to retrieve the webpage. Status code: {response.status_code}")

Using this method, it allows us to keep our code in a separate file, while also preventing our code from getting much longer than the inline example.

Using Namespace Files with CLI arguments

We can pass arguments directly to our code at execution. In Python, we can use argparse to achieve this.

Before we can execute our code, we will need to modify it to receive our argument correctly using the following:

import argparse
import requests


# Setup command line argument parsing
parser = argparse.ArgumentParser(description="Fetch the content of a given URL")
parser.add_argument("url", type=str, help="The URL to fetch")
args = parser.parse_args()


# Perform the GET request
response = requests.get(args.url)


# Check if the request was successful
if response.status_code == 200:
    # Print the content of the page
    print(response.text)
else:
    print(f"Failed to retrieve the webpage. Status code: {response.status_code}")

You can pass the arguments to your code using an expression. The expression will be rendered, and the evaluated values will be passed to the script via argparse:

id: expressions_argparse
namespace: company.team


inputs:
  - id: uri
    type: URI
    defaults: https://www.google.com/


tasks:
  - id: hello
    type: io.kestra.plugin.scripts.python.Commands
    namespaceFiles:
      enabled: true
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: ghcr.io/kestra-io/pydata:latest
    commands:
      - python main.py "{{ inputs.uri }}"

Using this method, the code has grows longer due to handling of arguments using argparse. This can be clearly seen by comparing the script in the Editor to the inline script used in the first approach. However, this method is very helpful from reusability perspective where the same script can be used in multiple flows without any code duplication of the script.