Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions .github/scripts/type-checks.sh
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,10 @@ mypy --install-types --non-interactive \
packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel \
packages/aws-durable-execution-sdk-python-otel/tests

mypy --install-types --non-interactive \
packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight \
packages/aws-durable-execution-sdk-python-insight/tests

# comment out this for now as there are many type check errors in this package
#mypy --install-types --non-interactive \
# packages/aws-durable-execution-sdk-python-testing/src/aws_durable_execution_sdk_python_testing \
Expand Down
3 changes: 3 additions & 0 deletions .github/workflows/pypi-publish.yml
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@ jobs:
path: packages/aws-durable-execution-sdk-python-otel
- name: aws-durable-execution-sdk-python-testing
path: packages/aws-durable-execution-sdk-python-testing
- name: aws-durable-execution-sdk-python-insight
path: packages/aws-durable-execution-sdk-python-insight

steps:
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
Expand Down Expand Up @@ -64,6 +66,7 @@ jobs:
- name: aws-durable-execution-sdk-python
- name: aws-durable-execution-sdk-python-otel
- name: aws-durable-execution-sdk-python-testing
- name: aws-durable-execution-sdk-python-insight
permissions:
id-token: write

Expand Down
62 changes: 62 additions & 0 deletions packages/aws-durable-execution-sdk-python-insight/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
# AWS Durable Execution SDK for Python — Workflow Insight plugin

Workflow Insight instrumentation plugin for the AWS Durable Execution SDK for
Python. A port of the JavaScript SDK's `workflowInsight()` plugin: it listens to
the SDK's instrumentation hooks and emits one curated `WorkflowInsight` record
per execution to the configured exporters. The wire record keeps the JS
camelCase field names so records read identically across SDKs.

> **Experimental.** Like its JS counterpart, this plugin is experimental and may
> change or be removed in future releases.

## Install

```bash
pip install aws-durable-execution-sdk-python-insight
# with the S3 exporter's local-dev dependency:
pip install "aws-durable-execution-sdk-python-insight[s3]"
```

## Usage

```python
from aws_durable_execution_sdk_python import durable_execution
from aws_durable_execution_sdk_python_insight import (
WorkflowInsightConfig,
workflow_insight,
)
from aws_durable_execution_sdk_python_insight.exporters import S3Exporter

@durable_execution(
plugins=[
workflow_insight(
WorkflowInsightConfig(
exporters=[
S3Exporter(bucket="my-bucket", prefix="workflow-insight/")
],
)
)
Comment thread
wangyb-A marked this conversation as resolved.
]
)
def handler(event, context):
...
```

With no exporter configured, records are written to the function's own
CloudWatch log group as single JSON lines (the `LambdaLogExporter` default),
carrying the name-keyed `operationsByName` summary. The `S3Exporter` writes the
lossless per-occurrence `operations` array, one object per execution
(upsert-by-execution-name, so re-emission overwrites rather than appends).

Emission behavior, record schema (`recordType: WorkflowInsight`,
`schemaVersion: "1.0"`), sampling, content configuration (input/output
omission, `include_errors`, per-operation result opt-in), truncation phases,
and `top-level` vs `full-tree` operation detail all mirror the JS plugin.
Behavior is validated cross-SDK by the `insight` conformance suite
(`aws-durable-execution-conformance-tests-insight`).

## Requirements

