Merge branch 'main' into devin/1748296708-junit-test-results

Author: aaronsteers

airbyte_cdk/cli/airbyte_cdk/_connector.py

@@ -162,6 +162,12 @@ def test(
         pytest_args.append("--collect-only")
 
     pytest_args.append(str(test_file_path))
+
+    test_results_dir = connector_directory / "build" / "test-results"
+    test_results_dir.mkdir(parents=True, exist_ok=True)
+    junit_xml_path = test_results_dir / "standard-tests-junit.xml"
+    pytest_args.extend(["--junitxml", str(junit_xml_path)])
+
     click.echo(f"Running tests from connector directory: {connector_directory}...")
     click.echo(f"Test file: {test_file_path}")
     click.echo(f"Collect only: {collect_only}")

airbyte_cdk/manifest_migrations/README.md

@@ -20,7 +20,8 @@ This directory contains the logic and registry for manifest migrations in the Ai
 
 3. **Register the Migration:**
    - Open `migrations/registry.yaml`.
-   - Add an entry under the appropriate version, or create a new version section if needed.
+   - Add an entry under the appropriate version, or create a new version section if needed. 
+     - Version can be: "*", "==6.48.3", "~=1.2", ">=1.0.0,<2.0.0", "6.48.3" 
    - Each migration entry should include:
      - `name`: The filename (without `.py`)
      - `order`: The order in which this migration should be applied for the version

airbyte_cdk/manifest_migrations/migration_handler.py

@@ -5,9 +5,11 @@
 
 import copy
 import logging
+import re
 from datetime import datetime, timezone
-from typing import Type
+from typing import Tuple, Type
 
+from packaging.specifiers import SpecifierSet
 from packaging.version import Version
 
 from airbyte_cdk.manifest_migrations.exceptions import (
@@ -25,7 +27,7 @@
 METADATA_TAG = "metadata"
 MANIFEST_VERSION_TAG = "version"
 APPLIED_MIGRATIONS_TAG = "applied_migrations"
-
+WILDCARD_VERSION_PATTERN = ".*"
 LOGGER = logging.getLogger("airbyte.cdk.manifest_migrations")
 
 
@@ -77,11 +79,14 @@ def _handle_migration(
         """
         try:
             migration_instance = migration_class()
-            if self._version_is_valid_for_migration(manifest_version, migration_version):
+            can_apply_migration, should_bump_version = self._version_is_valid_for_migration(
+                manifest_version, migration_version
+            )
+            if can_apply_migration:
                 migration_instance._process_manifest(self._migrated_manifest)
                 if migration_instance.is_migrated:
-                    # set the updated manifest version, after migration has been applied
-                    self._set_manifest_version(migration_version)
+                    if should_bump_version:
+                        self._set_manifest_version(migration_version)
                     self._set_migration_trace(migration_class, manifest_version, migration_version)
             else:
                 LOGGER.info(
@@ -112,18 +117,30 @@ def _version_is_valid_for_migration(
         self,
         manifest_version: str,
         migration_version: str,
-    ) -> bool:
+    ) -> Tuple[bool, bool]:
+        """
+        Decide whether *manifest_version* satisfies the *migration_version* rule.
+
+        Rules
+        -----
+        1. ``"*"``
+           – Wildcard: anything matches.
+        2. String starts with a PEP 440 operator (``==``, ``!=``, ``<=``, ``>=``,
+           ``<``, ``>``, ``~=``, etc.)
+           – Treat *migration_version* as a SpecifierSet and test the manifest
+             version against it.
+        3. Plain version
+           – Interpret both strings as concrete versions and return
+             ``manifest_version <= migration_version``.
         """
-        Checks if the given manifest version is less than or equal to the specified migration version.
+        if re.match(WILDCARD_VERSION_PATTERN, migration_version):
+            return True, False
 
-        Args:
-            manifest_version (str): The version of the manifest to check.
-            migration_version (str): The migration version to compare against.
+        if migration_version.startswith(("=", "!", ">", "<", "~")):
+            spec = SpecifierSet(migration_version)
+            return spec.contains(Version(manifest_version)), False
 
-        Returns:
-            bool: True if the manifest version is less than or equal to the migration version, False otherwise.
-        """
-        return Version(manifest_version) <= Version(migration_version)
+        return Version(manifest_version) <= Version(migration_version), True
 
     def _set_manifest_version(self, version: str) -> None:
         """

airbyte_cdk/manifest_migrations/migrations/__init__.py

@@ -2,3 +2,18 @@
 # Copyright (c) 2025 Airbyte, Inc., all rights reserved.
 #
 
+from airbyte_cdk.manifest_migrations.migrations.http_requester_path_to_url import (
+    HttpRequesterPathToUrl,
+)
+from airbyte_cdk.manifest_migrations.migrations.http_requester_request_body_json_data_to_request_body import (
+    HttpRequesterRequestBodyJsonDataToRequestBody,
+)
+from airbyte_cdk.manifest_migrations.migrations.http_requester_url_base_to_url import (
+    HttpRequesterUrlBaseToUrl,
+)
+
+__all__ = [
+    "HttpRequesterPathToUrl",
+    "HttpRequesterRequestBodyJsonDataToRequestBody",
+    "HttpRequesterUrlBaseToUrl",
+]

airbyte_cdk/manifest_migrations/migrations/registry.yaml

@@ -3,7 +3,7 @@
 #
 
 manifest_migrations:
-  - version: 6.48.3
+  - version: "*"
     migrations: 
       - name: http_requester_url_base_to_url
         order: 1

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -3630,6 +3630,9 @@ definitions:
       delimiter:
         type: string
         default: ","
+      set_empty_cell_to_none:
+        type: boolean
+        default: false
   AsyncJobStatusMap:
     description: Matches the api job status to Async Job Status.
     type: object
@@ -4055,6 +4058,11 @@ definitions:
         title: Value Type
         description: The expected data type of the value. If omitted, the type will be inferred from the value provided.
         "$ref": "#/definitions/ValueType"
+      create_or_update:
+        title: Create or Update
+        description: Determines whether to create a new path if it doesn't exist (true) or only update existing paths (false). When set to true, the resolver will create new paths in the stream template if they don't exist. When false (default), it will only update existing paths.
+        type: boolean
+        default: false
       $parameters:
         type: object
         additionalProperties: true
@@ -4106,6 +4114,12 @@ definitions:
           - ["data"]
           - ["data", "streams"]
           - ["data", "{{ parameters.name }}"]
+      default_values:
+        title: Default Values
+        description: A list of default values, each matching the structure expected from the parsed component value.
+        type: array
+        items:
+          type: object
       $parameters:
         type: object
         additionalProperties: true
@@ -4117,7 +4131,11 @@ definitions:
         type: string
         enum: [ConfigComponentsResolver]
       stream_config:
-        "$ref": "#/definitions/StreamConfig"
+        anyOf:
+          - type: array
+            items:
+              "$ref": "#/definitions/StreamConfig"
+          - "$ref": "#/definitions/StreamConfig"
       components_mapping:
         type: array
         items:

airbyte_cdk/sources/declarative/decoders/composite_raw_decoder.py

@@ -103,6 +103,7 @@ class CsvParser(Parser):
     # TODO: migrate implementation to re-use file-base classes
     encoding: Optional[str] = "utf-8"
     delimiter: Optional[str] = ","
+    set_empty_cell_to_none: Optional[bool] = False
 
     def _get_delimiter(self) -> Optional[str]:
         """
@@ -121,6 +122,8 @@ def parse(self, data: BufferedIOBase) -> PARSER_OUTPUT_TYPE:
         text_data = TextIOWrapper(data, encoding=self.encoding)  # type: ignore
         reader = csv.DictReader(text_data, delimiter=self._get_delimiter() or ",")
         for row in reader:
+            if self.set_empty_cell_to_none:
+                row = {k: (None if v == "" else v) for k, v in row.items()}
             yield row
 
 

airbyte_cdk/sources/declarative/interpolation/macros.py

@@ -6,6 +6,7 @@
 import datetime
 import typing
 from typing import Optional, Union
+from urllib.parse import quote_plus
 
 import isodate
 import pytz
@@ -182,6 +183,17 @@ def format_datetime(
     return DatetimeParser().format(dt=dt_datetime, format=format)
 
 
+def sanitize_url(value: str) -> str:
+    """
+    Sanitizes a value by via urllib.parse.quote_plus
+
+    Usage:
+    `"{{ sanitize_url('https://example.com/path?query=value') }}"`
+    """
+    sanitization_strategy = quote_plus
+    return sanitization_strategy(value)
+
+
 _macros_list = [
     now_utc,
     today_utc,
@@ -193,5 +205,6 @@ def format_datetime(
     format_datetime,
     today_with_timezone,
     str_to_datetime,
+    sanitize_url,
 ]
 macros = {f.__name__: f for f in _macros_list}

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -128,7 +128,7 @@ def __init__(
             component_factory
             if component_factory
             else ModelToComponentFactory(
-                emit_connector_builder_messages,
+                emit_connector_builder_messages=emit_connector_builder_messages,
                 max_concurrent_async_job_count=source_config.get("max_concurrent_async_job_count"),
             )
         )
@@ -300,9 +300,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             }
         )
 
-        stream_configs = self._stream_configs(self._source_config) + self._dynamic_stream_configs(
-            self._source_config, config
-        )
+        stream_configs = self._stream_configs(self._source_config) + self.dynamic_streams
 
         api_budget_model = self._source_config.get("api_budget")
         if api_budget_model:

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1383,6 +1383,7 @@ class CsvDecoder(BaseModel):
     type: Literal["CsvDecoder"]
     encoding: Optional[str] = "utf-8"
     delimiter: Optional[str] = ","
+    set_empty_cell_to_none: Optional[bool] = False
 
 
 class AsyncJobStatusMap(BaseModel):
@@ -1478,6 +1479,11 @@ class ComponentMappingDefinition(BaseModel):
         description="The expected data type of the value. If omitted, the type will be inferred from the value provided.",
         title="Value Type",
     )
+    create_or_update: Optional[bool] = Field(
+        False,
+        description="Determines whether to create a new path if it doesn't exist (true) or only update existing paths (false). When set to true, the resolver will create new paths in the stream template if they don't exist. When false (default), it will only update existing paths.",
+        title="Create or Update",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
@@ -1489,12 +1495,17 @@ class StreamConfig(BaseModel):
         examples=[["data"], ["data", "streams"], ["data", "{{ parameters.name }}"]],
         title="Configs Pointer",
     )
+    default_values: Optional[List[Dict[str, Any]]] = Field(
+        None,
+        description="A list of default values, each matching the structure expected from the parsed component value.",
+        title="Default Values",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
 class ConfigComponentsResolver(BaseModel):
     type: Literal["ConfigComponentsResolver"]
-    stream_config: StreamConfig
+    stream_config: Union[List[StreamConfig], StreamConfig]
     components_mapping: List[ComponentMappingDefinition]
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")