feat: replace pip with uv and add use_python parameter

airbyte/_executors/python.py

@@ -1,6 +1,7 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 from __future__ import annotations
 
+import os
 import shlex
 import subprocess
 import sys
@@ -32,6 +33,7 @@ def __init__(
         target_version: str | None = None,
         pip_url: str | None = None,
         install_root: Path | None = None,
+        use_python: bool | Path | str | None = None,
     ) -> None:
         """Initialize a connector executor that runs a connector in a virtual environment.
 
@@ -42,6 +44,11 @@ def __init__(
             pip_url: (Optional.) The pip URL of the connector to install.
             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
+                - False: Use Docker instead (handled by factory)
+                - Path: Use interpreter at this path
+                - str: Use uv-managed Python version
         """
         super().__init__(name=name, metadata=metadata, target_version=target_version)
 
@@ -59,6 +66,11 @@ def __init__(
             else f"airbyte-{self.name}"
         )
         self.install_root = install_root or Path.cwd()
+        self.use_python = use_python
+
+    def _should_use_uv(self) -> bool:
+        """Check if uv should be used based on AIRBYTE_NO_UV environment variable."""
+        return os.environ.get("AIRBYTE_NO_UV", "").lower() not in {"1", "true", "yes"}
 
     def _get_venv_name(self) -> str:
         return f".venv-{self.name}"
@@ -106,20 +118,65 @@ def install(self) -> None:
 
         After installation, the installed version will be stored in self.reported_version.
         """
-        self._run_subprocess_and_raise_on_failure(
-            [sys.executable, "-m", "venv", str(self._get_venv_path())]
-        )
-
-        pip_path = str(get_bin_dir(self._get_venv_path()) / "pip")
-        print(
-            f"Installing '{self.name}' into virtual environment '{self._get_venv_path()!s}'.\n"
-            f"Running 'pip install {self.pip_url}'...\n",
-            file=sys.stderr,
-        )
-        try:
+        if self.use_python is True or self.use_python is None:
             self._run_subprocess_and_raise_on_failure(
-                args=[pip_path, "install", *shlex.split(self.pip_url)]
+                [sys.executable, "-m", "venv", str(self._get_venv_path())]
+            )
+
+            if self._should_use_uv():
+                install_cmd = [
+                    "uv",
+                    "pip",
+                    "install",
+                    "--python",
+                    str(self._get_venv_path()),
+                    *shlex.split(self.pip_url),
+                ]
+                tool_name = "uv pip"
+            else:
+                pip_path = str(get_bin_dir(self._get_venv_path()) / "pip")
+                install_cmd = [pip_path, "install", *shlex.split(self.pip_url)]
+                tool_name = "pip"
+
+            print(
+                f"Installing '{self.name}' into virtual environment '{self._get_venv_path()!s}'.\n"
+                f"Running '{tool_name} install {self.pip_url}'...\n",
+                file=sys.stderr,
             )
+        elif isinstance(self.use_python, (str, Path)):
+            if isinstance(self.use_python, str):
+                venv_cmd = ["uv", "venv", str(self._get_venv_path()), "--python", self.use_python]
+            else:
+                venv_cmd = [
+                    "uv",
+                    "venv",
+                    str(self._get_venv_path()),
+                    "--python",
+                    str(self.use_python),
+                ]
+
+            self._run_subprocess_and_raise_on_failure(venv_cmd)
+            install_cmd = [
+                "uv",
+                "pip",
+                "install",
+                "--python",
+                str(self._get_venv_path()),
+                *shlex.split(self.pip_url),
+            ]
+            print(
+                f"Installing '{self.name}' into virtual environment '{self._get_venv_path()!s}'.\n"
+                f"Running 'uv pip install {self.pip_url}'...\n",
+                file=sys.stderr,
+            )
+        else:
+            raise exc.PyAirbyteInputError(
+                message="Invalid use_python parameter type",
+                input_value=str(self.use_python),
+            )
+
+        try:
+            self._run_subprocess_and_raise_on_failure(install_cmd)
         except exc.AirbyteSubprocessFailedError as ex:
             # If the installation failed, remove the virtual environment
             # Otherwise, the connector will be considered as installed and the user may not be able

airbyte/_executors/util.py

@@ -166,6 +166,7 @@ def get_connector_executor(  # noqa: PLR0912, PLR0913, PLR0914, PLR0915, C901 #
     source_manifest: bool | dict | Path | str | None = None,
     install_if_missing: bool = True,
     install_root: Path | None = None,
+    use_python: bool | Path | str | None = None,
 ) -> Executor:
     """This factory function creates an executor for a connector.
 
