feat: normalize state and catalog to Airbyte protocol format by default (#1017)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: patrick.nilan@airbyte.io <patrick.nilan@airbyte.io>

Author: devin-ai-integration[bot]

airbyte/cloud/_case_conversion.py

@@ -0,0 +1,55 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+"""Generic camelCase ↔ snake_case key conversion helpers.
+
+These helpers convert dict keys between camelCase (Config API format)
+and snake_case (Airbyte protocol format). Values are left unchanged;
+conversion is shallow (one level of dict keys) so opaque payloads
+like `streamState` blobs and `jsonSchema` dicts are never modified.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+
+_CAMEL_BOUNDARY = re.compile(r"([A-Z])")
+
+
+def _camel_to_snake(key: str) -> str:
+    """Convert a single camelCase key to snake_case.
+
+    Examples: `streamDescriptor` → `stream_descriptor`,
+    `jsonSchema` → `json_schema`, `name` → `name`.
+    """
+    return _CAMEL_BOUNDARY.sub(r"_\1", key).lower().lstrip("_")
+
+
+def _snake_to_camel(key: str) -> str:
+    """Convert a single snake_case key to camelCase.
+
+    Examples: `stream_descriptor` → `streamDescriptor`,
+    `json_schema` → `jsonSchema`, `name` → `name`.
+    """
+    parts = key.split("_")
+    return parts[0] + "".join(p.capitalize() for p in parts[1:])
+
+
+def camel_to_snake_keys(data: dict[str, Any]) -> dict[str, Any]:
+    """Shallow-convert all dict keys from camelCase to snake_case.
+
+    Only the immediate keys of `data` are converted; nested dicts and
+    lists inside values are returned as-is. This prevents accidental
+    mutation of opaque payloads (connector state blobs, JSON Schema, etc.).
+    """
+    return {_camel_to_snake(k): v for k, v in data.items()}
+
+
+def snake_to_camel_keys(data: dict[str, Any]) -> dict[str, Any]:
+    """Shallow-convert all dict keys from snake_case to camelCase.
+
+    Only the immediate keys of `data` are converted; nested dicts and
+    lists inside values are returned as-is. This prevents accidental
+    mutation of opaque payloads (connector state blobs, JSON Schema, etc.).
+    """
+    return {_snake_to_camel(k): v for k, v in data.items()}

airbyte/cloud/_connection_catalog.py

@@ -0,0 +1,98 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+"""Normalization helpers for connection catalog format conversion.
+
+The Config API returns catalogs using camelCase keys (`syncCatalog` format).
+The Airbyte protocol uses snake_case keys (`ConfiguredAirbyteCatalog` format).
+These helpers convert between the two representations using generic
+camelCase ↔ snake_case key converters rather than hard-coded field maps.
+
+Structural differences (flattening/nesting the `config` block) are handled
+by thin wrappers around the generic converters.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from airbyte.cloud._case_conversion import camel_to_snake_keys, snake_to_camel_keys
+
+
+def _normalize_catalog_to_protocol(
+    sync_catalog: dict[str, Any],
+) -> dict[str, Any]:
+    """Convert a `syncCatalog` dict to `ConfiguredAirbyteCatalog` protocol format.
+
+    Transforms camelCase keys to snake_case and flattens the `config` block into
+    each stream entry, matching the structure expected by connector `--catalog` flags.
+
+    Args:
+        sync_catalog: Catalog dict from the Config API (camelCase keys, nested `config`).
+
+    Returns:
+        Catalog dict in Airbyte protocol format (snake_case keys, flat stream entries).
+    """
+    configured_streams: list[dict[str, Any]] = []
+
+    for stream_config in sync_catalog.get("streams", []):
+        stream_info = stream_config.get("stream", {})
+        config_info = stream_config.get("config", {})
+
+        # Convert stream-level keys from camelCase → snake_case (shallow).
+        # Opaque values like jsonSchema content are preserved as-is.
+        normalized_stream = camel_to_snake_keys(stream_info)
+
+        # Build the configured stream entry with flattened config keys.
+        configured_entry: dict[str, Any] = {"stream": normalized_stream}
+        configured_entry.update(camel_to_snake_keys(config_info))
+
+        configured_streams.append(configured_entry)
+
+    return {"streams": configured_streams}
+
+
+def _denormalize_catalog_to_api(
+    configured_catalog: dict[str, Any],
+) -> dict[str, Any]:
+    """Convert a `ConfiguredAirbyteCatalog` dict back to `syncCatalog` API format.
+
+    Reverses `_normalize_catalog_to_protocol`: converts snake_case keys to camelCase
+    and nests config fields back under a `config` block.
+
+    Args:
+        configured_catalog: Catalog dict in Airbyte protocol format.
+
+    Returns:
+        Catalog dict in Config API format (camelCase keys, nested `config`).
+    """
+    api_streams: list[dict[str, Any]] = []
+
+    for entry in configured_catalog.get("streams", []):
+        stream_info = entry.get("stream", {})
+
+        # Convert stream-level keys from snake_case → camelCase (shallow).
+        api_stream = snake_to_camel_keys(stream_info)
+
+        # Everything except "stream" is a config field — nest back under "config".
+        config = snake_to_camel_keys({k: v for k, v in entry.items() if k != "stream"})
+
+        api_streams.append({"stream": api_stream, "config": config})
+
+    return {"streams": api_streams}
+
+
+def _is_protocol_catalog_format(catalog: dict[str, Any]) -> bool:
+    """Detect whether a catalog dict is in Airbyte protocol format.
+
+    Checks the first stream entry for snake_case config keys (`sync_mode`)
+    at the top level, which indicates protocol format. API format nests
+    these under a `config` key with camelCase (`syncMode`).
+
+    Returns `True` for protocol format, `False` for Config API format.
+    """
+    streams = catalog.get("streams", [])
+    if not streams:
+        return False
+
+    first = streams[0]
+    # Protocol format has sync_mode at top level; API format has it inside config
+    return "sync_mode" in first or "destination_sync_mode" in first

airbyte/cloud/_connection_state.py

@@ -15,6 +15,8 @@
 
 from pydantic import BaseModel, ConfigDict, Field
 
+from airbyte.cloud._case_conversion import snake_to_camel_keys
+
 
 class StreamDescriptor(BaseModel):
     """Descriptor for a single stream within a connection state."""
@@ -107,6 +109,145 @@ def __str__(self) -> str:
         )
 
 
