Implementation:Neuml Txtai StorageTask
| Knowledge Sources | |
|---|---|
| Domains | Workflow, Cloud Storage, ETL |
| Last Updated | 2026-02-10 01:00 GMT |
Overview
Concrete tool for listing and processing cloud object storage buckets within workflows provided by txtai.
Description
The StorageTask class extends the base Task class to process object storage buckets backed by Apache libcloud. It accepts storage-style URL elements (e.g., s3://bucket/prefix), parses the provider scheme, connects to the cloud storage backend, lists all objects in the specified container (with optional prefix filtering), and expands each storage URL into a list of CDN URLs for individual objects. These expanded URLs are then processed through the task's action pipeline. Non-storage elements are passed through unchanged. The class supports provider-specific credentials via constructor parameters or environment variables (ACCESS_KEY, ACCESS_SECRET).
Usage
Use StorageTask in txtai workflows when you need to process files stored in cloud object storage services (S3, GCS, Azure, MinIO, etc.). It is typically placed at the beginning of a workflow to expand a bucket reference into individual file URLs, which are then processed by downstream tasks (e.g., RetrieveTask, text extraction). Configure it with cloud credentials and optional connection parameters.
Code Reference
Source Location
- Repository: Neuml_Txtai
- File: src/python/txtai/workflow/task/storage.py
Signature
class StorageTask(Task):
PREFIX = r"(\w+):\/\/.*"
PATH = r"\w+:\/\/(.*)"
def register(self, key=None, secret=None, host=None, port=None, token=None, region=None)
def __call__(self, elements, executor=None)
def matches(self, element)
def list(self, element)
Import
from txtai.workflow.task.storage import StorageTask
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| elements | list | Yes | List of elements; storage URLs (e.g., "s3://bucket/prefix") are expanded, non-storage elements pass through |
| executor | object | No | Optional executor instance for concurrent task actions |
| key | str | No (register) | Provider-specific access key; falls back to ACCESS_KEY environment variable |
| secret | str | No (register) | Provider-specific access secret; falls back to ACCESS_SECRET environment variable |
| host | str | No (register) | Server host name for custom endpoints |
| port | int | No (register) | Server port number |
| token | str | No (register) | Temporary session token |
| region | str | No (register) | Storage region identifier |
Outputs
| Name | Type | Description |
|---|---|---|
| outputs | list | Aggregated list of processed results; storage URLs are expanded to individual object CDN URLs and run through task actions; non-storage elements are passed through unchanged |
Usage Examples
from txtai.workflow.task.storage import StorageTask
# Create a storage task to list and process S3 objects
storage = StorageTask(
action=lambda urls: urls,
key="AKIAIOSFODNN7EXAMPLE",
secret="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
region="us-east-1"
)
# Process a storage URL - expands to individual object URLs
results = storage(["s3://my-bucket/data-prefix"])
# Returns list of CDN URLs for all objects in my-bucket with prefix "data-prefix"
# Mix storage URLs with regular elements
results = storage([
"s3://my-bucket/documents",
"regular text element"
])
# Storage URL is expanded; regular element passes through unchanged
# Use with environment variables (ACCESS_KEY, ACCESS_SECRET)
storage_env = StorageTask(
action=lambda urls: urls
)
results = storage_env(["gcs://my-gcs-bucket"])