build: prepare PyAirbyte slim package boundaries

.github/workflows/release_drafter.yml

@@ -41,8 +41,10 @@ jobs:
       - name: Build package
         env:
           VERSION: ${{ steps.dry-run.outputs.tag-name }}
-        # Standardize by removing preceding 'v' char:
-        run: UV_DYNAMIC_VERSIONING_BYPASS="${VERSION#v}" uv build
+        # Build the main and slim variants with same version.
+        run: |
+          UV_DYNAMIC_VERSIONING_BYPASS="${VERSION#v}" uv build
+          UV_DYNAMIC_VERSIONING_BYPASS="${VERSION#v}" uv run scripts/build_slim_package.uv
 
       - name: Create or update draft release
         uses: aaronsteers/semantic-pr-release-drafter@v1.0.0

airbyte/__init__.py

@@ -123,21 +123,40 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+# ruff: noqa: F822  # Lazy exports are resolved by __getattr__ at runtime.
+from importlib import import_module
+from typing import TYPE_CHECKING, Any, cast
 
 from airbyte import registry
-from airbyte.caches.bigquery import BigQueryCache
-from airbyte.caches.duckdb import DuckDBCache
-from airbyte.caches.util import get_colab_cache, get_default_cache, new_local_cache
-from airbyte.datasets import CachedDataset
-from airbyte.destinations.base import Destination
-from airbyte.destinations.util import get_destination
 from airbyte.records import StreamRecord
 from airbyte.registry import get_available_connectors
-from airbyte.results import ReadResult, WriteResult
 from airbyte.secrets import SecretSourceEnum, get_secret
-from airbyte.sources.base import Source
-from airbyte.sources.util import get_source
+
+
+_LAZY_EXPORTS = {
+    "BigQueryCache": "airbyte.caches.bigquery",
+    "CachedDataset": "airbyte.datasets",
+    "Destination": "airbyte.destinations.base",
+    "DuckDBCache": "airbyte.caches.duckdb",
+    "ReadResult": "airbyte.results",
+    "Source": "airbyte.sources.base",
+    "WriteResult": "airbyte.results",
+    "get_colab_cache": "airbyte.caches.util",
+    "get_default_cache": "airbyte.caches.util",
+    "get_destination": "airbyte.destinations.util",
+    "get_source": "airbyte.sources.util",
+    "new_local_cache": "airbyte.caches.util",
+}
+
+
+def __getattr__(name: str) -> Any:  # noqa: ANN401
+    if name not in _LAZY_EXPORTS:
+        raise AttributeError(f"module 'airbyte' has no attribute {name!r}")
+
+    module = import_module(_LAZY_EXPORTS[name])
+    value = cast("Any", getattr(module, name))
+    globals()[name] = value
+    return value
 
 
 # Submodules imported here for documentation reasons: https://github.com/mitmproxy/pdoc/issues/757
@@ -161,6 +180,15 @@
         secrets,
         sources,
     )
