id: dlt-elt-chess-api-to-duckdb
namespace: company.team

tasks:
  - id: chess_api_to_duckdb
    type: io.kestra.plugin.scripts.python.Script
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:slim
    beforeCommands:
      - pip install dlt[duckdb]
    warningOnStdErr: false
    script: |
      import dlt
      import requests

      pipeline = dlt.pipeline(
          pipeline_name='chess_pipeline',
          destination='duckdb',
          dataset_name='player_data'
      )
      data = []
      for player in ['magnuscarlsen', 'rpragchess']:
          response = requests.get(f'https://api.chess.com/pub/player/{player}')
          response.raise_for_status()
          data.append(response.json())
      # Extract, normalize, and load the data
      pipeline.run(data, table_name='player')

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    cron: 0 9 * * *