Search OpenAlex for Related Scholarly Works

id: openalex-related-work-search
namespace: company.team
description: |
  Search OpenAlex for related scholarly works, inspect the response with DuckDB,
  and export a compact CSV artifact. Run it from the UI with a search query or
  call the keyed webhook with a JSON query.

triggers:
  - id: search_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Start a search from another workflow or application. The caller
      supplies a JSON body with a `query` field.
    key: "{{ secret('OPENALEX_WEBHOOK_KEY') }}"

inputs:
  - id: search_query
    type: STRING
    defaults: workflow orchestration
    description: Free-text OpenAlex search used by UI executions. A `query` field in
      the webhook request body overrides this value when supplied.

tasks:
  - id: download_works
    type: io.kestra.plugin.core.http.Download
    description: Fetch up to 25 matching works from the public OpenAlex Works API.
      OpenAlex encodes the search query as a request parameter; no API key is
      required for this request.
    uri: https://api.openalex.org/works
    params:
      search: "{{ ((trigger.body.query ?? '') | trim) | default(inputs.search_query)
        }}"
      per_page: "25"
      select: "id,doi,display_name,publication_date,cited_by_count"

  - id: count_matches
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: Read OpenAlex's total match count from the downloaded JSON response
      so the flow can report an empty search cleanly.
    inputFiles:
      openalex-response.json: "{{ outputs.download_works.uri }}"
    sql: |
      SELECT CAST(json_extract_string(response.json, '$.meta.count') AS BIGINT) AS total_matches
      FROM read_json_objects('openalex-response.json', format = 'unstructured') AS response;
    fetchType: FETCH_ONE

  - id: check_for_results
    type: io.kestra.plugin.core.flow.If
    description: Log whether OpenAlex found any matches. DuckDB writes the CSV
      afterward in either case.
    condition: "{{ outputs.count_matches.row.total_matches > 0 }}"
    then:
      - id: log_matches_found
        type: io.kestra.plugin.core.log.Log
        description: Report the total number of OpenAlex matches. The CSV contains at
          most the first 25 works returned by the API.
        message: "OpenAlex found {{ outputs.count_matches.row.total_matches }} works for
          '{{ ((trigger.body.query ?? '') | trim) | default(inputs.search_query)
          }}'. Exporting up to 25."
    else:
      - id: log_no_matches
        type: io.kestra.plugin.core.log.Log
        description: Make a zero-match search visible in the execution logs.
        message: "OpenAlex found no works for '{{ ((trigger.body.query ?? '') | trim) |
          default(inputs.search_query) }}'. The CSV export will contain no data
          rows."

  - id: shape_results
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: Flatten the Works response and write the selected fields to a CSV
      artifact, including a header when there are no matches.
    inputFiles:
      openalex-response.json: "{{ outputs.download_works.uri }}"
    outputFiles:
      - openalex-results.csv
    sql: |
      COPY (
        SELECT
          json_extract_string(work.value, '$.id') AS openalex_id,
          json_extract_string(work.value, '$.display_name') AS title,
          json_extract_string(work.value, '$.doi') AS doi,
          json_extract_string(work.value, '$.publication_date') AS publication_date,
          CAST(json_extract_string(work.value, '$.cited_by_count') AS BIGINT) AS cited_by_count
        FROM read_json_objects('openalex-response.json', format = 'unstructured') AS response,
             json_each(response.json, '$.results') AS work
        ORDER BY cited_by_count DESC NULLS LAST,
                 publication_date DESC NULLS LAST
      ) TO '{{ outputFiles["openalex-results.csv"] }}'
      (HEADER, DELIMITER ',');

outputs:
  - id: csv_uri
    type: URI
    displayName: OpenAlex results CSV
    description: URI of the CSV artifact with up to 25 works and their OpenAlex ID,
      title, DOI, publication date, and citation count.
    value: "{{ outputs.shape_results.outputFiles['openalex-results.csv'] }}"