Metadata-Version: 2.4
Name: acceldata_aws_states_adapter
Version: 26.9.0
Summary: Acceldata AWS States Adapter
Author-email: acceldata <apisupport@acceldata.io>
License-Expression: MIT
Keywords: acceldata-aws-states-adapter
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: openlineage-python
Requires-Dist: boto3
Requires-Dist: attrs
Provides-Extra: release
Requires-Dist: build>=1; extra == "release"
Requires-Dist: twine>=5; extra == "release"
Provides-Extra: test
Requires-Dist: pytest>=8; extra == "test"
Dynamic: license-file

# Acceldata AWS States Adapter

`acceldata_aws_states_adapter` emits [OpenLineage](https://openlineage.io/) events for **AWS Step Functions** executions into **Acceldata ADOC**.

It is designed to run as an **AWS Lambda** (often behind EventBridge) that:

1. Receives Step Functions execution status-change events
2. Emits a **START** OpenLineage run when an execution begins (`RUNNING`)
3. Walks execution history on terminal statuses and emits OpenLineage events for individual states (Task, Map, Parallel, Choice, Wait, Pass, Succeed, Fail)
4. Sends events over HTTP to ADOC’s lineage endpoint

Python package import root: `acceldata`  
PyPI / distribution name: `acceldata_aws_states_adapter`

---

## Requirements

- Python **>= 3.10**
- Runtime dependencies: `openlineage-python`, `boto3`, `attrs`
- AWS permissions for the Lambda role (see [IAM permissions](#iam-permissions))
- ADOC URL + credentials for the OpenLineage HTTP transport

---

## Installation

### From PyPI

```bash
pip install acceldata_aws_states_adapter
```

### From TestPyPI

```bash
pip install acceldata-aws-states-adapter \
  --index-url https://test.pypi.org/simple/ \
  --extra-index-url https://pypi.org/simple/
```

### As a Lambda layer

Build a layer zip with a `python/` prefix (Lambda’s expected layout):

```bash
mkdir -p python
pip install acceldata_aws_states_adapter \
  --index-url https://test.pypi.org/simple/ \
  --extra-index-url https://pypi.org/simple/ \
  -t python/
zip -r layer.zip python
```

Attach `layer.zip` as a Lambda layer and keep your handler code (for example `lambda_function.py`) in the function package.

### Editable / local development

```bash
cd modules/acceldata-aws-states-adapter
pip install -e .
```

---

## How it works

```text
EventBridge (aws.states / Step Functions Execution Status Change)
        │
        ▼
   AWS Lambda handler
        │
        ├── status == RUNNING
        │       └── ProcessorService.send_openlineage_event_for_state_machine_start(...)
        │               • describe state machine definition
        │               • build job facets (upstream / child states)
        │               • optional ParentRunFacet from execution input
        │               • emit OpenLineage START for the state machine run
        │
        └── terminal status (SUCCEEDED / FAILED / TIMED_OUT / ABORTED / ...)
                └── ProcessorService.process_step_function_execution(...)
                        • get_execution_history
                        • dispatch per-state handlers (FlowFactory)
                        • emit START/COMPLETE/FAIL/... events per state
                        • emit terminal state-machine run event + run facet
```

### Supported Step Functions state types

| State type | Handler |
|------------|---------|
| Task | `Task` |
| Map | `Map` |
| Parallel | `Parallel` |
| Choice | `Choice` |
| Wait | `Wait` |
| Pass | `Pass` |
| Succeed | `Success` |
| Fail | `Fail` |

Handlers live under `acceldata.step_functions.flows` and are registered in `FlowFactory`.

### OpenLineage mapping (high level)

| Concept | Typical value |
|---------|----------------|
| Job namespace | From `ADOC__OPENLINEAGE__NAMESPACE`, or default `aws-<accountId>-<region>` derived from the state machine ARN |
| State machine job name | State machine name (ARN suffix) |
| State job name | `<stateMachineName>.<stateName>` |
| State machine run id | Step Functions execution name / id |
| Parent run | Optional, from execution input keys `aws.states.openlineage.*` |

Custom facets include:

- `AWSStateMachineJobFacet` — upstream and child state topology
- `AWSStateMachineRunFacet` — per-state terminal status map
- `AWSStateMachineStateInputDetailsFacet` / `AWSStateMachineStateOutputDetailsFacet` — entered/exited state details

Events are posted to ADOC at:

```text
{ADOC__OPENLINEAGE__TRANSPORT__URL}/torch-pipeline/api/v1/lineage
```

using HTTP headers `accessKey` and `secretKey`.

---

## Configuration

Configure the Lambda (or process) with these environment variables.

### Required ADOC transport

| Environment variable | Description |
|----------------------|-------------|
| `ADOC__OPENLINEAGE__TRANSPORT__URL` | ADOC base URL (host only; endpoint path is appended by the client) |
| `ADOC__OPENLINEAGE__TRANSPORT__ACCESS_KEY` | ADOC access key |
| `ADOC__OPENLINEAGE__TRANSPORT__SECRET_KEY` | ADOC secret key (unless Secrets Manager is enabled) |

### Optional

| Environment variable | Description |
|----------------------|-------------|
| `ADOC__OPENLINEAGE__NAMESPACE` | OpenLineage job namespace override (recommended human-readable account/env name) |
| `ADOC__OPENLINEAGE__TRANSPORT__SECRET_MANAGER_ENABLED` | Set to `true` to load the secret key from AWS Secrets Manager |
| `ADOC__OPENLINEAGE__TRANSPORT__SECRET_MANAGER_SECRET_NAME` | Secrets Manager secret id/name |
| `ADOC__OPENLINEAGE__TRANSPORT__SECRET_MANAGER_REGION` | Region for the Secrets Manager client |

If `ADOC__OPENLINEAGE__NAMESPACE` is unset, the adapter defaults to:

```text
aws-<accountId>-<region>
```

parsed from the state machine ARN (for example `aws-241571801844-ap-south-1`).

### Parent run (optional)

If the Step Functions execution **input** JSON includes all of:

- `aws.states.openlineage.parentRunId`
- `aws.states.openlineage.parentJobName`
- `aws.states.openlineage.parentJobNamespace`

the START event for the state machine run includes an OpenLineage `ParentRunFacet`.

Example execution input fragment:

```json
{
  "TransactionType": "PURCHASE",
  "aws.states.openlineage.parentRunId": "1f0f2305-6017-42a6-af78-002fec15eba2",
  "aws.states.openlineage.parentJobName": "Step Functions StartExecution",
  "aws.states.openlineage.parentJobNamespace": "TriggerStateMachine"
}
```

---

## IAM permissions

Minimum permissions typically required by the Lambda role:

```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "states:DescribeStateMachine",
        "states:GetExecutionHistory"
      ],
      "Resource": "*"
    }
  ]
}
```

If using Secrets Manager for the ADOC secret key, also allow:

```json
{
  "Effect": "Allow",
  "Action": ["secretsmanager:GetSecretValue"],
  "Resource": "arn:aws:secretsmanager:<region>:<account-id>:secret:<secret-name>*"
}
```

Tighten `Resource` ARNs to specific state machines / secrets in production.

---

## Usage (Lambda handler)

Wire EventBridge rule(s) for `Step Functions Execution Status Change` to a Lambda. Example handler:

```python
import json

from acceldata.client.client_initialiser import ClientInitialiser
from acceldata.service.processor_service import ProcessorService


def lambda_handler(event, context):
    event_details = event["detail"]
    status = event_details["status"]

    client = ClientInitialiser.initialise_client_configuration().client
    service = ProcessorService(openlineage_client=client)

    if status == "RUNNING":
        service.send_openlineage_event_for_state_machine_start(
            state_machine_arn=event_details["stateMachineArn"],
            execution_id=event_details["name"],
            event_time=event_details["startDate"],
            event_details=event_details,
        )
    else:
        # SUCCEEDED / FAILED / TIMED_OUT / ABORTED / etc.
        service.process_step_function_execution(
            state_machine_arn=event_details["stateMachineArn"],
            state_machine_execution_arn=event_details["executionArn"],
            client=client,
        )

    return {"statusCode": 200, "body": json.dumps("ok")}
```

### Expected EventBridge shape

The adapter expects the standard EventBridge envelope. The handler uses `event["detail"]`, which includes fields such as:

- `status`
- `name` (execution id)
- `stateMachineArn`
- `executionArn`
- `startDate` (epoch millis)
- `input` (JSON **string** of the execution input, when included)

---

## Package layout

```text
acceldata/
  client/           # ADOC OpenLineage HTTP client + env/Secrets Manager bootstrap
  constants/        # Env var names and Step Functions field constants
  facets/           # Custom OpenLineage facets for AWS Step Functions
  service/          # ProcessorService + Step Functions describe/history services
  step_functions/   # FlowFactory, utils, per-state flow handlers
```

Primary entry points for integrators:

- `acceldata.client.client_initialiser.ClientInitialiser`
- `acceldata.service.processor_service.ProcessorService`

---

## Building and publishing

```bash
cd modules/acceldata-aws-states-adapter
pip install -U build twine
rm -rf dist/ build/ *.egg-info
python -m build
twine check dist/*

# TestPyPI
twine upload --repository testpypi dist/*

# PyPI (after validation)
twine upload dist/*
```

Version is read from the `VERSION` file (dynamic setuptools version).

---

## License

MIT — see `LICENSE`.
