Building a Python Framework to Monitor IDMC Taskflow Execution
Starting an Informatica Intelligent Data Management Cloud (IDMC) taskflow from code is easy. Knowing what happened to it afterward is harder. In most pipelines I've reviewed, the "monitoring" is a scheduled job that fir
Starting an Informatica Intelligent Data Management Cloud (IDMC) taskflow from code is easy. Knowing what happened to it afterward is harder.
In most pipelines I've reviewed, the "monitoring" is a scheduled job that fires the taskflow and hopes for the best. When a run hangs, gets suspended, or quietly fails at 2 a.m., nobody finds out until a downstream report is empty.
This post walks through a small Python framework that fixes that. It:
Authenticates to IDMC and renews the session when it expires
Starts a taskflow and captures its run ID
Polls run status with exponential backoff and a hard timeout
Treats FAILED, SUSPENDED, and timeouts as distinct outcomes
Sends alerts through pluggable notifiers (Slack, Amazon SNS, or anything else)
Emits structured JSON logs you can query in CloudWatch, Datadog, or Splunk
All of it fits in about 300 lines and works the same on a laptop, in a container, or inside AWS Lambda.
Note: IDMC endpoint paths and status values can vary by pod and release. The ones below match the public REST API docs at the time of writing. Check them against your org's API reference before going to production.
The architecture
The framework has four parts:
Step 1: Configuration
Keep credentials out of code. A small dataclass reads them from environment variables, and in AWS those can come from Secrets Manager.
# idmc_monitor/config.py
import os
from dataclasses import dataclass, field
@dataclass(frozen=True)
class IDMCConfig:
username: str
password: str
login_url: str = "https://dm-us.informaticacloud.com/saas/public/core/v3/login"
poll_initial_seconds: float = 10.0
poll_max_seconds: float = 120.0
timeout_seconds: float = 3600.0
request_timeout: float = 30.0
terminal_failure_states: frozenset = field(
default_factory=lambda: frozenset({"FAILED", "SUSPENDED", "FAULTED", "ABORTED"})
)
terminal_success_states: frozenset = field(
default_factory=lambda: frozenset({"SUCCESS", "COMPLETED"})
)
@classmethod
def from_env(cls) -> "IDMCConfig":
return cls(
username=os.environ["IDMC_USERNAME"],
password=os.environ["IDMC_PASSWORD"],
login_url=os.environ.get("IDMC_LOGIN_URL", cls.login_url),
timeout_seconds=float(os.environ.get("IDMC_TIMEOUT_SECONDS", 3600)),
)
The login URL depends on your region (dm-us, dm-em, dm-ap, and so on), so it's configurable.
Step 2: A session that renews itself
An IDMC session times out after a period of inactivity. Long-running taskflows can outlast it, so the client has to notice a 401 and log in again without failing the run.
The v3 login response returns a sessionId and a baseApiUrl for your pod. Application Integration endpoints (/active-bpel/...) live on the same pod host, without the /saas suffix.
# idmc_monitor/session.py
import logging
from urllib.parse import urlparse
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from .config import IDMCConfig
log = logging.getLogger(__name__)
class IDMCAuthError(RuntimeError):
pass
class IDMCSession:
"""Wraps requests.Session with IDMC login and automatic re-login."""
def __init__(self, config: IDMCConfig):
self.config = config
self._http = requests.Session()
retry = Retry(
total=5,
backoff_factor=1.5,
status_forcelist=(429, 500, 502, 503, 504),
allowed_methods=frozenset({"GET", "POST"}),
)
self._http.mount("https://", HTTPAdapter(max_retries=retry))
self.session_id: str | None = None
self.base_api_url: str | None = None
self.pod_url: str | None = None
def login(self) -> None:
resp = self._http.post(
self.config.login_url,
json={"username": self.config.username, "password": self.config.password},
timeout=self.config.request_timeout,
)
if resp.status_code != 200:
raise IDMCAuthError(f"Login failed: HTTP {resp.status_code}")
body = resp.json()
self.session_id = body["userInfo"]["sessionId"]
self.base_api_url = body["products"][0]["baseApiUrl"]
parsed = urlparse(self.base_api_url)
self.pod_url = f"{parsed.scheme}://{parsed.netloc}"
log.info("idmc_login_ok", extra={"pod": self.pod_url})
def request(self, method: str, url: str, **kwargs) -> requests.Response:
if not self.session_id:
self.login()
for attempt in (1, 2):
headers = kwargs.pop("headers", {}) | {
"INFA-SESSION-ID": self.session_id,
"icSessionId": self.session_id,
"Accept": "application/json",
}
resp = self._http.request(
method, url, headers=headers,
timeout=self.config.request_timeout, **kwargs,
)
if resp.status_code == 401 and attempt == 1:
log.warning("idmc_session_expired_relogin")
self.login()
continue
resp.raise_for_status()
return resp
raise IDMCAuthError("Request still unauthorized after re-login")
Two layers of resilience run here. urllib3.Retry handles transient network errors and 5xx/429 responses. The explicit 401 branch handles session expiry, which a generic retry can't fix.
Step 3: The monitor
This is the core of the framework. It starts the taskflow, then polls until one of three things happens: success, failure, or timeout.
# idmc_monitor/monitor.py
import logging
import random
import time
from dataclasses import dataclass, field
from enum import Enum
from .config import IDMCConfig
from .notifiers import Notifier
from .session import IDMCSession
log = logging.getLogger(__name__)
class Outcome(str, Enum):
SUCCESS = "SUCCESS"
FAILED = "FAILED"
TIMEOUT = "TIMEOUT"
@dataclass
class RunResult:
taskflow: str
run_id: str
outcome: Outcome
final_status: str
duration_seconds: float
details: dict = field(default_factory=dict)
class TaskflowMonitor:
def __init__(self, session: IDMCSession, config: IDMCConfig,
notifiers: list[Notifier] | None = None):
self.session = session
self.config = config
self.notifiers = notifiers or []
# -- API calls -----------------------------------------------------
def start(self, taskflow: str, params: dict | None = None) -> str:
if not self.session.pod_url:
self.session.login()
url = f"{self.session.pod_url}/active-bpel/rt/{taskflow}"
resp = self.session.request("POST", url, json=params or {})
run_id = str(resp.json()["RunId"])
log.info("taskflow_started", extra={"taskflow": taskflow, "run_id": run_id})
return run_id
def get_status(self, run_id: str) -> dict:
url = f"{self.session.pod_url}/active-bpel/services/tf/status/{run_id}"
return self.session.request("GET", url).json()
# -- Polling loop --------------------------------------------------
def _next_delay(self, current: float) -> float:
"""Exponential backoff with full jitter, capped at poll_max_seconds."""
grown = min(current * 2, self.config.poll_max_seconds)
return random.uniform(self.config.poll_initial_seconds, grown)
def wait(self, taskflow: str, run_id: str) -> RunResult:
started = time.monotonic()
delay = self.config.poll_initial_seconds
last_status = "UNKNOWN"
payload: dict = {}
while True:
elapsed = time.monotonic() - started
if elapsed > self.config.timeout_seconds:
return self._finish(taskflow, run_id, Outcome.TIMEOUT,
last_status, elapsed, payload)
payload = self.get_status(run_id)
last_status = str(payload.get("status", "UNKNOWN")).upper()
log.info("taskflow_status", extra={
"taskflow": taskflow, "run_id": run_id,
"status": last_status, "elapsed": round(elapsed, 1),
})
if last_status in self.config.terminal_success_states:
return self._finish(taskflow, run_id, Outcome.SUCCESS,
last_status, elapsed, payload)
if last_status in self.config.terminal_failure_states:
return self._finish(taskflow, run_id, Outcome.FAILED,
last_status, elapsed, payload)
time.sleep(delay)
delay = self._next_delay(delay)
def run(self, taskflow: str, params: dict | None = None) -> RunResult:
run_id = self.start(taskflow, params)
return self.wait(taskflow, run_id)
# -- Outcome handling ----------------------------------------------
def _finish(self, taskflow, run_id, outcome, status, elapsed, payload) -> RunResult:
result = RunResult(taskflow, run_id, outcome, status, round(elapsed, 1), payload)
log.info("taskflow_finished", extra={
"taskflow": taskflow, "run_id": run_id,
"outcome": outcome.value, "status": status,
"duration_seconds": result.duration_seconds,
})
if outcome is not Outcome.SUCCESS:
for notifier in self.notifiers:
try:
notifier.send(result)
except Exception: # an alert failure must never mask the run result
log.exception("notifier_failed", extra={"notifier": type(notifier).__name__})
return result
A few design choices are worth explaining.
Why jittered backoff? If you run 40 taskflows at once and they all poll every 10 seconds, they hit the API in lockstep. Jitter spreads the calls out and keeps you clear of rate limits. The cap at poll_max_seconds keeps a long run from going unchecked for too long.
Why is TIMEOUT its own outcome? A timeout doesn't mean the taskflow failed. It may still be running. Keeping it separate lets on-call engineers decide whether to wait, cancel, or escalate, instead of treating every timeout as a failure.
Why catch notifier exceptions? If Slack is down, you still want the correct RunResult returned and logged. Alerting is best-effort. The run result is the source of truth.
Step 4: Pluggable notifiers
Each channel implements one method. Adding Teams, PagerDuty, or email means adding one small class.
# idmc_monitor/notifiers.py
import json
from typing import Protocol
import requests
class Notifier(Protocol):
def send(self, result) -> None: ...
def _summary(result) -> str:
return (f"IDMC taskflow *{result.taskflow}* ended as *{result.outcome.value}* "
f"(status={result.final_status}, run_id={result.run_id}, "
f"{result.duration_seconds}s)")
class SlackNotifier:
def __init__(self, webhook_url: str):
self.webhook_url = webhook_url
def send(self, result) -> None:
requests.post(self.webhook_url, json={"text": _summary(result)}, timeout=10)
class SNSNotifier:
def __init__(self, topic_arn: str, client=None):
import boto3
self.topic_arn = topic_arn
self.client = client or boto3.client("sns")
def send(self, result) -> None:
self.client.publish(
TopicArn=self.topic_arn,
Subject=f"IDMC {result.outcome.value}: {result.taskflow}"[:100],
Message=json.dumps({
"taskflow": result.taskflow,
"run_id": result.run_id,
"outcome": result.outcome.value,
"status": result.final_status,
"duration_seconds": result.duration_seconds,
}),
)
SNS fans out to email, SMS, Lambda, or an SQS queue, so one notifier covers a lot of ground in AWS.
Step 5: Structured logs
Plain-text logs are hard to query. A short JSON formatter turns every extra={...} field into a searchable key.
# idmc_monitor/logging_setup.py
import json
import logging
_STANDARD = set(vars(logging.makeLogRecord({})))
class JsonFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
data = {"level": record.levelname, "event": record.getMessage(),
"logger": record.name, "ts": self.formatTime(record)}
data.update({k: v for k, v in vars(record).items() if k not in _STANDARD})
return json.dumps(data, default=str)
def configure(level: int = logging.INFO) -> None:
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logging.basicConfig(level=level, handlers=[handler], force=True)
Now a CloudWatch Logs Insights query like this works out of the box:
fields ts, taskflow, outcome, duration_seconds
| filter event = "taskflow_finished"
| stats avg(duration_seconds), count(*) by taskflow, outcome
That one query gives you average run time and failure counts for each taskflow.
Step 6: A CLI entry point
# idmc_monitor/cli.py
import argparse
import json
import os
import sys
from .config import IDMCConfig
from .logging_setup import configure
from .monitor import Outcome, TaskflowMonitor
from .notifiers import SlackNotifier
from .session import IDMCSession
def main() -> int:
parser = argparse.ArgumentParser(description="Run and monitor an IDMC taskflow")
parser.add_argument("taskflow", help="Published taskflow API name")
parser.add_argument("--params", default="{}", help="JSON input parameters")
args = parser.parse_args()
configure()
config = IDMCConfig.from_env()
notifiers = []
if hook := os.environ.get("SLACK_WEBHOOK_URL"):
notifiers.append(SlackNotifier(hook))
monitor = TaskflowMonitor(IDMCSession(config), config, notifiers)
result = monitor.run(args.taskflow, json.loads(args.params))
exit_codes = {Outcome.SUCCESS: 0, Outcome.FAILED: 1, Outcome.TIMEOUT: 2}
return exit_codes[result.outcome]
if __name__ == "__main__":
sys.exit(main())
Different exit codes for failure and timeout let schedulers like Airflow, Control-M, or a CI pipeline react to each one differently.
export IDMC_USERNAME=svc_integration
export IDMC_PASSWORD='********'
python -m idmc_monitor.cli tf_daily_gl_load --params '{"run_date": "2026-11-10"}'
Step 7: Test it without touching IDMC
Because the session is injected, you can swap it for a fake and test the polling logic in milliseconds.
# tests/test_monitor.py
from unittest.mock import MagicMock, patch
from idmc_monitor.config import IDMCConfig
from idmc_monitor.monitor import Outcome, TaskflowMonitor
def make_monitor(statuses, notifier=None, timeout=60):
cfg = IDMCConfig(username="u", password="p", poll_initial_seconds=0.01,
poll_max_seconds=0.02, timeout_seconds=timeout)
session = MagicMock(pod_url="https://pod.example.com")
responses = iter(statuses)
session.request.side_effect = lambda m, url, **kw: MagicMock(
json=lambda: {"RunId": "42"} if m == "POST" else {"status": next(responses)}
)
return TaskflowMonitor(session, cfg, [notifier] if notifier else [])
@patch("time.sleep", return_value=None)
def test_success_after_running(_):
result = make_monitor(["RUNNING", "RUNNING", "SUCCESS"]).run("tf_demo")
assert result.outcome is Outcome.SUCCESS
assert result.run_id == "42"
@patch("time.sleep", return_value=None)
def test_failure_triggers_notifier(_):
notifier = MagicMock()
result = make_monitor(["RUNNING", "FAILED"], notifier).run("tf_demo")
assert result.outcome is Outcome.FAILED
notifier.send.assert_called_once()
@patch("time.sleep", return_value=None)
def test_notifier_error_does_not_mask_result(_):
notifier = MagicMock()
notifier.send.side_effect = RuntimeError("slack down")
result = make_monitor(["SUSPENDED"], notifier).run("tf_demo")
assert result.outcome is Outcome.FAILED
Run them with pytest -q. A timeout test works the same way: set timeout_seconds=0 and feed it endless RUNNING statuses.
Running it on AWS
The framework has no AWS dependency at its core, so you can choose where it runs:
AWS Lambda: Good for taskflows that finish within the 15-minute Lambda limit. Package the module as a Lambda layer and call monitor.run() from the handler.
AWS Step Functions: Better for long runs. Use one Lambda to call start() and another to call get_status() once. Then let a Step Functions Wait + Choice loop do the polling. You pay nothing while waiting, and there's no timeout ceiling.
ECS Fargate or a scheduler container: Good when the monitor is one step in a bigger batch job.
For Step Functions, split the monitor like this:
def start_handler(event, _ctx):
run_id = monitor.start(event["taskflow"], event.get("params"))
return {**event, "run_id": run_id}
def status_handler(event, _ctx):
status = str(monitor.get_status(event["run_id"]).get("status", "")).upper()
return {**event, "status": status}
The state machine checks status and either loops back through a Wait state or branches to success or failure.
Production checklist
Before you rely on this in production:
☐ Store credentials in AWS Secrets Manager or Parameter Store, never in environment files committed to Git.
☐ Use a dedicated IDMC service account with only the permissions the taskflows need.
☐ Confirm the status values your pod returns and adjust terminal_*_states to match.
☐ Set timeout_seconds per taskflow based on historical run times. A 95th-percentile duration plus a buffer works well.
☐ Add a CloudWatch alarm on the taskflow_finished metric filter for outcome != SUCCESS, as a backstop in case a notifier fails.
☐ Make your taskflows idempotent so a retry after a timeout doesn't double-load data.
Wrapping up
Starting a taskflow is one API call. Trusting it in production takes a bit more: a session that renews itself, polling that respects rate limits, clear outcomes, and alerts that can't hide the real result.
With this framework, every IDMC run either reports success or tells someone exactly what went wrong, with the run ID, the final status, and how long it took.
What's next: In a follow-up post, I'll pull IDMC activity-log data into a dashboard so you can see run-time trends and failure hotspots across every taskflow in your org.
If you're running IDMC at scale, I'd like to hear how you handle monitoring today. Share your approach in the comments.
Originally published by Dev.to AI. Aggregated on AIWithGhost for educational purposes — full credit and traffic to the original publisher.