Skip to content
Draft
Show file tree
Hide file tree
Changes from 7 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 run scripts/build_slim_package.uv
Comment thread
aaronsteers marked this conversation as resolved.
Outdated

- 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