feat: replace pip with uv and add use_python parameter

airbyte/_executors/python.py

@@ -17,6 +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 NO_UV
 
 
 if TYPE_CHECKING:
@@ -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 or interpreter name/command
+                - str: Use uv-managed Python version (semver patterns like "3.12", "3.11.5")
         """
         super().__init__(name=name, metadata=metadata, target_version=target_version)
 
@@ -59,6 +66,7 @@ def __init__(
             else f"airbyte-{self.name}"
         )
         self.install_root = install_root or Path.cwd()
+        self.use_python = use_python
 
     def _get_venv_name(self) -> str:
         return f".venv-{self.name}"
@@ -106,20 +114,62 @@ 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())]
-        )
+        if not (
+            self.use_python is None
+            or self.use_python is True
+            or self.use_python is False
+            or isinstance(self.use_python, (str, Path))
+        ):
+            raise exc.PyAirbyteInputError(
+                message="Invalid use_python parameter type",
+                input_value=str(self.use_python),
+            )
+
+        python_override: str | None = None
+        if not NO_UV and isinstance(self.use_python, Path):
+            python_override = str(self.use_python.absolute())
+
+        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 []
 
-        pip_path = str(get_bin_dir(self._get_venv_path()) / "pip")
+        venv_cmd: list[str] = [
+            *(uv_cmd_prefix or [sys.executable, "-m"]),
+            "venv",
+            str(self._get_venv_path()),
+            *python_clause,
+        ]
         print(
-            f"Installing '{self.name}' into virtual environment '{self._get_venv_path()!s}'.\n"
-            f"Running 'pip install {self.pip_url}'...\n",
+            f"Creating '{self.name}' virtual environment with command '{' '.join(venv_cmd)}'",
+            file=sys.stderr,
+        )
+        self._run_subprocess_and_raise_on_failure(venv_cmd)
+
+        install_cmd = (
+            [
+                "uv",
+                "pip",
+                "install",
+                "--python",  # uv requires --python after the subcommand
+                str(self.interpreter_path),
+            ]
+            if not NO_UV
+            else [
+                "pip",
+                "--python",  # pip requires --python before the subcommand
+                str(self.interpreter_path),
+                "install",
+            ]
+        ) + shlex.split(self.pip_url)
+        print(
+            f"Installing '{self.name}' into virtual environment '{self._get_venv_path()!s}' with "
+            f"command '{' '.join(install_cmd)}'...\n",
             file=sys.stderr,
         )
         try:
-            self._run_subprocess_and_raise_on_failure(
-                args=[pip_path, "install", *shlex.split(self.pip_url)]
-            )
+            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,19 +166,25 @@ 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.
 
     For documentation of each arg, see the function `airbyte.sources.util.get_source()`.
     """
-    install_method_count = sum(
-        [
-            bool(local_executable),
-            bool(docker_image),
-            bool(pip_url),
-            bool(source_manifest),
-        ]
-    )
+    install_method_count = sum([
+        bool(local_executable),
+        bool(docker_image),
+        bool(pip_url) or bool(use_python),
+        bool(source_manifest),
+    ])
+
+    if use_python is False:
+        docker_image = True
+
+    if use_python is None and pip_url is not None:
+        # If pip_url is set, we assume the user wants to use Python.
+        use_python = True
 
     if version and pip_url:
         raise exc.PyAirbyteInputError(
@@ -341,6 +347,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

@@ -61,6 +61,7 @@
 
 from __future__ import annotations
 
+import re
 import sys
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
@@ -174,12 +175,43 @@ def _get_connector_name(connector: str) -> str:
     return connector
 
 
+def _parse_use_python(use_python_str: str | None) -> bool | Path | str | None:
+    r"""Parse the use_python CLI parameter.
+
+    Args:
+        use_python_str: The raw string value from CLI input.
+
+    Returns:
+        - None: No parameter provided
+        - True: Use current Python interpreter ("true")
+        - False: Use Docker instead ("false")
+        - Path: Use interpreter at this path (paths containing / or \ or starting with .)
+        - str: Use uv-managed Python version (semver patterns like "3.12", "3.11.5")
+               or existing interpreter name (non-semver strings like "python3.10")
+    """
+    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 "\\" in use_python_str or use_python_str.startswith("."):
+        return Path(use_python_str)
+
+    semver_pattern = r"^\d+\.\d+(?:\.\d+)?$"
+    if re.match(semver_pattern, use_python_str):
+        return use_python_str  # Return as string for uv-managed version
+
+    return Path(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 +223,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 +240,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 +260,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 +277,7 @@ def _resolve_source_job(
         config=config_dict,
         streams=streams_list,
         pip_url=pip_url,
+        use_python=use_python_parsed,
     )
 
 
@@ -256,6 +294,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.
 
@@ -266,8 +305,11 @@ 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 {}
+    use_python_parsed = _parse_use_python(use_python)
+
     destination_name = _get_connector_name(destination)
 
     if destination_name == "destination-dev-null" and not config:
@@ -279,6 +321,7 @@ def _resolve_destination_job(
             docker_image=destination,
             config=config_dict,
             pip_url=pip_url,
+            use_python=use_python_parsed,
         )
 
     if destination and (destination.startswith(".") or "/" in destination):
@@ -296,6 +339,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.
@@ -304,6 +348,7 @@ def _resolve_destination_job(
         name=destination,
         config=config_dict,
         pip_url=pip_url,
+        use_python=use_python_parsed,
     )
 
 
@@ -334,10 +379,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:
@@ -352,12 +407,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)
@@ -413,12 +470,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.
 
@@ -441,6 +508,7 @@ def benchmark(
             source=source,
             config=config,
             streams=streams,
+            use_python=use_python,
         )
         if source
         else get_benchmark_source(
@@ -451,6 +519,7 @@ def benchmark(
         _resolve_destination_job(
             destination=destination,
             config=config,
+            use_python=use_python,
         )
         if destination
         else get_noop_destination()
@@ -513,6 +582,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,
@@ -522,6 +600,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.
 
@@ -536,11 +615,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...")