- `aws-durable-execution-sdk-python` with the plugin invocation hooks that
surface `execution_input` / `execution_result` (included since the version
this package declares as its minimum).
79 changes: 79 additions & 0 deletions packages/aws-durable-execution-sdk-python-insight/pyproject.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,79 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "aws-durable-execution-sdk-python-insight"
Comment thread
wangyb-A marked this conversation as resolved.
Comment thread
wangyb-A marked this conversation as resolved.
dynamic = ["version"]
description = 'Workflow Insight instrumentation plugin for the AWS Durable Execution SDK for Python'
readme = "README.md"
requires-python = ">=3.11"
license = "Apache-2.0"
keywords = ["observability", "workflow-insight", "durable-execution"]
authors = [{ name = "AWS durable-execution-dev", email = "durable-execution-dev@amazon.com" }]
classifiers = [
"Development Status :: 4 - Beta",
"Programming Language :: Python",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: 3.14",
"Programming Language :: Python :: Implementation :: CPython",
]
dependencies = [
# >=1.8.0: first release carrying the plugin invocation-hook fields
# (InvocationInfo.execution_input / InvocationEndInfo.execution_result).
"aws-durable-execution-sdk-python>=1.8.0",
]

[project.optional-dependencies]
# boto3 is provided by the Lambda runtime; declared as an extra for local dev
# (e.g. the S3Exporter) without vendoring it into deployments.
s3 = ["boto3>=1.26.0"]

[project.urls]
Documentation = "https://github.com/aws/aws-durable-execution-sdk-python#readme"
Issues = "https://github.com/aws/aws-durable-execution-sdk-python/issues"
Source = "https://github.com/aws/aws-durable-execution-sdk-python"

[tool.hatch.build.targets.sdist.force-include]
"../../LICENSE" = "LICENSE"
"../../NOTICE" = "NOTICE"

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

add a plugin entry point here to allow this plugin to be auto loaded. See otel plugin for reference


[tool.hatch.build.targets.wheel]
packages = ["src/aws_durable_execution_sdk_python_insight"]

[tool.hatch.build.targets.wheel.force-include]
"../../LICENSE" = "aws_durable_execution_sdk_python_insight/LICENSE"
"../../NOTICE" = "aws_durable_execution_sdk_python_insight/NOTICE"

[tool.hatch.version]
path = "src/aws_durable_execution_sdk_python_insight/__about__.py"

[tool.hatch.publish.index]
disable = true

[tool.coverage.run]
source_pkgs = ["aws_durable_execution_sdk_python_insight"]
branch = true
parallel = true
omit = ["src/aws_durable_execution_sdk_python_insight/__about__.py"]

[tool.coverage.report]
exclude_lines = ["no cov", "if __name__ == .__main__.:", "if TYPE_CHECKING:"]

[tool.ruff]
line-length = 88
target-version = "py311"

[tool.ruff.lint]
preview = true
select = ["E4", "E7", "E9", "F", "TID252"]

[tool.ruff.lint.isort]
known-first-party = ["aws_durable_execution_sdk_python_insight"]
force-single-line = false
lines-after-imports = 2

[tool.ruff.lint.per-file-ignores]
"tests/**" = ["ARG001", "ARG002", "ARG005", "S101", "PLR2004", "PLR6301", "SIM117", "TRY301"]
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates.
#
# SPDX-License-Identifier: Apache-2.0
__version__ = "0.0.1"
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates.
#
# SPDX-License-Identifier: Apache-2.0
"""Workflow Insight instrumentation plugin for the AWS Durable Execution Python SDK."""

from aws_durable_execution_sdk_python_insight.__about__ import __version__
from aws_durable_execution_sdk_python_insight.exporters import (
LambdaLogExporter,
S3Exporter,
S3Partitioning,
)
from aws_durable_execution_sdk_python_insight.operations_index import (
build_operations_by_name,
with_operations_by_name,
)
from aws_durable_execution_sdk_python_insight.plugin import (
WorkflowInsightPlugin,
workflow_insight,
)
from aws_durable_execution_sdk_python_insight.truncation import truncate_record
from aws_durable_execution_sdk_python_insight.types import (
ContentConfig,
ContentOperations,
EmitMode,
InsightExporter,
OperationDetail,
OperationOverride,
WorkflowInsightConfig,
)


