airbyte_cdk.sources.utils.check_section

  1#
  2# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
  3#
  4"""
  5Section-aware CHECK for the stepped setup form.
  6
  7A connector whose specification declares `sections` (each listing the `groups` of properties it shows) can be checked
  8one section at a time. The form order is the webapp's: sections with a numeric `order` come first, sorted by it, then
  9the others in list order. A section which declares `"check": true` tells the platform to run a CHECK when
 10the user completes it; the configuration JSON then carries a top-level `check-section` key naming the section, and only
 11holds the fields of that section and of the sections before it. The entrypoint validates that much of the configuration
 12against the specification, runs the full check for the last section (the configuration is complete by then) and
 13delegates every other section to the connector's `check_section`. A section without `"check": true` is never checked
 14on its own; the last one is covered by the full check the platform runs before creating the source. Without the key,
 15CHECK validates the complete configuration, exactly as before.
 16"""
 17
 18from __future__ import annotations
 19
 20import copy
 21from typing import Any, Dict, List, Mapping, Optional, Set, Tuple
 22
 23from airbyte_cdk.models import ConnectorSpecification, FailureType
 24from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 25
 26CHECK_SECTION_KEY = "check-section"
 27"""The optional top-level key of the configuration JSON object which turns a CHECK into a section check."""
 28
 29
 30def _form_position(section: Mapping[str, Any]) -> Tuple[int, float]:
 31    """Sections with a numeric `order` come first, sorted by it; the others keep their list order after them."""
 32    order = section.get("order")
 33    if isinstance(order, (int, float)) and not isinstance(order, bool):
 34        return (0, order)
 35    return (1, 0)
 36
 37
 38def _sections(spec: ConnectorSpecification) -> List[Mapping[str, Any]]:
 39    """
 40    The `sections` of the specification which have a string `id`, in form order: the webapp shows the sections with a
 41    numeric `order` first, sorted by it, then the others in list order, so the list position alone is not the order.
 42    """
 43    return sorted(
 44        (
 45            section
 46            for section in spec.connectionSpecification.get("sections") or []
 47            if isinstance(section, Mapping) and isinstance(section.get("id"), str)
 48        ),
 49        key=_form_position,
 50    )
 51
 52
 53def declared_sections(spec: ConnectorSpecification) -> List[str]:
 54    """The `sections[].id` values of the specification, in form order; empty when the specification declares none."""
 55    return [section["id"] for section in _sections(spec)]
 56
 57
 58def checkable_sections(spec: ConnectorSpecification) -> List[str]:
 59    """The ids of the sections which declare `"check": true`, the only ones a `check-section` may name."""
 60    return [section["id"] for section in _sections(spec) if section.get("check") is True]
 61
 62
 63def requested_section(config: Mapping[str, Any], spec: ConnectorSpecification) -> Optional[str]:
 64    """
 65    The section named by `check-section` in the config, or None for a full check.
 66
 67    Raises a config error when the key is present but not a non-empty string (a null value included), when the
 68    specification declares no sections, when the section is not one of the declared ones, or when the section does not
 69    declare `"check": true`.
 70    """
 71    if CHECK_SECTION_KEY not in config:
 72        return None
 73    section = config[CHECK_SECTION_KEY]
 74    declared = declared_sections(spec)
 75    if not isinstance(section, str) or not section.strip():
 76        message = f"'{CHECK_SECTION_KEY}' must be a non-empty string."
 77    elif not declared:
 78        message = f"'{CHECK_SECTION_KEY}' is not supported: the connector specification declares no sections."
 79    elif section not in declared:
 80        message = (
 81            f"Unknown {CHECK_SECTION_KEY} '{section}'; "
 82            f"the connector specification declares: {', '.join(declared)}."
 83        )
 84    elif section not in checkable_sections(spec):
 85        message = (
 86            f"Section '{section}' has no check: "
 87            f'the connector specification does not declare "check": true for it.'
 88        )
 89    else:
 90        return section
 91    raise AirbyteTracedException(
 92        internal_message=message, message=message, failure_type=FailureType.config_error
 93    )
 94
 95
 96def is_last_section(spec: ConnectorSpecification, section: str) -> bool:
 97    """Whether `section` is the last one declared, which completes the configuration."""
 98    declared = declared_sections(spec)
 99    return bool(declared) and section == declared[-1]
100
101
102def strip_check_section(config: Mapping[str, Any]) -> Dict[str, Any]:
103    """The config without the `check-section` key, which is not a connector field."""
104    return {key: value for key, value in config.items() if key != CHECK_SECTION_KEY}
105
106
107CROSS_PROPERTY_KEYWORDS = (
108    "allOf",
109    "anyOf",
110    "oneOf",
111    "not",
112    "if",
113    "then",
114    "else",
115    "dependencies",
116    "dependentRequired",
117    "dependentSchemas",
118)
119"""
120Root-level JSON Schema keywords which can require properties of later sections (for instance a `dependentRequired`
121tying a connection field to a credentials field). A section check drops them, see `section_spec`.
122"""
123
124
125def section_spec(spec: ConnectorSpecification, section: str) -> ConnectorSpecification:
126    """
127    The specification narrowed to what the form has collected by `section`:
128    - `required` only keeps the properties whose `group` belongs to that section or to an earlier one;
129    - the root-level cross-property keywords (`CROSS_PROPERTY_KEYWORDS`) are dropped, since they can require
130      properties of later sections.
131    Every property keeps its own schema, so the fields which are present are still validated in full, and the dropped
132    constraints apply again in the full check. The last section completes the configuration, so its specification is
133    the full one.
134    """
135    if is_last_section(spec, section):
136        return spec
137    schema = copy.deepcopy(spec.connectionSpecification)
138    for keyword in CROSS_PROPERTY_KEYWORDS:
139        schema.pop(keyword, None)
140    collected_groups: Set[str] = set()
141    for declared_section in _sections(spec):
142        collected_groups.update(declared_section.get("groups") or [])
143        if declared_section["id"] == section:
144            break
145    properties = schema.get("properties") or {}
146    schema["required"] = [
147        name
148        for name in schema.get("required") or []
149        if isinstance(properties.get(name), Mapping)
150        and properties[name].get("group") in collected_groups
151    ]
152    narrowed = copy.copy(spec)
153    narrowed.connectionSpecification = schema
154    return narrowed