REP-1561-feat(export-jobs): add create, list, get, and wait commands

.gitignore

@@ -8,3 +8,5 @@ build/
 *.spec
 .idea
 .tokensave
+.tokensave/*
+.env

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## [Unreleased]
+
+- Add `export-jobs` commands (`create`, `list`, `get`, `wait`) for on-demand destination data refreshes via the public API
+
 ## [0.1.18] - 2026-09-01
 
 

README.md

@@ -72,6 +72,16 @@ export X_API_KEY=your_api_key
 dualentry bills list --status posted --format json
 ```
 
+### Request a destination data export and wait for it
+
+```bash
+# Kick off an off-cycle Snowflake (or other destination) refresh
+dualentry export-jobs create --format json
+
+# Poll until completed or failed (exit 1 on failure / timeout)
+dualentry export-jobs wait 42 --interval 10 --timeout 1800
+```
+
 ## Available Resources
 
 | Category | Resources |
@@ -83,8 +93,10 @@ dualentry bills list --status posted --format json
 | **Automation** | Recurring Invoices, Recurring Bills, Workflows, Contracts |
 | **Close Management** | Bank Match |
 | **Bank Feeds** | Bank Connections |
+| **Developers** | Export Jobs (destination data refresh) |
 
 Most resources support `list`, `get`, `create`, and `update`; some (e.g. Bank Connections, Bank Match) expose a different verb set — see `--help`.
+Export jobs support `create`, `list`, `get`, and `wait` (poll until finished).
 
 ## Output Formats
 

src/dualentry_cli/commands/export_jobs.py

@@ -0,0 +1,151 @@
+"""On-demand destination data export jobs (Snowflake / warehouse refresh)."""
+
+from __future__ import annotations
+
+import time
+from enum import StrEnum
+
+import typer
+
+from dualentry_cli.commands import AllPages, Format, Limit, Offset
+from dualentry_cli.commands.actions import make_action_app, run_get, run_list
+from dualentry_cli.output import format_output
+
+_TERMINAL = frozenset({"completed", "failed"})
+
+
+class ExportJobStatus(StrEnum):
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+
+app = make_action_app("Manage Snowflake / warehouse export jobs")
+
+
+@app.command("create")
+def create_export_job(
+    integration_id: int | None = typer.Option(
+        None,
+        "--integration-id",
+        help="Destination integration ID. Optional when the org has exactly one connected destination.",
+    ),
+    output: str = Format,
+):
+    """
+    Request an off-cycle data export refresh.
+
+    DualEntry accepts the request and starts the export when capacity allows.
+    Calling this while an export is already running returns that job with
+    deduplicated=true instead of starting a second one.
+    """
+    from dualentry_cli.main import get_client
+
+    body: dict = {}
+    if integration_id is not None:
+        body["integration_id"] = integration_id
+
+    data = get_client().post("/export-jobs/", json=body or None)
+
+    if output == "json":
+        format_output(data, resource="export-job", fmt="json")
+        return
+
+    if data.get("deduplicated"):
+        typer.secho(
+            "An export was already running; returning that job (deduplicated).",
+            fg=typer.colors.YELLOW,
+            err=True,
+        )
+    job = data.get("job") or data
+    format_output(job, resource="export-job", fmt=output)
+
+
+@app.command("list")
+def list_export_jobs(
+    limit: int = Limit,
+    offset: int = Offset,
+    all_pages: bool = AllPages,
+    status: list[ExportJobStatus] | None = typer.Option(
+        None,
+        "--status",
+        help="Filter by status: pending, running, completed, or failed. Repeatable.",
+    ),
+    output: str = Format,
+):
+    """List destination data export jobs, most recent first."""
+    statuses = [s.value for s in status] if status else None
+    run_list(
+        "export-jobs",
+        resource="export-job",
+        limit=limit,
+        offset=offset,
+        all_pages=all_pages,
+        output=output,
+        status=statuses,
+        status_param="status",
+    )
+
+
+@app.command("get")
+def get_export_job(
+    job_id: int = typer.Argument(help="Export job ID from create or list"),
+    output: str = Format,
+):
+    """Get one export job by ID. Poll this after create."""
+    run_get(f"/export-jobs/{job_id}/", resource="export-job", output=output)
+
+
+@app.command("wait")
+def wait_export_job(
+    job_id: int = typer.Argument(help="Export job ID to poll until finished"),
+    interval: float = typer.Option(
+        5.0,
+        "--interval",
+        min=0.5,
+        help="Seconds between status polls",
+    ),
+    timeout: float = typer.Option(
+        1800.0,
+        "--timeout",
+        min=1.0,
+        help="Give up after this many seconds (default 30 minutes)",
+    ),
+    output: str = Format,
+):
+    """
+    Poll an export job until it completes or fails.
+
+    Exit 0 when status is completed, 1 when failed or the timeout elapses.
+    Useful in CI after create.
+    """
+    from dualentry_cli.main import get_client
+
+    client = get_client()
+    deadline = time.monotonic() + timeout
+    job: dict = {}
+
+    while True:
+        job = client.get(f"/export-jobs/{job_id}/")
+        status = job.get("status")
+        if status in _TERMINAL:
+            break
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            typer.secho(
+                f"Timed out after {timeout:.0f}s waiting for export job {job_id} (last status: {status}).",
+                fg=typer.colors.RED,
+                err=True,
+            )
+            format_output(job, resource="export-job", fmt=output)
+            raise typer.Exit(code=1)
+        typer.secho(
+            f"export-jobs wait: job {job_id} is {status}; next poll in {min(interval, remaining):.0f}s ({remaining:.0f}s left)",
+            err=True,
+        )
+        time.sleep(min(interval, remaining))
+
+    format_output(job, resource="export-job", fmt=output)
+    if job.get("status") == "failed":
+        raise typer.Exit(code=1)

src/dualentry_cli/main.py

@@ -10,6 +10,7 @@
 from dualentry_cli.commands.accounts import app as accounts_app
 from dualentry_cli.commands.bank_connections import app as bank_connections_app
 from dualentry_cli.commands.bank_match import app as bank_match_app
+from dualentry_cli.commands.export_jobs import app as export_jobs_app
 from dualentry_cli.commands.ije_extras import IJE_CHECKS, IJE_ONLINE_EXTRA_CHECKS, IJE_TEMPLATE
 from dualentry_cli.config import Config
 
@@ -109,6 +110,7 @@
 app.add_typer(make_resource_app("inbox items", "inbox-item", "inbox", has_get=False, has_create=False, has_update=False, filters={"search"}), name="inbox")
 app.add_typer(bank_match_app, name="bank-match")
 app.add_typer(bank_connections_app, name="bank-connections")
+app.add_typer(export_jobs_app, name="export-jobs")
 
 
 def version_callback(value: bool):

src/dualentry_cli/output.py

@@ -1192,6 +1192,63 @@ def _workflow_detail(r):
 _register("workflow", _workflow_list, _workflow_detail)
 
 
+# ── Export jobs (destination data refresh) ───────────────────────────
+
+
+def _export_job_status_style(status: str | None) -> str:
+    return {
+        "pending": "yellow",
+        "running": "cyan",
+        "completed": "green",
+        "failed": "red",
+    }.get(status or "", "white")
+
+
+def _export_job_list(items: list[dict]) -> None:
+    table = Table(title="Export Jobs", show_header=True, header_style="bold")
+    table.add_column("ID", style="cyan")
+    table.add_column("Status")
+    table.add_column("Integration")
+    table.add_column("Created")
+    table.add_column("Started")
+    table.add_column("Completed")
+    for item in items:
+        status = item.get("status") or "-"
+        table.add_row(
+            str(item.get("id", "-")),
+            Text(str(status), style=_export_job_status_style(status if isinstance(status, str) else None)),
+            str(item.get("integration_id") or "-"),
+            str(item.get("created_at") or "-")[:19],
+            str(item.get("started_at") or "-")[:19],
+            str(item.get("completed_at") or "-")[:19],
+        )
+    console.print(table)
+
+
+def _export_job_detail(r: dict) -> None:
+    status = r.get("status") or "-"
+    header = Text()
+    header.append("Export Job ", style="bold")
+    header.append(str(r.get("id", "-")), style="bold cyan")
+    header.append("  ")
+    header.append(str(status), style=_export_job_status_style(status if isinstance(status, str) else None))
+    console.print(Panel(header, expand=False))
+
+    details = Table(show_header=False, box=None, padding=(0, 1))
+    details.add_column(style="dim", min_width=16)
+    details.add_column()
+    details.add_row("Integration:", str(r.get("integration_id") or "-"))
+    details.add_row("Created:", str(r.get("created_at") or "-"))
+    details.add_row("Started:", str(r.get("started_at") or "-"))
+    details.add_row("Completed:", str(r.get("completed_at") or "-"))
+    if r.get("error"):
+        details.add_row("Error:", Text(str(r["error"]), style="red"))
+    console.print(details)
+
+
+_register("export-job", _export_job_list, _export_job_detail)
+
+
 # ── Recurring records (use generic transaction pattern) ──────────────
 
 for _prefix in ("recurring-invoice", "recurring-bill", "recurring-journal-entry"):

tests/conftest.py

@@ -0,0 +1,23 @@
+"""Shared pytest fixtures for dualentry-cli tests."""
+
+from __future__ import annotations
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def disable_cli_update_checks(request, monkeypatch):
+    """
+    Keep update notices out of CliRunner output.
+
+    ``check_for_updates`` prints to stderr; Click's CliRunner folds that into
+    ``result.output``, which breaks tests that ``json.loads`` the whole stream.
+    Once a newer tag exists than ``__version__``, a background cache refresh
+    makes that flaky across the suite. Skip for ``test_updater``, which covers
+    the notice itself.
+    """
+    if request.module.__name__.endswith("test_updater"):
+        yield
+        return
+    monkeypatch.setattr("dualentry_cli.updater.check_for_updates", lambda: None)
+    yield