__all__ = [
"__version__",
"ContentConfig",
"ContentOperations",
"EmitMode",
"InsightExporter",
"LambdaLogExporter",
"OperationDetail",
"OperationOverride",
"S3Exporter",
"S3Partitioning",
"WorkflowInsightConfig",
"WorkflowInsightPlugin",
"build_operations_by_name",
"truncate_record",
"with_operations_by_name",
"workflow_insight",
]
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates.
#
# SPDX-License-Identifier: Apache-2.0
"""First-party Workflow Insight exporters.

One module per exporter, mirroring the JS package's ``src/exporters/`` layout
(``aws-durable-execution-sdk-js-insight``). Each destination lives in its own
module so the set can grow to the full JS parity surface (S3, CloudWatch Logs,
DynamoDB, Firehose, EventBridge, SQS, OpenSearch, Redshift, Aurora, HTTP, OTel,
file, ...) without any single file accreting every backend's imports and
optional dependencies.

Concrete exporters are re-exported here so the public import path is stable:
``from aws_durable_execution_sdk_python_insight.exporters import S3Exporter``
keeps working exactly as before this package was split out of a single module.
Shared serialization helpers live in the private ``_common`` module.

Both shipped exporters serialize the curated record with JS-compatible compact
JSON (no whitespace) so the wire bytes match across SDKs. Records are written
verbatim -- no synthetic emission.
"""

from __future__ import annotations

from aws_durable_execution_sdk_python_insight.exporters.lambda_log_exporter import (
LambdaLogExporter,
)
from aws_durable_execution_sdk_python_insight.exporters.s3_exporter import (
S3Exporter,
S3Partitioning,
)


__all__ = [
"LambdaLogExporter",
"S3Exporter",
"S3Partitioning",
]
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates.
#
# SPDX-License-Identifier: Apache-2.0
"""Shared serialization helpers for the Workflow Insight exporters.

Kept private to the ``exporters`` package: every backend needs the same
JS-compatible compact JSON encoding and the same key/file-name sanitizer, so
they live here rather than being duplicated per exporter module.
"""

from __future__ import annotations

import json
import re
from typing import Any


def compact_dumps(value: Any) -> str:
"""Serialize ``value`` as compact JSON (no whitespace, non-ASCII preserved).

Matches the JS exporters' ``JSON.stringify`` output so the wire bytes are
identical across SDKs.
"""
return json.dumps(value, separators=(",", ":"), ensure_ascii=False)


def sanitize(value: str) -> str:
"""Replace characters unsafe for object keys / file names with ``_``."""
return re.sub(r"[^a-zA-Z0-9._-]", "_", value)
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates.
#
# SPDX-License-Identifier: Apache-2.0
"""Lambda log (CloudWatch) Workflow Insight exporter."""

from __future__ import annotations

from typing import Any

from aws_durable_execution_sdk_python_insight.exporters._common import compact_dumps
from aws_durable_execution_sdk_python_insight.operations_index import (
with_operations_by_name,
)


class LambdaLogExporter:
"""Writes ``operationsByName`` records to the function's own log group via ``print``.

Port of the JS ``LambdaLogExporter``: ``console.log(JSON.stringify(
withOperationsByName(record)))``. Requires no extra IAM. Emits the name-keyed
summary map (``OPERATIONS_BY_NAME``).
"""

def __init__(self, max_record_size_bytes: int | None = None) -> None:
self.max_record_size_bytes: int | None = (
256_000 if max_record_size_bytes is None else max_record_size_bytes
)

def render(self, record: dict[str, Any]) -> dict[str, Any]:
return with_operations_by_name(record)

def export(self, record: dict[str, Any]) -> None:
# Raw JSON line to stdout -> the function's CloudWatch log group. The
# conformance CloudWatch sink json.loads each line (and unwraps the
# Lambda structured-log envelope when present).
print(compact_dumps(self.render(record)), flush=True) # noqa: T201

def flush(self) -> None:
return None
Loading
Loading