Commands

id: python_venv
namespace: company.team

inputs:
  - id: nr
    type: INT
    defaults: 21

tasks:
  - id: python
    type: io.kestra.plugin.scripts.python.Commands
    namespaceFiles:
      enabled: true
      folderPerNamespace: true
      namespaces:
        - company
        - company.team
        - company.team.project # Explicitly name all namespaces to include; by default only the flow namespace is loaded.
      include:
        - etl_script.py
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    beforeCommands:
      - uv venv --python 3.13
      - uv pip install requests
    commands:
      - python etl_script.py --num {{inputs.nr}}

id: python_commands_example
namespace: company.team

tasks:
  - id: wdir
    type: io.kestra.plugin.core.flow.WorkingDirectory
    tasks:
      - id: clone_repository
        type: io.kestra.plugin.git.Clone
        url: https://github.com/kestra-io/examples
        branch: main

      - id: git_python_scripts
        type: io.kestra.plugin.scripts.python.Commands
        containerImage: ghcr.io/kestra-io/pydata:latest
        beforeCommands:
          - pip install faker > /dev/null
        commands:
          - python examples/scripts/etl_script.py
          - python examples/scripts/generate_orders.py
        outputFiles:
          - orders.csv

  - id: load_csv_to_s3
    type: io.kestra.plugin.aws.s3.Upload
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_KEY_ID') }}"
    region: eu-central-1
    bucket: kestraio
    key: stage/orders.csv
    from: "{{ outputs.gitPythonScripts.outputFiles['orders.csv'] }}"

id: gpu_task
namespace: company.team

tasks:
  - id: python
    type: io.kestra.plugin.scripts.python.Commands
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    commands:
      - python ml_on_gpu.py
    workerGroup:
      key: gpu

id: python_input_as_env_variable
namespace: company.team

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

tasks:
  - id: code
    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 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}")
    env:
      URI: "{{ inputs.uri }}"
    commands:
      - python main.py

id: s3_trigger_commands
namespace: company.team
description: process CSV file from S3 trigger

tasks:
  - id: wdir
    type: io.kestra.plugin.core.flow.WorkingDirectory
    tasks:
      - id: clone_repository
        type: io.kestra.plugin.git.Clone
        url: https://github.com/kestra-io/examples
        branch: main

      - id: python
        type: io.kestra.plugin.scripts.python.Commands
        inputFiles:
          data.csv: "{{ trigger.objects | jq('.[].uri') | first }}"
        description: this script reads a file `data.csv` from S3 trigger
        containerImage: ghcr.io/kestra-io/pydata:latest
        commands:
          - python examples/scripts/clean_messy_dataset.py
        outputFiles:
          - "*.csv"
          - "*.parquet"

triggers:
  - id: wait_for_s3_object
    type: io.kestra.plugin.aws.s3.Trigger
    bucket: declarative-orchestration
    maxKeys: 1
    interval: PT1S
    filter: FILES
    action: MOVE
    prefix: raw/
    moveTo:
      key: archive/raw/
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_KEY_ID') }}"
    region: "{{ secret('AWS_DEFAULT_REGION') }}"

id: python_in_container
namespace: company.team

tasks:
  - id: wdir
    type: io.kestra.plugin.core.flow.WorkingDirectory
    tasks:
      - id: clone_repository
        type: io.kestra.plugin.git.Clone
        url: https://github.com/kestra-io/examples
        branch: main

      - id: git_python_scripts
        type: io.kestra.plugin.scripts.python.Commands
        commands:
          - python examples/scripts/etl_script.py
        outputFiles:
          - "*.csv"
          - "*.parquet"
        containerImage: annageller/kestra:latest
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          config: |
            {
              "auths": {
                  "https://index.docker.io/v1/": {
                      "username": "annageller",
                      "password": "{{ secret('DOCKER_PAT') }}"
                  }
              }
            }

id: script_in_venv
namespace: company.team
tasks:
  - id: python
    type: io.kestra.plugin.scripts.python.Commands
    packageManager: UV
    inputFiles:
      main.py: |
        import requests
        from kestra import Kestra

        response = requests.get('https://google.com')
        print(response.status_code)
        Kestra.outputs({'status': response.status_code, 'text': response.text})
    beforeCommands:
      - python -m venv venv
      - . venv/bin/activate
      - pip install requests kestra > /dev/null
    commands:
      - python main.py