+def _normalize_state_to_protocol(
+    raw_state: dict[str, Any],
+) -> list[dict[str, Any]]:
+    """Convert a raw Config API state blob to Airbyte protocol state messages.
+
+    Transforms the camelCase API format into a list of `AirbyteStateMessage` dicts
+    with snake_case keys, suitable for passing to a connector's `--state` flag.
+
+    Args:
+        raw_state: Full state dict from the Config API, including `stateType`.
+
+    Returns:
+        List of protocol-format state message dicts. Empty list if state is `not_set`.
+    """
+    parsed = ConnectionStateResponse(**raw_state)
+
+    if parsed.state_type == "not_set":
+        return []
+
+    if parsed.state_type == "legacy":
+        return [{"type": "LEGACY", "data": parsed.state or {}}]
+
+    if parsed.state_type == "global" and parsed.global_state:
+        return [
+            {
+                "type": "GLOBAL",
+                "global": parsed.global_state.model_dump(by_alias=False),
+            }
+        ]
+
+    # state_type == "stream"
+    if not parsed.stream_state:
+        return []
+
+    return [
+        {
+            "type": "STREAM",
+            "stream": entry.model_dump(by_alias=False),
+        }
+        for entry in parsed.stream_state
+    ]
+
+
+def _stream_entry_to_api(stream_entry: dict[str, Any]) -> dict[str, Any]:
+    """Convert a single protocol-format stream entry to Config API format.
+
+    Converts `stream_descriptor` keys to camelCase (via generic helper) and
+    preserves the opaque `stream_state` blob as-is. Strips `None` values from
+    the descriptor (the API omits null fields like `namespace`).
+    """
+    descriptor = stream_entry.get("stream_descriptor", {})
+    api_descriptor = {k: v for k, v in snake_to_camel_keys(descriptor).items() if v is not None}
+    return {
+        "streamDescriptor": api_descriptor,
+        "streamState": stream_entry.get("stream_state"),
+    }
+
+
+def _denormalize_protocol_state_to_api(
+    protocol_messages: list[dict[str, Any]],
+    connection_id: str,
+) -> dict[str, Any]:
+    """Convert Airbyte protocol state messages back to Config API format.
+
+    Reverses `_normalize_state_to_protocol`, producing a dict suitable for
+    `import_raw_state()` / the Config API `create_or_update_safe` endpoint.
+    Uses generic snake_case → camelCase key conversion for stream entries;
+    the opaque state blobs are preserved as-is.
+
+    Args:
+        protocol_messages: List of protocol-format `AirbyteStateMessage` dicts.
+        connection_id: Connection ID to embed in the result.
+
+    Returns:
+        A Config API state dict with camelCase keys and `stateType`.
+    """
+    if not protocol_messages:
+        return {
+            "stateType": "not_set",
+            "connectionId": connection_id,
+        }
+
+    first = protocol_messages[0]
+    msg_type = first.get("type", "").upper()
+
+    if msg_type == "LEGACY":
+        return {
+            "stateType": "legacy",
+            "connectionId": connection_id,
+            "state": first.get("data", {}),
+        }
+
+    if msg_type == "GLOBAL":
+        global_body = first.get("global", {})
+        return {
+            "stateType": "global",
+            "connectionId": connection_id,
+            "globalState": {
+                "sharedState": global_body.get("shared_state"),
+                "streamStates": [
+                    _stream_entry_to_api(s) for s in global_body.get("stream_states", [])
+                ],
+            },
+        }
+
+    # STREAM type
+    return {
+        "stateType": "stream",
+        "connectionId": connection_id,
+        "streamState": [_stream_entry_to_api(s.get("stream", {})) for s in protocol_messages],
+    }
+
+
+def _is_protocol_state_format(
+    state_input: dict[str, Any] | list[dict[str, Any]],
+) -> bool:
+    """Detect whether the input is in Airbyte protocol format.
+
+    Returns `True` for protocol format:
+    - an empty list (representing no protocol state), or
+    - a non-empty list where every item is a dict with a known `type`, or
+    - a single dict with top-level snake_case `type` and no `stateType`.
+
+    Returns `False` for Config API format (for example, a dict with `stateType`
+    or a raw `streamState` list from the Config API).
+    """
+    protocol_message_types = {"STREAM", "GLOBAL", "LEGACY"}
+
+    if isinstance(state_input, list):
+        if not state_input:
+            return True
+        return all(
+            isinstance(message, dict) and message.get("type") in protocol_message_types
+            for message in state_input
+        )
+    # A dict with snake_case 'type' at top-level is a single protocol message
+    return "type" in state_input and "stateType" not in state_input
+
+
 def _match_stream(
     stream: StreamState,
     stream_name: str,