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 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.