Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions .github/workflows/release_drafter.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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 build packages/airbyte-slim --out-dir dist

- name: Create or update draft release
uses: aaronsteers/semantic-pr-release-drafter@v1.0.0
Expand Down
48 changes: 38 additions & 10 deletions airbyte/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
}
Comment on lines +136 to +149

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't follow what 'lazy exports' is supposed to mean. If obvious/easy, answer here. If shameful/controversial, answer in slack.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not shameful. Here "lazy exports" means the top-level airbyte module keeps the existing public API names, but it no longer imports full-package modules until the specific name is accessed.

Example: import airbyte as ab stays lightweight; ab.get_source then imports airbyte.sources.util on first access and caches the resolved function. This lets airbyte-slim import the shared top-level package without immediately requiring full-only modules like caches, sources, destinations, CLI, MCP, DuckDB, SQLAlchemy, or CDK.


Devin session



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
Expand All @@ -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__ = [
Expand Down
57 changes: 41 additions & 16 deletions airbyte/_executors/python.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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()))
Expand Down Expand Up @@ -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 = (
[
Expand Down
1 change: 1 addition & 0 deletions airbyte/_executors/util.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
10 changes: 3 additions & 7 deletions airbyte/_message_iterators.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -26,7 +25,6 @@


if TYPE_CHECKING:
import datetime
from collections.abc import Callable, Generator, Iterable, Iterator
from pathlib import Path

Expand All @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
25 changes: 25 additions & 0 deletions airbyte/_util/datetime_parse.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
"""Datetime parsing helpers."""

from __future__ import annotations

from datetime import datetime, timezone


def parse_datetime(value: str | int) -> datetime:
"""Parse an ISO datetime string or Unix timestamp as a timezone-aware 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")
parsed = datetime.fromisoformat(normalized)
if parsed.tzinfo is None:
return parsed.replace(tzinfo=timezone.utc)

return parsed
18 changes: 10 additions & 8 deletions airbyte/cloud/sync_results.py
Original file line number Diff line number Diff line change
Expand Up @@ -107,12 +107,9 @@

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._util.datetime_parse import parse_datetime
from airbyte.cloud.constants import FAILED_STATUSES, FINAL_STATUSES
from airbyte.datasets import CachedDataset
from airbyte.exceptions import AirbyteConnectionSyncError, AirbyteConnectionSyncTimeoutError


Expand All @@ -128,6 +125,7 @@
from airbyte.caches.base import CacheBase
from airbyte.cloud.connections import CloudConnection
from airbyte.cloud.workspaces import CloudWorkspace
from airbyte.datasets import CachedDataset


@dataclass
Expand Down Expand Up @@ -168,7 +166,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."""
Expand Down Expand Up @@ -307,7 +305,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
Expand All @@ -321,7 +319,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]:
Expand Down Expand Up @@ -421,6 +419,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
Expand Down Expand Up @@ -449,6 +449,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,
Expand Down Expand Up @@ -482,7 +484,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__(
Expand Down
32 changes: 20 additions & 12 deletions airbyte/cloud/workspaces.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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

Expand Down Expand Up @@ -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,
Expand All @@ -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()})"
Expand Down
Loading
Loading