+    from airbyte.caches.bigquery import BigQueryCache
+    from airbyte.caches.duckdb import DuckDBCache
+    from airbyte.caches.util import get_colab_cache, get_default_cache, new_local_cache
+    from airbyte.datasets import CachedDataset
+    from airbyte.destinations.base import Destination
+    from airbyte.destinations.util import get_destination
+    from airbyte.results import ReadResult, WriteResult
+    from airbyte.sources.base import Source
+    from airbyte.sources.util import get_source
 
 
 __all__ = [

airbyte/_executors/python.py

@@ -17,7 +17,7 @@
 from airbyte._util.meta import is_windows
 from airbyte._util.telemetry import EventState, log_install_state
 from airbyte._util.venv_util import get_bin_dir
-from airbyte.constants import DEFAULT_INSTALL_DIR, NO_UV
+from airbyte.constants import DEFAULT_CONNECTOR_PYTHON_VERSIONS, DEFAULT_INSTALL_DIR, NO_UV
 
 
 if TYPE_CHECKING:
@@ -45,7 +45,7 @@ def __init__(
             install_root: (Optional.) The root directory where the virtual environment will be
                 created. If not provided, the current working directory will be used.
             use_python: (Optional.) Python interpreter specification:
-                - True: Use current Python interpreter
+                - True: Use uv-managed default connector Python versions
                 - False: Use Docker instead (handled by factory)
                 - Path: Use interpreter at this path or interpreter name/command
                 - str: Use uv-managed Python version (semver patterns like "3.12", "3.11.5")
@@ -98,6 +98,41 @@ def _run_subprocess_and_raise_on_failure(self, args: list[str]) -> None:
                 log_text=result.stderr.decode("utf-8"),
             )
 
+    def _create_venv(self, python_override: str | None) -> None:
+        uv_cmd_prefix = ["uv"] if not NO_UV else []
+        python_clause: list[str] = ["--python", python_override] if python_override else []
+        venv_cmd: list[str] = [
+            *(uv_cmd_prefix or [sys.executable, "-m"]),
+            "venv",
+            str(self._get_venv_path()),
+            *python_clause,
+        ]
+        print(
+            f"Creating '{self.name}' virtual environment with command '{' '.join(venv_cmd)}'",
+            file=sys.stderr,
+        )
+        self._run_subprocess_and_raise_on_failure(venv_cmd)
+
+    def _create_venv_with_fallbacks(self, python_versions: tuple[str, ...]) -> None:
+        failures: list[exc.AirbyteSubprocessFailedError] = []
+        for python_version in python_versions:
+            try:
+                self._create_venv(python_version)
+            except exc.AirbyteSubprocessFailedError as ex:
+                failures.append(ex)
+            else:
+                self.use_python = python_version
+                return
+
+        raise exc.AirbyteConnectorInstallationError(
+            message="Connector virtual environment could not be created.",
+            context={
+                "connector_name": self.name,
+                "python_versions": python_versions,
+            },
+            log_text=[str(failure.log_text or "") for failure in failures],
+        )
+
     def uninstall(self) -> None:
         if self._get_venv_path().exists():
             rmtree(str(self._get_venv_path()))
@@ -134,20 +169,10 @@ def install(self) -> None:
         elif not NO_UV and isinstance(self.use_python, str):
             python_override = self.use_python
 
-        uv_cmd_prefix = ["uv"] if not NO_UV else []
-        python_clause: list[str] = ["--python", python_override] if python_override else []
-
-        venv_cmd: list[str] = [
-            *(uv_cmd_prefix or [sys.executable, "-m"]),
-            "venv",
-            str(self._get_venv_path()),
-            *python_clause,
-        ]
-        print(
-            f"Creating '{self.name}' virtual environment with command '{' '.join(venv_cmd)}'",
-            file=sys.stderr,
-        )
-        self._run_subprocess_and_raise_on_failure(venv_cmd)
+        if not NO_UV and self.use_python is True:
+            self._create_venv_with_fallbacks(DEFAULT_CONNECTOR_PYTHON_VERSIONS)
+        else:
+            self._create_venv(python_override)
 
         install_cmd = (
             [

airbyte/_executors/util.py

@@ -270,6 +270,7 @@ def get_connector_executor(  # noqa: PLR0912, PLR0913, PLR0914, PLR0915, C901 #
                 case InstallType.PYTHON:
                     pip_url = metadata.pypi_package_name
                     pip_url = f"{pip_url}=={version}" if version else pip_url
+                    use_python = True
                 case _:
                     docker_image = True
 

airbyte/_message_iterators.py

@@ -4,13 +4,12 @@
 from __future__ import annotations
 
 import sys
-from collections.abc import Iterator
+from datetime import datetime, timezone
 from typing import IO, TYPE_CHECKING, cast
 
 import pydantic
 from typing_extensions import final
 
-from airbyte_cdk.utils.datetime_helpers import ab_datetime_now
 from airbyte_protocol.models import (
     AirbyteMessage,
     AirbyteRecordMessage,
@@ -26,7 +25,6 @@
 
 
 if TYPE_CHECKING:
-    import datetime
     from collections.abc import Callable, Generator, Iterable, Iterator
     from pathlib import Path
 
@@ -39,7 +37,7 @@ def _new_stream_success_message(stream_name: str) -> AirbyteMessage:
         type=Type.TRACE,
         trace=AirbyteTraceMessage(
             type=TraceType.STREAM_STATUS,
-            emitted_at=ab_datetime_now().timestamp(),
+            emitted_at=datetime.now(tz=timezone.utc).timestamp(),
             stream_status=AirbyteStreamStatusTraceMessage(
                 stream_descriptor=StreamDescriptor(
                     name=stream_name,
@@ -104,9 +102,7 @@ def generator() -> Generator[AirbyteMessage, None, None]:
                             stream=stream_name,
                             data=record,
                             emitted_at=int(
-                                cast(
-                                    "datetime.datetime", record.get(AB_EXTRACTED_AT_COLUMN)
-                                ).timestamp()
+                                cast("datetime", record.get(AB_EXTRACTED_AT_COLUMN)).timestamp()
                             ),
                             # `meta` and `namespace` are not handled:
                             meta=None,

airbyte/cloud/sync_results.py

@@ -103,31 +103,41 @@
 import time
 from collections.abc import Iterator, Mapping
 from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any
 
 from typing_extensions import final
 
-from airbyte_cdk.utils.datetime_helpers import ab_datetime_parse
-
 from airbyte._util import api_util
-from airbyte.caches._utils._dest_to_cache import destination_to_cache
 from airbyte.cloud.constants import FAILED_STATUSES, FINAL_STATUSES
-from airbyte.datasets import CachedDataset
 from airbyte.exceptions import AirbyteConnectionSyncError, AirbyteConnectionSyncTimeoutError
 
 
+def _parse_datetime(value: str | int) -> datetime:
+    if isinstance(value, int) or (
+        isinstance(value, str)
+        and (value.isdigit() or (value.startswith("-") and value[1:].isdigit()))
+    ):
+        return datetime.fromtimestamp(int(value), tz=timezone.utc)
+
+    if not isinstance(value, str):
+        raise TypeError(f"Could not parse datetime string: {value}")
+
+    normalized = value.replace("Z", "+00:00")
+    return datetime.fromisoformat(normalized)
+
+
 DEFAULT_SYNC_TIMEOUT_SECONDS = 30 * 60  # 30 minutes
 """The default timeout for waiting for a sync job to complete, in seconds."""
 
 if TYPE_CHECKING:
-    from datetime import datetime
-
     import sqlalchemy
 
     from airbyte._util.api_imports import ConnectionResponse, JobResponse, JobStatusEnum
     from airbyte.caches.base import CacheBase
     from airbyte.cloud.connections import CloudConnection
     from airbyte.cloud.workspaces import CloudWorkspace
+    from airbyte.datasets import CachedDataset
 
 
 @dataclass
@@ -168,7 +178,7 @@ def records_synced(self) -> int:
     def created_at(self) -> datetime:
         """Return the creation time of the attempt."""
         timestamp = self._get_attempt_data()["createdAt"]
-        return ab_datetime_parse(timestamp)
+        return _parse_datetime(timestamp)
 
     def _get_attempt_data(self) -> dict[str, Any]:
         """Get attempt data from the provided attempt data."""
@@ -307,7 +317,7 @@ def records_synced(self) -> int:
     def start_time(self) -> datetime:
         """Return the start time of the sync job in UTC."""
         try:
-            return ab_datetime_parse(self._fetch_latest_job_info().start_time)
+            return _parse_datetime(self._fetch_latest_job_info().start_time)
         except (ValueError, TypeError) as e:
             if "Invalid isoformat string" in str(e):
                 job_info_raw = api_util._make_config_api_request(  # noqa: SLF001
@@ -321,7 +331,7 @@ def start_time(self) -> datetime:
                 )
                 raw_start_time = job_info_raw.get("startTime")
                 if raw_start_time:
-                    return ab_datetime_parse(raw_start_time)
+                    return _parse_datetime(raw_start_time)
             raise
 
     def _fetch_job_with_attempts(self) -> dict[str, Any]:
@@ -421,6 +431,8 @@ def get_sql_cache(self) -> CacheBase:
         if self._cache:
             return self._cache
 
+        from airbyte.caches._utils._dest_to_cache import destination_to_cache  # noqa: PLC0415
+
         destination_configuration = self._get_destination_configuration()
         self._cache = destination_to_cache(destination_configuration=destination_configuration)
         return self._cache
@@ -449,6 +461,8 @@ def get_dataset(self, stream_name: str) -> CachedDataset:
               (catalog information) to the `CachedDataset` object via the "Get stream properties"
               API: https://reference.airbyte.com/reference/getstreamproperties
         """
+        from airbyte.datasets import CachedDataset  # noqa: PLC0415
+
         return CachedDataset(
             self.get_sql_cache(),
             stream_name=stream_name,
@@ -482,7 +496,7 @@ def streams(
         """
         return self._SyncResultStreams(self)
 
-    class _SyncResultStreams(Mapping[str, CachedDataset]):
+    class _SyncResultStreams(Mapping[str, "CachedDataset"]):
         """A mapping of stream names to cached datasets."""
 
         def __init__(

airbyte/cloud/workspaces.py

@@ -38,7 +38,7 @@
 from dataclasses import dataclass, field
 from functools import cached_property
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Literal, overload
+from typing import TYPE_CHECKING, Any, Literal, Protocol, overload
 
 import yaml
 
@@ -54,14 +54,22 @@
     CustomCloudSourceDefinition,
 )
 from airbyte.cloud.organizations import CloudOrganization
-from airbyte.destinations.base import Destination
 from airbyte.exceptions import AirbyteError
 
 
+class _DestinationLike(Protocol):
+    name: str
+
+    @property
+    def _hydrated_config(self) -> dict[str, Any]:
+        return {}
+
+
 if TYPE_CHECKING:
     from collections.abc import Callable
 
     from airbyte._util.api_imports import WorkspaceResponse
+    from airbyte.destinations.base import Destination
     from airbyte.secrets.base import SecretString
     from airbyte.sources.base import Source
 
@@ -415,7 +423,7 @@ def deploy_source(
     def deploy_destination(
         self,
         name: str,
-        destination: Destination | dict[str, Any],
+        destination: Destination | _DestinationLike | dict[str, Any],
         *,
         unique: bool = True,
         random_name_suffix: bool = False,
@@ -432,16 +440,16 @@ def deploy_destination(
                 are not allowed. Defaults to `True`.
             random_name_suffix: Whether to append a random suffix to the name.
         """
-        if isinstance(destination, Destination):
-            destination_conf_dict = destination._hydrated_config.copy()  # noqa: SLF001 (non-public API)
-            destination_conf_dict["destinationType"] = destination.name.replace("destination-", "")
-            # raise ValueError(destination_conf_dict)
-        else:
+        if isinstance(destination, dict):
             destination_conf_dict = destination.copy()
-            if "destinationType" not in destination_conf_dict:
-                raise exc.PyAirbyteInputError(
-                    message="Missing `destinationType` in configuration dictionary.",
-                )
+        else:
+            destination_conf_dict = destination._hydrated_config.copy()  # noqa: SLF001
+            destination_conf_dict["destinationType"] = destination.name.replace("destination-", "")
+
+        if "destinationType" not in destination_conf_dict:
+            raise exc.PyAirbyteInputError(
+                message="Missing `destinationType` in configuration dictionary.",
+            )
 
         if random_name_suffix:
             name += f" (ID: {text_util.generate_random_suffix()})"