Implementation:Neuml Txtai ExportTask
| Knowledge Sources | |
|---|---|
| Domains | Workflow, Data Export, ETL |
| Last Updated | 2026-02-10 01:00 GMT |
Overview
Concrete tool for exporting workflow task results to CSV or Excel files provided by txtai.
Description
The ExportTask class extends the base Task class to export workflow task outputs to file using Pandas. After executing the parent task's action pipeline on the input elements, it writes the results to either a CSV or Excel (.xlsx) file depending on the output file extension. The task optionally adds a UTC timestamp to the output filename for versioned exports. It requires the Pandas library, which is installed via the "workflow" extra.
Usage
Use ExportTask in txtai workflows when you need to persist workflow results to disk as structured tabular data. Configure it with an output file path and optionally enable timestamping. It is particularly useful for ETL pipelines where intermediate or final results need to be exported for downstream consumption or reporting.
Code Reference
Source Location
- Repository: Neuml_Txtai
- File: src/python/txtai/workflow/task/export.py
Signature
class ExportTask(Task):
def register(self, output=None, timestamp=None)
def __call__(self, elements, executor=None)
Import
from txtai.workflow.task.export import ExportTask
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| elements | list | Yes | List of data elements to process and export |
| executor | object | No | Optional executor instance for concurrent task actions |
| output | str | Yes (register) | Output file path; extension determines format (.xlsx for Excel, anything else for CSV) |
| timestamp | bool | No (register) | If True, appends a UTC timestamp (YYYYMMDDTHHMMSSz) to the output filename before the extension |
Outputs
| Name | Type | Description |
|---|---|---|
| outputs | list | The processed task results (same data that was written to file), returned for downstream pipeline consumption |
| file | CSV or XLSX | Side effect: written to the configured output path |
Usage Examples
from txtai.workflow.task.export import ExportTask
# Create an export task that writes results to CSV
export = ExportTask(
action=lambda x: [{"id": i, "text": t} for i, t in enumerate(x)],
output="results.csv",
timestamp=True
)
# Execute the task - processes elements and writes to timestamped CSV file
results = export(["text one", "text two", "text three"])
# Writes file like: results.20260210T010000Z.csv
# Export to Excel format
excel_export = ExportTask(
action=lambda x: x,
output="report.xlsx"
)
results = excel_export([{"name": "item1", "score": 0.95}])