sxd-sdk 0.1.0

SDK for building and deploying data pipelines on SentientX Data Platform

SXD SDK

Python library for building data pipelines on SentientX Data Platform.

Installation

pip install sxd-sdk

With optional dependencies:

pip install sxd-sdk[pytorch]   # PyTorch pipelines
pip install sxd-sdk[opencv]    # OpenCV pipelines

Quick Start

1. Create Pipeline

Create a new directory with this structure:

my-pipeline/
├── my_pipeline/
│   ├── __init__.py
│   ├── activities.py
│   └── workflows.py
├── sxd.yaml
└── pyproject.toml

2. Define Activities

# my_pipeline/activities.py
from temporalio import activity
from sxd_sdk import ProcessingResult

@activity.defn
async def process_video(video_path: str) -> ProcessingResult:
    # Your processing logic
    return ProcessingResult.ok(output_path="/output/result.mp4")

3. Define Workflow

# my_pipeline/workflows.py
from datetime import timedelta
from temporalio import workflow
from sxd_sdk import PipelineInput, PipelineOutput

with workflow.unsafe.imports_passed_through():
    from my_pipeline.activities import process_video

@workflow.defn(name="my-pipeline")
class MyPipelineWorkflow:
    @workflow.run
    async def run(self, input: PipelineInput) -> PipelineOutput:
        result = await workflow.execute_activity(
            process_video,
            args=[input.source_url],
            start_to_close_timeout=timedelta(minutes=30),
        )
        return PipelineOutput(status="success", output_path=result.output_path)

4. Configure (sxd.yaml)

base_image: sxd-base    # or: sxd-pytorch, sxd-opencv, sxd-cuda
timeout: 3600           # optional, default 3600s
gpu: false              # optional, default false

5. Publish

Use the sxd CLI (from the main SXD repo):

sxd publish .

This builds the Docker image, pushes it to the registry, and registers with the cluster.

6. Submit Jobs

sxd submit my-pipeline '{"source_url": "s3://bucket/video.mp4"}'

API Reference

Schemas

from sxd_sdk import (
    PipelineInput,    # Base input with source_url, customer_id, options
    PipelineOutput,   # Base output with status, output_path, metrics, errors
    ProcessingResult, # Activity result with success, output_path, error, metadata
)

# Convenience constructors
result = ProcessingResult.ok("/output/path", count=10)
result = ProcessingResult.fail("Something went wrong")

Testing

from sxd_sdk import MockActivityEnvironment, WorkflowSimulator

# Test activities
async def test_my_activity():
    env = MockActivityEnvironment()
    result = await env.run(my_activity, "input")
    assert result.success

# Test workflows
async def test_my_workflow():
    sim = WorkflowSimulator()
    sim.mock_activity("process_video", ProcessingResult.ok("/output"))
    result = await sim.run(MyWorkflow, input_data)
    assert result.status == "success"

Client

from sxd_sdk import SXDClient, connect

async with await connect(host="cluster.example.com") as client:
    job = await client.submit("my-pipeline", {"source_url": "..."})
    result = await client.wait(job.workflow_id)

Base Images

Image	Use Case
sxd-base	General Python processing
sxd-pytorch	ML/AI with PyTorch
sxd-opencv	Video/image processing
sxd-cuda	GPU workloads

License

Apache 2.0