feat(sync): add --force-missing-dependencies to import command

[michael-richey]

Summary

• Adds _discover_missing_dependencies() to ResourcesHandler: scans imported source state via _resource_connections() and returns the set of (resource_type, id) pairs not yet in source state
• Adds _import_missing_dep_cb(): async worker callback that imports a single dep by ID from the source API, emits success/skip/failure events, then recursively discovers and enqueues transitive deps
• Modifies import_resources() to run the discovery + worker loop between import_resources_without_saving() and dump_state(Origin.SOURCE) when force_missing_dependencies=True
• Wires --force-missing-dependencies into the integration test helper's import_resources() method
• 27 new unit tests (TDD: RED→GREEN cycles A–D covering discovery, callback, wiring, and integration helper)

Stacked on: #549

Use case

In split import/sync workflows (import on machine A, sync on machine B), this flag lets import produce a fully self-contained local state. sync can then run without any source API access for missing deps.

datadog-sync import \
  --resources="dashboard_lists" \
  --filter='Type=dashboard_lists;Name=id;Value=^539651$' \
  --force-missing-dependencies
# source/ now contains dashboard_lists + dashboards + monitors + ...

datadog-sync sync --resources="dashboard_lists"
# no source API calls for missing deps

Test plan

• pytest tests/unit/test_import_force_missing_deps.py -v — 27 tests, all green
• pytest tests/unit/ -v — 363 tests, full regression
• tox -e ruff,black — lint clean
• Manual smoke test: import with --force-missing-dependencies, verify transitive dep types appear in resources/source/

🤖 Generated with Claude Code

[michael-richey]

Review notes from my pass:

P2: import dependency discovery is still affected by destination state

ResourcesHandler._discover_missing_dependencies() reuses _resource_connections(), which calls each resource class connect_id() path. That path is sync-oriented and checks config.state.destination when deciding whether a referenced dependency is unresolved. During import, this means a stale or existing destination mapping can make a source dependency look complete even when it was never fetched into state.source. Example: if state.destination["dashboards"]["dash-1"] exists, a dashboard list that references dash-1 will not enqueue dashboards/dash-1 for import, so the local source state is not self-contained.

Suggested fix: add an import-specific dependency extractor that compares referenced IDs only against config.state.source, and use that from _discover_missing_dependencies() and _import_missing_dep_cb() instead of _resource_connections().

def _missing_source_dependencies_for_resource(self, resource_type: str, _id: str) -> set[tuple[str, str]]:
    missing = set()
    resource = deepcopy(self.config.state.source[resource_type][_id])
    r_class = self.config.resources[resource_type]

    def source_connect_id(key: str, r_obj: dict, dep_type: str) -> list[str]:
        values = r_obj[key] if isinstance(r_obj[key], list) else [r_obj[key]]
        return [str(value) for value in values if str(value) not in self.config.state.source[dep_type]]

    for dep_type, paths in r_class.resource_config.resource_connections.items():
        for path in paths:
            failed = find_attr(path, dep_type, resource, source_connect_id)
            for dep_id in failed or []:
                missing.add((dep_type, dep_id))

    return missing

P2: dependency discovery does not compute closure through already-present source deps

_discover_missing_dependencies() only scans resources_arg, then _import_missing_dep_cb() scans newly imported missing deps. If a direct dependency is already present in state.source from an earlier partial import, it is treated as complete and its own transitive dependencies are never scanned. Example: importing only dashboard_lists --force-missing-dependencies where dashboards/dash-1 already exists locally but references missing monitor 123 leaves monitors/123 absent.

Suggested fix: make discovery a source-state closure walk. Start from the imported resources_arg nodes, scan every reachable node once, enqueue already-present dependencies for scanning, and collect dependencies that must be fetched. After each fetched dependency, scan that node with the same helper.

def _discover_missing_dependencies(self) -> set[tuple[str, str]]:
    missing = set()
    seen = set()
    queue = [
        (resource_type, _id)
        for resource_type in self.config.resources_arg
        for _id in self.config.state.source[resource_type].keys()
    ]

    while queue:
        resource_type, _id = queue.pop()
        node = (resource_type, _id)
        if node in seen:
            continue
        seen.add(node)

        r_class = self.config.resources[resource_type]
        if not r_class.resource_config.resource_connections:
            continue

        for dep_type, dep_id in self._dependencies_for_source_resource(resource_type, _id):
            dep = (dep_type, dep_id)
            if dep_id in self.config.state.source[dep_type]:
                queue.append(dep)
            else:
                missing.add(dep)

    return missing

The key point is that import-time resolution should be source-state closure logic only. Sync-time _resource_connections() can stay destination-oriented because it also performs ID remapping for apply.

The base branch was changed.