We are excited to share the release of watsonx.data integration SDK version 1.3. This update brings real-time observability to streaming jobs, stronger guardrails for project assets, and smarter Python code generation, making it easier than ever to monitor, manage, and migrate your data integration workloads programmatically.

Streaming Job Run Logs and Metrics

Monitoring streaming pipelines is now a first class experience. With 1.3.0, the JobRun.logs and JobRun.metrics properties are fully supported for streaming jobs, giving you live visibility into your continuously running pipelines directly from Python.

Accessing streaming job run logs

Retrieve runtime logs as a simple list of strings to troubleshoot your streaming jobs without leaving your notebook or script.

Code:

streaming_job = project.jobs.get(name="My Streaming Job")
job_run = streaming_job.start(name="Production Run")

# Refresh the run once it is active
job_run = streaming_job.job_runs.get(job_run_id=job_run.job_run_id)

job_run.logs

Output:

['2026-06-10 20:53:34,085 [INFO] Received 1 messages from Kafka',
 '2026-06-10 20:53:34,140 [INFO] Received 1 messages from Kafka',
 '2026-06-10 20:53:34,193 [INFO] Received 1 messages from Kafka',
 ...]

Live streaming metrics

The new JobRun.metrics property returns a structured StreamingJobRunMetrics object with a pipeline level summary and per stage breakdowns, so you can track throughput and error rates in real time.

Code: job_run.metrics.pipeline_summary

Output:

{
    "input_records": 30375,
    "output_records": 30375,
    "error_records": 0,
    "stage_errors": 0,
    "batch_count": 0,
    "total_runners": 0,
    "available_runners": 0,
    "input_rate_per_sec": 1976.0952511675935,
    "output_rate_per_sec": 1976.0952511675935,
    "error_rate_per_sec": 0.0
}

You can also drill into individual stages, including processing time percentiles:

Code: job_run.metrics.stages[0]

Output:

{
    "stage_id": "KafkaMultitopicConsumer_01",
    "stage_name": "KafkaMultitopicConsumer_01",
    "input_records": 0,
    "output_records": 30375,
    "error_records": 0,
    "stage_errors": 0,
    "input_rate_per_sec": 0.0,
    "output_rate_per_sec": 1976.0952511675935,
    "avg_processing_time_sec": 2.021533333333333,
    "min_processing_time_sec": 2.0010000000000003,
    "max_processing_time_sec": 2.055,
    "p95_processing_time_sec": 2.0511500000000003,
    "p99_processing_time_sec": 2.055,
    "batch_count": 63
}

Note: Streaming job run logs and metrics are only available while the run is active. They are not stored after the job run reaches a terminal state, so fetch them while the job is still running.

Project Asset Ownership Validation

The SDK now validates that an asset actually belongs to the project you are operating on before performing updates or deletions. Instead of confusing API errors when a flow or job from one project is accidentally passed to another, you get an immediate, clear validation error. This protects asset integrity in environments where teams work across many projects.

Code:

project_a = platform.projects.get(name="Project A")
project_b = platform.projects.get(name="Project B")

flow = project_a.flows.get(name="My Flow")

# The SDK now catches this mismatch before the request is sent
project_b.update_flow(flow)

Output:

AssetNotInProjectException: Asset StreamingFlow(name='My Flow' ...) does not belong to project Project B

The operation fails fast with a clear exception before any request reaches the platform, so a mismatched asset can never be written into the wrong project.

Python Generator Improvements

The PythonGenerator, which converts existing flows into executable Python SDK code, received significant enhancements across batch flows, streaming flows, and connections. These improvements produce more complete and accurate generated code, making it easier to bring existing pipelines under programmatic control.

Code:

from ibm_watsonx_data_integration.codegen import PythonGenerator

generator = PythonGenerator(
    source=flow,                                                      # Flow object to export
    destination="generated_flow.py",              # Output file path
    auth=auth,
    base_api_url=base_api_url,
    mask_credentials=True,                                # Replace secrets with env vars
)
generator.save()

The generated script is a complete, runnable SDK program that recreates your flow — authentication, project lookup, every stage with its configuration, and the links between them:

Output (excerpt of generated_flow.py):

kafka_multitopic_consumer_1 = flow.add_stage("Kafka Multitopic Consumer", type="origin")
kafka_multitopic_consumer_1.broker_uri = "kafka:9092"
kafka_multitopic_consumer_1.consumer_group = "fraud-demo-enrichment"
kafka_multitopic_consumer_1.topic_list = ["authorizations"]
kafka_multitopic_consumer_1.data_format = "JSON"

redis_lookup_processor_1 = flow.add_stage("Redis Lookup Processor", type="processor")
redis_lookup_processor_1.port_number = 6379
redis_lookup_processor_1.enable_local_caching = True
...

With mask_credentials=True (the default), sensitive values like API keys and passwords are replaced with environment variable lookups, so generated code is safe to commit and share.

Quality of Life Enhancements and Bug Fixes

In addition to these features, 1.3.0 includes numerous quality of life enhancements and bug fixes that improve the overall stability and reliability of the SDK.

To get started with our SDK install the ibm-watsonx-data-integration via pip today

pip3 install ibm-watsonx-data-integration

To see more details and view code examples, visit our documentation here.

Try out watsonx.data integration for free today!