diff --git a/airbyte/_util/api_util.py b/airbyte/_util/api_util.py index bb6ebc4da..84cc80b31 100644 --- a/airbyte/_util/api_util.py +++ b/airbyte/_util/api_util.py @@ -26,7 +26,11 @@ from airbyte_api import api, models from airbyte_api.errors import SDKError -from airbyte._util.cloud_errors import CloudErrorInfo, parse_cloud_error +from airbyte._util.cloud_errors import ( + CloudErrorInfo, + parse_cloud_error, + parse_problem_payload, +) from airbyte._util.meta import ( AIRBYTE_CLOUD_ANALYTIC_SOURCE_HEADER, get_cloud_api_analytic_source, @@ -1989,6 +1993,80 @@ def build_connection_schedule( ) +_MAX_CONFLICTING_DESTINATION_STREAMS = 10 + + +def _format_conflicting_destination_stream(stream: object) -> str | None: + if not isinstance(stream, dict): + return None + name = stream.get("streamName") + namespace = stream.get("streamNamespace", "") + connection_ids = stream.get("connectionIds") + if not isinstance(name, str) or not name: + return None + if namespace is None: + namespace = "" + if not isinstance(namespace, str): + return None + if not isinstance(connection_ids, list) or not connection_ids: + return None + if any(not isinstance(connection_id, str) for connection_id in connection_ids): + return None + + stream_name = f"{namespace}.{name}" if namespace else name + rendered_ids = ", ".join(f"`{connection_id}`" for connection_id in connection_ids) + return f"`{stream_name}` (connection {rendered_ids})" + + +def _format_conflicting_destination_streams(body: str) -> str | None: + payload = parse_problem_payload(body) + if payload is None: + return None + data = payload.get("data") + streams = data.get("streams") if isinstance(data, dict) else None + if not isinstance(streams, list) or not streams: + return None + + formatted_streams: list[str] = [] + for stream in streams: + formatted_stream = _format_conflicting_destination_stream(stream) + if formatted_stream is not None: + formatted_streams.append(formatted_stream) + if not formatted_streams: + return None + + stream_list = ", ".join(formatted_streams[:_MAX_CONFLICTING_DESTINATION_STREAMS]) + if len(formatted_streams) > _MAX_CONFLICTING_DESTINATION_STREAMS: + remaining_stream_count = len(formatted_streams) - _MAX_CONFLICTING_DESTINATION_STREAMS + stream_list = f"{stream_list}, and {remaining_stream_count} more" + return stream_list + + +def _conflicting_destination_stream_guidance(error: SDKError) -> str | None: + """Explain how to avoid destination table collisions for a known Cloud problem.""" + body = error_response_body(error) + problem = parse_cloud_error(error.status_code, body) + if problem.slug != "error:connection-conflicting-destination-stream": + return None + + generic_guidance = ( + "Another connection on this destination may already write to the same tables. " + "Set `table_prefix` (for example `_`) so this connection writes to " + "different table names." + ) + if body is None: + return generic_guidance + stream_list = _format_conflicting_destination_streams(body) + if stream_list is None: + return generic_guidance + + return ( + "Another connection on this destination already writes to these tables: " + f"{stream_list}. Set `table_prefix` (for example `_`) so this " + "connection writes to different table names." + ) + + def create_connection( # noqa: PLR0913 # Too many arguments name: str, *, @@ -2029,6 +2107,7 @@ def create_connection( # noqa: PLR0913 # Too many arguments if e.status_code == HTTPStatus.BAD_REQUEST: raise AirbyteLibInputError( message=sdk_error_message(e), + guidance=_conflicting_destination_stream_guidance(e), context={**base_context, **_get_sdk_error_context(e)}, ) from e raise _wrap_sdk_error(e, base_context) from e diff --git a/airbyte/_util/cloud_errors.py b/airbyte/_util/cloud_errors.py index fec2e161d..3fb82ed70 100644 --- a/airbyte/_util/cloud_errors.py +++ b/airbyte/_util/cloud_errors.py @@ -83,22 +83,32 @@ def _reject_json_constant(value: str) -> NoReturn: raise json.JSONDecodeError("Invalid JSON constant", value, 0) +def parse_problem_payload(body: str | None) -> dict[str, Any] | None: + """Parse a JSON problem body no larger than 64 KiB.""" + if not isinstance(body, str) or len(body) > _MAX_PROBLEM_BODY: + return None + try: + if len(body.encode("utf-8")) > _MAX_PROBLEM_BODY: + return None + # Numeric fields are unused; avoid Python's integer-conversion size limit. + payload = json.loads( + body, + parse_int=lambda _: None, + parse_constant=_reject_json_constant, + ) + if isinstance(payload, dict): + # Surface unpaired-surrogate failures here, not in callers. + _problem_message(payload) + except (UnicodeEncodeError, json.JSONDecodeError, RecursionError): + return None + return payload if isinstance(payload, dict) else None + + def parse_cloud_error(status_code: int | None, body: str | None) -> CloudErrorInfo: """Select detail/title from a JSON object of at most 64 KiB, or a status fallback.""" - payload: Any = None - message = None - if isinstance(body, str) and len(body) <= _MAX_PROBLEM_BODY: - try: - if len(body.encode("utf-8")) <= _MAX_PROBLEM_BODY: - # Numeric fields are unused; avoid Python's integer-conversion size limit. - payload = json.loads( - body, parse_int=lambda _: None, parse_constant=_reject_json_constant - ) - if isinstance(payload, dict): - message = _problem_message(payload) - except (UnicodeEncodeError, json.JSONDecodeError, RecursionError): - payload = None - problem = payload if isinstance(payload, dict) else {} + payload = parse_problem_payload(body) + message = _problem_message(payload) if payload is not None else None + problem = payload if payload is not None else {} problem_type = problem.get("type") return CloudErrorInfo( status_code=status_code, diff --git a/airbyte/cloud/workspaces.py b/airbyte/cloud/workspaces.py index 101e7d1eb..f0b77ca90 100644 --- a/airbyte/cloud/workspaces.py +++ b/airbyte/cloud/workspaces.py @@ -997,7 +997,9 @@ def deploy_connection( source: The deployed source. You can pass a source ID or a CloudSource object. destination: The deployed destination. You can pass a destination ID or a CloudDestination object. - table_prefix: Optional. The table prefix to use when syncing to the destination. + table_prefix: Optional prefix for destination table names. Use it to keep + connections to the same destination from writing to the same table + (namespace + name). selected_streams: The selected stream names to sync within the connection. """ if not selected_streams: diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index 6c6aa69e1..b1498d70e 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -908,7 +908,9 @@ def create_connection_on_cloud( table_prefix: Annotated[ str | None, Field( - description="Optional table prefix to use when syncing to the destination.", + description="Optional prefix for destination table names. Two connections to " + "the same destination cannot write to the same table (namespace + name); set " + "a prefix such as `_` to keep them separate.", default=None, ), TraceArg.FINGERPRINT, @@ -922,6 +924,9 @@ def create_connection_on_cloud( When `selected_streams` is not provided, the connection selects the source connector's suggested streams from the connector registry. If the connector has no suggested streams, the call fails with guidance to pass streams explicitly. + + Two connections to the same destination cannot write to the same table (namespace + + name). Use `table_prefix` (for example `_`) to keep them separate. """ resolved_streams_list: list[str] = resolve_list_of_strings(selected_streams) or [] workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id) diff --git a/tests/unit_tests/test_cloud_api_util.py b/tests/unit_tests/test_cloud_api_util.py index f4907260b..3fde2e91a 100644 --- a/tests/unit_tests/test_cloud_api_util.py +++ b/tests/unit_tests/test_cloud_api_util.py @@ -27,6 +27,7 @@ AirbyteWorkspaceNotEmptyError, AirbyteLibInputError, ) +from airbyte.mcp._error_handling import format_user_facing_error from airbyte.registry import ConnectorType from airbyte.secrets.base import SecretString from airbyte_api import api, models, utils @@ -2291,23 +2292,113 @@ def _call_create_connection() -> None: ) -def test_create_connection_400_raises_input_error_without_api_message( +@pytest.mark.parametrize( + ("problem_type", "data", "expected_guidance"), + [ + pytest.param( + "https://reference.airbyte.com/reference/errors" + "#error:connection-conflicting-destination-stream", + { + "streams": [ + { + "streamName": "users", + "connectionIds": ["users-connection"], + }, + { + "streamName": "orders", + "streamNamespace": "", + "connectionIds": ["orders-connection"], + }, + ] + }, + "Another connection on this destination already writes to these tables: " + "`users` (connection `users-connection`), `orders` " + "(connection `orders-connection`). Set `table_prefix` (for example " + "`_`) so this connection writes to different table names.", + id="conflicting-streams", + ), + pytest.param( + "https://reference.airbyte.com/reference/errors" + "#error:connection-conflicting-destination-stream", + { + "streams": [ + { + "streamName": f"stream-{index}", + "streamNamespace": "", + "connectionIds": [f"connection-{index}"], + } + for index in range(12) + ] + }, + ( + "Another connection on this destination already writes to these tables: " + + ", ".join( + f"`stream-{index}` (connection `connection-{index}`)" + for index in range(10) + ) + + ", and 2 more. Set `table_prefix` (for example `_`) so this " + "connection writes to different table names." + ), + id="stream-list-capped", + ), + pytest.param( + "https://reference.airbyte.com/reference/errors" + "#error:connection-conflicting-destination-stream", + { + "streams": [ + { + "streamName": "users", + "streamNamespace": "analytics", + "connectionIds": ["users-connection"], + }, + {"streamName": 42}, + { + "streamName": "orders", + "streamNamespace": None, + "connectionIds": ["orders-connection"], + }, + ] + }, + "Another connection on this destination already writes to these tables: " + "`analytics.users` (connection `users-connection`), `orders` " + "(connection `orders-connection`). Set `table_prefix` (for example " + "`_`) so this connection writes to different table names.", + id="malformed-stream-item-skipped", + ), + pytest.param( + "https://reference.airbyte.com/reference/errors" + "#error:connection-conflicting-destination-stream", + {"streams": [{"streamName": ""}, {"streamName": "orders"}]}, + "Another connection on this destination may already write to the same tables. " + "Set `table_prefix` (for example `_`) so this connection writes to " + "different table names.", + id="all-stream-items-malformed", + ), + pytest.param( + "https://reference.airbyte.com/reference/errors#bad-request", + {"message": "Internal API message"}, + None, + id="non-conflict-400", + ), + ], +) +def test_create_connection_400_guidance( monkeypatch: pytest.MonkeyPatch, + problem_type: str, + data: Any, + expected_guidance: str | None, ) -> None: - """A 400 raises `AirbyteLibInputError`; `data.message` is never relayed.""" - api_message = ( - "No streams found with name [no_such_stream] and namespace [null]. " - "Please call the https://reference.airbyte.com/reference/getstreamproperties " - "endpoint to see the valid stream name and namespace combinations." + """Use conflicting-stream data for guidance without exposing other API details.""" + detail = ( + "The request could not be understood by the server due to malformed syntax." ) body = json.dumps({ "status": 400, - "type": "https://reference.airbyte.com/reference/errors#bad-request", + "type": problem_type, "title": "bad-request", - "detail": "The request could not be understood by the server " - "due to malformed syntax.", + "detail": detail, "documentationUrl": None, - "data": {"message": api_message}, + "data": data, }) raw_response = requests.Response() raw_response.status_code = 400 @@ -2326,15 +2417,15 @@ def test_create_connection_400_raises_input_error_without_api_message( with pytest.raises(AirbyteLibInputError) as exc_info: _call_create_connection() - assert exc_info.value.get_message() == ( - "The request could not be understood by the server due to malformed syntax." - ) - assert exc_info.value.guidance is None - assert api_message not in str(exc_info.value) + assert exc_info.value.get_message() == detail + assert exc_info.value.guidance == expected_guidance + assert "Internal API message" not in str(exc_info.value) assert exc_info.value.context["source_id"] == "source-id" assert exc_info.value.context["destination_id"] == "destination-id" assert exc_info.value.context["status_code"] == 400 assert exc_info.value.__cause__ is sdk_error + expected_text = f"{detail} {expected_guidance}" if expected_guidance else detail + assert format_user_facing_error(exc_info.value) == expected_text def test_create_connection_400_with_non_json_body_uses_fallback_message(