@@ -180,6 +181,9 @@ def get_connector_executor(  # noqa: PLR0912, PLR0913, PLR0914, PLR0915, C901 #
         ]
     )
 
+    if use_python is False:
+        docker_image = True
+
     if version and pip_url:
         raise exc.PyAirbyteInputError(
             message=(
@@ -341,6 +345,7 @@ def get_connector_executor(  # noqa: PLR0912, PLR0913, PLR0914, PLR0915, C901 #
             target_version=version,
             pip_url=pip_url,
             install_root=install_root,
+            use_python=use_python,
         )
         if install_if_missing:
             executor.ensure_installation()

airbyte/cli.py

@@ -174,12 +174,26 @@ def _get_connector_name(connector: str) -> str:
     return connector
 
 
+def _parse_use_python(use_python_str: str | None) -> bool | Path | str | None:
+    """Parse the use_python CLI parameter."""
+    if use_python_str is None:
+        return None
+    if use_python_str.lower() == "true":
+        return True
+    if use_python_str.lower() == "false":
+        return False
+    if "/" in use_python_str or use_python_str.startswith("."):
+        return Path(use_python_str)
+    return use_python_str
+
+
 def _resolve_source_job(
     *,
     source: str | None = None,
     config: str | None = None,
     streams: str | None = None,
     pip_url: str | None = None,
+    use_python: str | None = None,
 ) -> Source:
     """Resolve the source job into a configured Source object.
 
@@ -191,12 +205,15 @@ def _resolve_source_job(
         streams: A comma-separated list of stream names to select for reading. If set to "*",
             all streams will be selected. If not provided, all streams will be selected.
         pip_url: Optional. A location from which to install the connector.
+        use_python: Optional. Python interpreter specification.
     """
     config_dict = _resolve_config(config) if config else None
     streams_list: str | list[str] = streams or "*"
     if isinstance(streams, str) and streams != "*":
         streams_list = [stream.strip() for stream in streams.split(",")]
 
+    use_python_parsed = _parse_use_python(use_python)
+
     source_obj: Source
     if source and _is_docker_image(source):
         source_obj = get_source(
@@ -205,6 +222,7 @@ def _resolve_source_job(
             config=config_dict,
             streams=streams_list,
             pip_url=pip_url,
+            use_python=use_python_parsed,
         )
         return source_obj
 
@@ -224,6 +242,7 @@ def _resolve_source_job(
             config=config_dict,
             streams=streams_list,
             pip_url=pip_url,
+            use_python=use_python_parsed,
         )
         return source_obj
 
@@ -240,6 +259,7 @@ def _resolve_source_job(
         config=config_dict,
         streams=streams_list,
         pip_url=pip_url,
+        use_python=use_python_parsed,
     )
 
 
@@ -248,6 +268,7 @@ def _resolve_destination_job(
     destination: str,
     config: str | None = None,
     pip_url: str | None = None,
+    use_python: str | None = None,
 ) -> Destination:
     """Resolve the destination job into a configured Destination object.
 
@@ -258,8 +279,10 @@ def _resolve_destination_job(
             and tag.
         config: The path to a configuration file for the named source or destination.
         pip_url: Optional. A location from which to install the connector.
+        use_python: Optional. Python interpreter specification.
     """
     config_dict = _resolve_config(config) if config else None
+    use_python_parsed = _parse_use_python(use_python)
 
     if destination and (destination.startswith(".") or "/" in destination):
         # Treat the destination as a path.
@@ -276,6 +299,7 @@ def _resolve_destination_job(
             local_executable=destination_executable,
             config=config_dict,
             pip_url=pip_url,
+            use_python=use_python_parsed,
         )
 
     # else: # Treat the destination as a name.
@@ -284,6 +308,7 @@ def _resolve_destination_job(
         name=destination,
         config=config_dict,
         pip_url=pip_url,
+        use_python=use_python_parsed,
     )
 
 
@@ -314,10 +339,20 @@ def _resolve_destination_job(
     required=False,
     help=CONFIG_HELP,
 )
+@click.option(
+    "--use-python",
+    type=str,
+    help=(
+        "Python interpreter specification. Use 'true' for current Python, "
+        "'false' for Docker, a path for specific interpreter, or a version "
+        "string for uv-managed Python (e.g., '3.11', 'python3.12')."
+    ),
+)
 def validate(
     connector: str | None = None,
     config: str | None = None,
     pip_url: str | None = None,
+    use_python: str | None = None,
 ) -> None:
     """CLI command to run a `benchmark` operation."""
     if not connector:
@@ -332,12 +367,14 @@ def validate(
             config=None,
             streams=None,
             pip_url=pip_url,
+            use_python=use_python,
         )
     else:  # destination
         connector_obj = _resolve_destination_job(
             destination=connector,
             config=None,
             pip_url=pip_url,
+            use_python=use_python,
         )
 
     print("Getting `spec` output from connector...", file=sys.stderr)
@@ -393,12 +430,22 @@ def validate(
     type=str,
     help=CONFIG_HELP,
 )
+@click.option(
+    "--use-python",
+    type=str,
+    help=(
+        "Python interpreter specification. Use 'true' for current Python, "
+        "'false' for Docker, a path for specific interpreter, or a version "
+        "string for uv-managed Python (e.g., '3.11', 'python3.12')."
+    ),
+)
 def benchmark(
     source: str | None = None,
     streams: str = "*",
     num_records: int | str = "5e5",  # 500,000 records
     destination: str | None = None,
     config: str | None = None,
+    use_python: str | None = None,
 ) -> None:
     """CLI command to run a `benchmark` operation.
 
@@ -421,6 +468,7 @@ def benchmark(
             source=source,
             config=config,
             streams=streams,
+            use_python=use_python,
         )
         if source
         else get_benchmark_source(
@@ -431,6 +479,7 @@ def benchmark(
         _resolve_destination_job(
             destination=destination,
             config=config,
+            use_python=use_python,
         )
         if destination
         else get_noop_destination()
@@ -493,6 +542,15 @@ def benchmark(
     type=str,
     help="Optional pip URL for the destination (Python connectors only). " + PIP_URL_HELP,
 )
+@click.option(
+    "--use-python",
+    type=str,
+    help=(
+        "Python interpreter specification. Use 'true' for current Python, "
+        "'false' for Docker, a path for specific interpreter, or a version "
+        "string for uv-managed Python (e.g., '3.11', 'python3.12')."
+    ),
+)
 def sync(
     *,
     source: str,
@@ -502,6 +560,7 @@ def sync(
     destination_config: str | None = None,
     destination_pip_url: str | None = None,
     streams: str | None = None,
+    use_python: str | None = None,
 ) -> None:
     """CLI command to run a `sync` operation.
 
@@ -516,11 +575,13 @@ def sync(
         config=source_config,
         streams=streams,
         pip_url=source_pip_url,
+        use_python=use_python,
     )
     destination_obj = _resolve_destination_job(
         destination=destination,
         config=destination_config,
         pip_url=destination_pip_url,
+        use_python=use_python,
     )
 
     click.echo("Running sync...")

airbyte/destinations/util.py

@@ -29,6 +29,8 @@ def get_destination(  # noqa: PLR0913 # Too many arguments
     docker_image: str | bool | None = None,
     use_host_network: bool = False,
     install_if_missing: bool = True,
+    install_root: Path | None = None,
+    use_python: bool | Path | str | None = None,
 ) -> Destination:
     """Get a connector by name and version.
 
@@ -58,6 +60,13 @@ def get_destination(  # noqa: PLR0913 # Too many arguments
             `docker_image` is not set.
         install_if_missing: Whether to install the connector if it is not available locally. This
             parameter is ignored when local_executable is set.
+        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
+            - False: Use Docker instead
+            - Path: Use interpreter at this path
+            - str: Use uv-managed Python version
     """
     return Destination(
         name=name,
@@ -71,6 +80,8 @@ def get_destination(  # noqa: PLR0913 # Too many arguments
             docker_image=docker_image,
             use_host_network=use_host_network,
             install_if_missing=install_if_missing,
+            install_root=install_root,
+            use_python=use_python,
         ),
     )