We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Proposed: Should we add explicit `virtualenv` dependency
Follows from:

- https://github.com/meltano/meltano/issues/6882

We used to recommend _first_ installing `virtualenv`, then using `virtualenv` to install Meltano. This meant that Meltano was able to use `virtualenv` for plugin installation even without any explicit dependency declaration.

Today, if the user installs with `pipx`, then they also get `virtualenv` installed, since `pipx` ships with `virtualenv`.

Question: Should we just create an explicit `virtualenv` dependency so we can be sure it is available during plugin installation?

A valid scenario where `pip` would be fine for Meltano installation would be on a new+dedicated docker image. Presumably, that would be a safe time to install with `pip` and skip `pipx` installation - without requiring virtualenv isolation for Meltano itself. However, even in those cases, Meltano still needs `virtualenv` for plugin installation, and plugin installations would likely fail unless the image already contains `virtualenv` installed.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/meltano/core/venv_service.py`
Content:
```
1 """Manage Python virtual environments."""
2 
3 from __future__ import annotations
4 
5 import asyncio
6 import hashlib
7 import logging
8 import os
9 import platform
10 import shutil
11 import subprocess
12 import sys
13 from asyncio.subprocess import Process
14 from collections import namedtuple
15 from collections.abc import Iterable
16 from pathlib import Path
17 
18 from meltano.core.error import AsyncSubprocessError
19 from meltano.core.project import Project
20 
21 logger = logging.getLogger(__name__)
22 
23 VenvSpecs = namedtuple("VenvSpecs", ("lib_dir", "bin_dir", "site_packages_dir"))
24 
25 POSIX = VenvSpecs(
26     lib_dir="lib",
27     bin_dir="bin",
28     site_packages_dir=os.path.join(
29         "lib",
30         f"python{'.'.join(str(part) for part in sys.version_info[:2])}",
31         "site-packages",
32     ),
33 )
34 
35 NT = VenvSpecs(
36     lib_dir="Lib",
37     bin_dir="Scripts",
38     site_packages_dir=os.path.join("Lib", "site-packages"),
39 )
40 
41 PLATFORM_SPECS = {"Linux": POSIX, "Darwin": POSIX, "Windows": NT}
42 
43 
44 def venv_platform_specs():
45     """Get virtual environment sub-path info for the current platform.
46 
47     Raises:
48         Exception: This platform is not supported.
49 
50     Returns:
51         Virtual environment sub-path info for the current platform.
52     """
53     system = platform.system()
54     try:
55         return PLATFORM_SPECS[system]
56     except KeyError as ex:
57         raise Exception(f"Platform {system!r} not supported.") from ex
58 
59 
60 PIP_PACKAGES = ("pip", "setuptools==57.5.0", "wheel")
61 
62 
63 class VirtualEnv:
64     """Info about a single virtual environment."""
65 
66     def __init__(self, root: Path):
67         """Initialize the `VirtualEnv` instance.
68 
69         Args:
70             root: The root directory of the virtual environment.
71         """
72         self.root = root.resolve()
73         self.specs = venv_platform_specs()
74 
75     def __getattr__(self, key: str):
76         """Get a specific attribute from this instance.
77 
78         Used to provide `VenvSpecs` attributes for this specific virtual environment.
79 
80         Args:
81             key: The attribute name. Must be one of the `VenvSpecs` attributes.
82 
83         Returns:
84             The root directory of this virtual environment joined to the requested
85             platform-specific path using this platform's `VenvSpecs` instance.
86         """
87         return self.root / getattr(self.specs, key)
88 
89     def __str__(self):
90         """_summary_.
91 
92         Returns:
93             _description_.
94         """
95         return str(self.root)
96 
97 
98 async def exec_async(*args, **kwargs) -> Process:
99     """Run an executable asyncronously in a subprocess.
100 
101     Args:
102         args: Positional arguments for `asyncio.create_subprocess_exec`.
103         kwargs: Keyword arguments for `asyncio.create_subprocess_exec`.
104 
105     Raises:
106         AsyncSubprocessError: The command failed.
107 
108     Returns:
109         The subprocess.
110     """
111     run = await asyncio.create_subprocess_exec(
112         *args,
113         stdout=subprocess.PIPE,
114         stderr=subprocess.PIPE,
115         **kwargs,
116     )
117     await run.wait()
118 
119     if run.returncode != 0:
120         raise AsyncSubprocessError("Command failed", run)
121 
122     return run
123 
124 
125 def fingerprint(pip_install_args: Iterable[str]) -> str:
126     """Generate a hash identifying pip install args.
127 
128     Arguments are sorted and deduplicated before the hash is generated.
129 
130     Args:
131         pip_install_args: Arguments for `pip install`.
132 
133     Returns:
134         The SHA256 hash hex digest of the sorted set of pip install args.
135     """
136     return hashlib.sha256(" ".join(sorted(set(pip_install_args))).encode()).hexdigest()
137 
138 
139 class VenvService:  # noqa: WPS214
140     """Manages virtual environments.
141 
142     The methods in this class are not threadsafe.
143     """
144 
145     def __init__(self, project: Project, namespace: str = "", name: str = ""):
146         """Initialize the `VenvService`.
147 
148         Args:
149             project: The Meltano project.
150             namespace: The namespace for the venv, e.g. a Plugin type.
151             name: The name of the venv, e.g. a Plugin name.
152         """
153         self.project = project
154         self.namespace = namespace
155         self.name = name
156         self.venv = VirtualEnv(self.project.venvs_dir(namespace, name))
157         self.plugin_fingerprint_path = self.venv.root / ".meltano_plugin_fingerprint"
158 
159     async def install(self, pip_install_args: list[str], clean: bool = False) -> None:
160         """Configure a virtual environment, then run pip install with the given args.
161 
162         Args:
163             pip_install_args: Arguments passed to `pip install`.
164             clean: Whether to not attempt to use an existing virtual environment.
165         """
166         if not clean and self.requires_clean_install(pip_install_args):
167             logger.debug(
168                 f"Packages for '{self.namespace}/{self.name}' have changed so performing a clean install."
169             )
170             clean = True
171 
172         self.clean_run_files()
173         await self._pip_install(pip_install_args=pip_install_args, clean=clean)
174         self.write_fingerprint(pip_install_args)
175 
176     def requires_clean_install(self, pip_install_args: list[str]) -> bool:
177         """Determine whether a clean install is needed.
178 
179         Args:
180             pip_install_args: The arguments being passed to `pip install`, used
181                 for fingerprinting the installation.
182 
183         Returns:
184             Whether virtual environment doesn't exist or can't be reused.
185         """
186         # A generator function is used to perform the checks lazily
187         def checks():
188             # The Python installation used to create this venv no longer exists
189             yield not self.exec_path("python").exists()
190             # The deprecated `meltano_venv.pth` feature is used by this venv
191             yield self.venv.site_packages_dir.joinpath("meltano_venv.pth").exists()
192             # The fingerprint of the venv does not match the pip install args
193             existing_fingerprint = self.read_fingerprint()
194             yield existing_fingerprint is None
195             yield existing_fingerprint != fingerprint(pip_install_args)
196 
197         return any(checks())
198 
199     def clean_run_files(self) -> None:
200         """Destroy cached configuration files, if they exist."""
201         try:
202             shutil.rmtree(self.project.run_dir(self.name, make_dirs=False))
203         except FileNotFoundError:
204             logger.debug("No cached configuration files to remove")
205 
206     def clean(self) -> None:
207         """Destroy the virtual environment, if it exists."""
208         try:
209             shutil.rmtree(self.venv.root)
210             logger.debug(
211                 "Removed old virtual environment for '%s/%s'",  # noqa: WPS323
212                 self.namespace,
213                 self.name,
214             )
215         except FileNotFoundError:
216             # If the VirtualEnv has never been created before do nothing
217             logger.debug("No old virtual environment to remove")
218 
219     async def create(self) -> Process:
220         """Create a new virtual environment.
221 
222         Raises:
223             AsyncSubprocessError: The virtual environment could not be created.
224 
225         Returns:
226             The Python process creating the virtual environment.
227         """
228         logger.debug(f"Creating virtual environment for '{self.namespace}/{self.name}'")
229         try:
230             return await exec_async(sys.executable, "-m", "venv", str(self.venv))
231         except AsyncSubprocessError as err:
232             raise AsyncSubprocessError(
233                 f"Could not create the virtualenv for '{self.namespace}/{self.name}'",
234                 err.process,
235             ) from err
236 
237     async def upgrade_pip(self) -> Process:
238         """Upgrade the `pip` package to the latest version in the virtual environment.
239 
240         Raises:
241             AsyncSubprocessError: Failed to upgrade pip to the latest version.
242 
243         Returns:
244             The process running `pip install --upgrade ...`.
245         """
246         logger.debug(f"Upgrading pip for '{self.namespace}/{self.name}'")
247         try:
248             return await self._pip_install(["--upgrade", *PIP_PACKAGES])
249         except AsyncSubprocessError as err:
250             raise AsyncSubprocessError(
251                 "Failed to upgrade pip to the latest version.", err.process
252             ) from err
253 
254     def read_fingerprint(self) -> str | None:
255         """Get the fingerprint of the existing virtual environment.
256 
257         Returns:
258             The fingerprint of the existing virtual environment if it exists.
259             `None` otherwise.
260         """
261         if not self.plugin_fingerprint_path.exists():
262             return None
263         with open(self.plugin_fingerprint_path) as fingerprint_file:
264             return fingerprint_file.read()
265 
266     def write_fingerprint(self, pip_install_args: list[str]) -> None:
267         """Save the fingerprint for this installation.
268 
269         Args:
270             pip_install_args: The arguments being passed to `pip install`.
271         """
272         with open(self.plugin_fingerprint_path, "w") as fingerprint_file:
273             fingerprint_file.write(fingerprint(pip_install_args))
274 
275     def exec_path(self, executable: str) -> Path:
276         """Return the absolute path for the given executable in the virtual environment.
277 
278         Args:
279             executable: The path to the executable relative to the venv bin directory.
280 
281         Returns:
282             The venv bin directory joined to the provided executable.
283         """
284         absolute_executable = self.venv.bin_dir / executable
285         if platform.system() != "Windows":
286             return absolute_executable
287 
288         # On Windows, try using the '.exe' suffixed version if it exists. Use the
289         # regular executable path as a fallback (and for backwards compatibility).
290         absolute_executable_windows = absolute_executable.with_suffix(".exe")
291         return (
292             absolute_executable_windows
293             if absolute_executable_windows.exists()
294             else absolute_executable
295         )
296 
297     async def _pip_install(
298         self, pip_install_args: list[str], clean: bool = False
299     ) -> Process:
300         """Install a package using `pip` in the proper virtual environment.
301 
302         Args:
303             pip_install_args: The arguments to pass to `pip install`.
304             clean: Whether the installation should be done in a clean venv.
305 
306         Raises:
307             AsyncSubprocessError: The command failed.
308 
309         Returns:
310             The process running `pip install` with the provided args.
311         """
312         if clean:
313             self.clean()
314             await self.create()
315             await self.upgrade_pip()
316 
317         pip_install_args_str = " ".join(pip_install_args)
318         log_msg_prefix = (
319             f"Upgrading with args {pip_install_args_str!r} in existing"
320             if "--upgrade" in pip_install_args
321             else f"Installing with args {pip_install_args_str!r} into"
322         )
323         logger.debug(
324             f"{log_msg_prefix} virtual environment for '{self.namespace}/{self.name}'"
325         )
326 
327         try:
328             return await exec_async(
329                 str(self.exec_path("python")), "-m", "pip", "install", *pip_install_args
330             )
331         except AsyncSubprocessError as err:
332             raise AsyncSubprocessError(
333                 f"Failed to install plugin '{self.name}'.", err.process
334             ) from err
335
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/meltano/core/venv_service.py b/src/meltano/core/venv_service.py
--- a/src/meltano/core/venv_service.py
+++ b/src/meltano/core/venv_service.py
@@ -227,7 +227,10 @@
         """
         logger.debug(f"Creating virtual environment for '{self.namespace}/{self.name}'")
         try:
-            return await exec_async(sys.executable, "-m", "venv", str(self.venv))
+            return await exec_async(
+                str(Path(sys.executable).parent / "virtualenv"),
+                str(self.venv),
+            )
         except AsyncSubprocessError as err:
             raise AsyncSubprocessError(
                 f"Could not create the virtualenv for '{self.namespace}/{self.name}'",

{"golden_diff": "diff --git a/src/meltano/core/venv_service.py b/src/meltano/core/venv_service.py\n--- a/src/meltano/core/venv_service.py\n+++ b/src/meltano/core/venv_service.py\n@@ -227,7 +227,10 @@\n         \"\"\"\n         logger.debug(f\"Creating virtual environment for '{self.namespace}/{self.name}'\")\n         try:\n-            return await exec_async(sys.executable, \"-m\", \"venv\", str(self.venv))\n+            return await exec_async(\n+                str(Path(sys.executable).parent / \"virtualenv\"),\n+                str(self.venv),\n+            )\n         except AsyncSubprocessError as err:\n             raise AsyncSubprocessError(\n                 f\"Could not create the virtualenv for '{self.namespace}/{self.name}'\",\n", "issue": "Proposed: Should we add explicit `virtualenv` dependency\nFollows from:\r\n\r\n- https://github.com/meltano/meltano/issues/6882\r\n\r\nWe used to recommend _first_ installing `virtualenv`, then using `virtualenv` to install Meltano. This meant that Meltano was able to use `virtualenv` for plugin installation even without any explicit dependency declaration.\r\n\r\nToday, if the user installs with `pipx`, then they also get `virtualenv` installed, since `pipx` ships with `virtualenv`.\r\n\r\nQuestion: Should we just create an explicit `virtualenv` dependency so we can be sure it is available during plugin installation?\r\n\r\nA valid scenario where `pip` would be fine for Meltano installation would be on a new+dedicated docker image. Presumably, that would be a safe time to install with `pip` and skip `pipx` installation - without requiring virtualenv isolation for Meltano itself. However, even in those cases, Meltano still needs `virtualenv` for plugin installation, and plugin installations would likely fail unless the image already contains `virtualenv` installed.\r\n\n", "before_files": [{"content": "\"\"\"Manage Python virtual environments.\"\"\"\n\nfrom __future__ import annotations\n\nimport asyncio\nimport hashlib\nimport logging\nimport os\nimport platform\nimport shutil\nimport subprocess\nimport sys\nfrom asyncio.subprocess import Process\nfrom collections import namedtuple\nfrom collections.abc import Iterable\nfrom pathlib import Path\n\nfrom meltano.core.error import AsyncSubprocessError\nfrom meltano.core.project import Project\n\nlogger = logging.getLogger(__name__)\n\nVenvSpecs = namedtuple(\"VenvSpecs\", (\"lib_dir\", \"bin_dir\", \"site_packages_dir\"))\n\nPOSIX = VenvSpecs(\n    lib_dir=\"lib\",\n    bin_dir=\"bin\",\n    site_packages_dir=os.path.join(\n        \"lib\",\n        f\"python{'.'.join(str(part) for part in sys.version_info[:2])}\",\n        \"site-packages\",\n    ),\n)\n\nNT = VenvSpecs(\n    lib_dir=\"Lib\",\n    bin_dir=\"Scripts\",\n    site_packages_dir=os.path.join(\"Lib\", \"site-packages\"),\n)\n\nPLATFORM_SPECS = {\"Linux\": POSIX, \"Darwin\": POSIX, \"Windows\": NT}\n\n\ndef venv_platform_specs():\n    \"\"\"Get virtual environment sub-path info for the current platform.\n\n    Raises:\n        Exception: This platform is not supported.\n\n    Returns:\n        Virtual environment sub-path info for the current platform.\n    \"\"\"\n    system = platform.system()\n    try:\n        return PLATFORM_SPECS[system]\n    except KeyError as ex:\n        raise Exception(f\"Platform {system!r} not supported.\") from ex\n\n\nPIP_PACKAGES = (\"pip\", \"setuptools==57.5.0\", \"wheel\")\n\n\nclass VirtualEnv:\n    \"\"\"Info about a single virtual environment.\"\"\"\n\n    def __init__(self, root: Path):\n        \"\"\"Initialize the `VirtualEnv` instance.\n\n        Args:\n            root: The root directory of the virtual environment.\n        \"\"\"\n        self.root = root.resolve()\n        self.specs = venv_platform_specs()\n\n    def __getattr__(self, key: str):\n        \"\"\"Get a specific attribute from this instance.\n\n        Used to provide `VenvSpecs` attributes for this specific virtual environment.\n\n        Args:\n            key: The attribute name. Must be one of the `VenvSpecs` attributes.\n\n        Returns:\n            The root directory of this virtual environment joined to the requested\n            platform-specific path using this platform's `VenvSpecs` instance.\n        \"\"\"\n        return self.root / getattr(self.specs, key)\n\n    def __str__(self):\n        \"\"\"_summary_.\n\n        Returns:\n            _description_.\n        \"\"\"\n        return str(self.root)\n\n\nasync def exec_async(*args, **kwargs) -> Process:\n    \"\"\"Run an executable asyncronously in a subprocess.\n\n    Args:\n        args: Positional arguments for `asyncio.create_subprocess_exec`.\n        kwargs: Keyword arguments for `asyncio.create_subprocess_exec`.\n\n    Raises:\n        AsyncSubprocessError: The command failed.\n\n    Returns:\n        The subprocess.\n    \"\"\"\n    run = await asyncio.create_subprocess_exec(\n        *args,\n        stdout=subprocess.PIPE,\n        stderr=subprocess.PIPE,\n        **kwargs,\n    )\n    await run.wait()\n\n    if run.returncode != 0:\n        raise AsyncSubprocessError(\"Command failed\", run)\n\n    return run\n\n\ndef fingerprint(pip_install_args: Iterable[str]) -> str:\n    \"\"\"Generate a hash identifying pip install args.\n\n    Arguments are sorted and deduplicated before the hash is generated.\n\n    Args:\n        pip_install_args: Arguments for `pip install`.\n\n    Returns:\n        The SHA256 hash hex digest of the sorted set of pip install args.\n    \"\"\"\n    return hashlib.sha256(\" \".join(sorted(set(pip_install_args))).encode()).hexdigest()\n\n\nclass VenvService:  # noqa: WPS214\n    \"\"\"Manages virtual environments.\n\n    The methods in this class are not threadsafe.\n    \"\"\"\n\n    def __init__(self, project: Project, namespace: str = \"\", name: str = \"\"):\n        \"\"\"Initialize the `VenvService`.\n\n        Args:\n            project: The Meltano project.\n            namespace: The namespace for the venv, e.g. a Plugin type.\n            name: The name of the venv, e.g. a Plugin name.\n        \"\"\"\n        self.project = project\n        self.namespace = namespace\n        self.name = name\n        self.venv = VirtualEnv(self.project.venvs_dir(namespace, name))\n        self.plugin_fingerprint_path = self.venv.root / \".meltano_plugin_fingerprint\"\n\n    async def install(self, pip_install_args: list[str], clean: bool = False) -> None:\n        \"\"\"Configure a virtual environment, then run pip install with the given args.\n\n        Args:\n            pip_install_args: Arguments passed to `pip install`.\n            clean: Whether to not attempt to use an existing virtual environment.\n        \"\"\"\n        if not clean and self.requires_clean_install(pip_install_args):\n            logger.debug(\n                f\"Packages for '{self.namespace}/{self.name}' have changed so performing a clean install.\"\n            )\n            clean = True\n\n        self.clean_run_files()\n        await self._pip_install(pip_install_args=pip_install_args, clean=clean)\n        self.write_fingerprint(pip_install_args)\n\n    def requires_clean_install(self, pip_install_args: list[str]) -> bool:\n        \"\"\"Determine whether a clean install is needed.\n\n        Args:\n            pip_install_args: The arguments being passed to `pip install`, used\n                for fingerprinting the installation.\n\n        Returns:\n            Whether virtual environment doesn't exist or can't be reused.\n        \"\"\"\n        # A generator function is used to perform the checks lazily\n        def checks():\n            # The Python installation used to create this venv no longer exists\n            yield not self.exec_path(\"python\").exists()\n            # The deprecated `meltano_venv.pth` feature is used by this venv\n            yield self.venv.site_packages_dir.joinpath(\"meltano_venv.pth\").exists()\n            # The fingerprint of the venv does not match the pip install args\n            existing_fingerprint = self.read_fingerprint()\n            yield existing_fingerprint is None\n            yield existing_fingerprint != fingerprint(pip_install_args)\n\n        return any(checks())\n\n    def clean_run_files(self) -> None:\n        \"\"\"Destroy cached configuration files, if they exist.\"\"\"\n        try:\n            shutil.rmtree(self.project.run_dir(self.name, make_dirs=False))\n        except FileNotFoundError:\n            logger.debug(\"No cached configuration files to remove\")\n\n    def clean(self) -> None:\n        \"\"\"Destroy the virtual environment, if it exists.\"\"\"\n        try:\n            shutil.rmtree(self.venv.root)\n            logger.debug(\n                \"Removed old virtual environment for '%s/%s'\",  # noqa: WPS323\n                self.namespace,\n                self.name,\n            )\n        except FileNotFoundError:\n            # If the VirtualEnv has never been created before do nothing\n            logger.debug(\"No old virtual environment to remove\")\n\n    async def create(self) -> Process:\n        \"\"\"Create a new virtual environment.\n\n        Raises:\n            AsyncSubprocessError: The virtual environment could not be created.\n\n        Returns:\n            The Python process creating the virtual environment.\n        \"\"\"\n        logger.debug(f\"Creating virtual environment for '{self.namespace}/{self.name}'\")\n        try:\n            return await exec_async(sys.executable, \"-m\", \"venv\", str(self.venv))\n        except AsyncSubprocessError as err:\n            raise AsyncSubprocessError(\n                f\"Could not create the virtualenv for '{self.namespace}/{self.name}'\",\n                err.process,\n            ) from err\n\n    async def upgrade_pip(self) -> Process:\n        \"\"\"Upgrade the `pip` package to the latest version in the virtual environment.\n\n        Raises:\n            AsyncSubprocessError: Failed to upgrade pip to the latest version.\n\n        Returns:\n            The process running `pip install --upgrade ...`.\n        \"\"\"\n        logger.debug(f\"Upgrading pip for '{self.namespace}/{self.name}'\")\n        try:\n            return await self._pip_install([\"--upgrade\", *PIP_PACKAGES])\n        except AsyncSubprocessError as err:\n            raise AsyncSubprocessError(\n                \"Failed to upgrade pip to the latest version.\", err.process\n            ) from err\n\n    def read_fingerprint(self) -> str | None:\n        \"\"\"Get the fingerprint of the existing virtual environment.\n\n        Returns:\n            The fingerprint of the existing virtual environment if it exists.\n            `None` otherwise.\n        \"\"\"\n        if not self.plugin_fingerprint_path.exists():\n            return None\n        with open(self.plugin_fingerprint_path) as fingerprint_file:\n            return fingerprint_file.read()\n\n    def write_fingerprint(self, pip_install_args: list[str]) -> None:\n        \"\"\"Save the fingerprint for this installation.\n\n        Args:\n            pip_install_args: The arguments being passed to `pip install`.\n        \"\"\"\n        with open(self.plugin_fingerprint_path, \"w\") as fingerprint_file:\n            fingerprint_file.write(fingerprint(pip_install_args))\n\n    def exec_path(self, executable: str) -> Path:\n        \"\"\"Return the absolute path for the given executable in the virtual environment.\n\n        Args:\n            executable: The path to the executable relative to the venv bin directory.\n\n        Returns:\n            The venv bin directory joined to the provided executable.\n        \"\"\"\n        absolute_executable = self.venv.bin_dir / executable\n        if platform.system() != \"Windows\":\n            return absolute_executable\n\n        # On Windows, try using the '.exe' suffixed version if it exists. Use the\n        # regular executable path as a fallback (and for backwards compatibility).\n        absolute_executable_windows = absolute_executable.with_suffix(\".exe\")\n        return (\n            absolute_executable_windows\n            if absolute_executable_windows.exists()\n            else absolute_executable\n        )\n\n    async def _pip_install(\n        self, pip_install_args: list[str], clean: bool = False\n    ) -> Process:\n        \"\"\"Install a package using `pip` in the proper virtual environment.\n\n        Args:\n            pip_install_args: The arguments to pass to `pip install`.\n            clean: Whether the installation should be done in a clean venv.\n\n        Raises:\n            AsyncSubprocessError: The command failed.\n\n        Returns:\n            The process running `pip install` with the provided args.\n        \"\"\"\n        if clean:\n            self.clean()\n            await self.create()\n            await self.upgrade_pip()\n\n        pip_install_args_str = \" \".join(pip_install_args)\n        log_msg_prefix = (\n            f\"Upgrading with args {pip_install_args_str!r} in existing\"\n            if \"--upgrade\" in pip_install_args\n            else f\"Installing with args {pip_install_args_str!r} into\"\n        )\n        logger.debug(\n            f\"{log_msg_prefix} virtual environment for '{self.namespace}/{self.name}'\"\n        )\n\n        try:\n            return await exec_async(\n                str(self.exec_path(\"python\")), \"-m\", \"pip\", \"install\", *pip_install_args\n            )\n        except AsyncSubprocessError as err:\n            raise AsyncSubprocessError(\n                f\"Failed to install plugin '{self.name}'.\", err.process\n            ) from err\n", "path": "src/meltano/core/venv_service.py"}], "after_files": [{"content": "\"\"\"Manage Python virtual environments.\"\"\"\n\nfrom __future__ import annotations\n\nimport asyncio\nimport hashlib\nimport logging\nimport os\nimport platform\nimport shutil\nimport subprocess\nimport sys\nfrom asyncio.subprocess import Process\nfrom collections import namedtuple\nfrom collections.abc import Iterable\nfrom pathlib import Path\n\nfrom meltano.core.error import AsyncSubprocessError\nfrom meltano.core.project import Project\n\nlogger = logging.getLogger(__name__)\n\nVenvSpecs = namedtuple(\"VenvSpecs\", (\"lib_dir\", \"bin_dir\", \"site_packages_dir\"))\n\nPOSIX = VenvSpecs(\n    lib_dir=\"lib\",\n    bin_dir=\"bin\",\n    site_packages_dir=os.path.join(\n        \"lib\",\n        f\"python{'.'.join(str(part) for part in sys.version_info[:2])}\",\n        \"site-packages\",\n    ),\n)\n\nNT = VenvSpecs(\n    lib_dir=\"Lib\",\n    bin_dir=\"Scripts\",\n    site_packages_dir=os.path.join(\"Lib\", \"site-packages\"),\n)\n\nPLATFORM_SPECS = {\"Linux\": POSIX, \"Darwin\": POSIX, \"Windows\": NT}\n\n\ndef venv_platform_specs():\n    \"\"\"Get virtual environment sub-path info for the current platform.\n\n    Raises:\n        Exception: This platform is not supported.\n\n    Returns:\n        Virtual environment sub-path info for the current platform.\n    \"\"\"\n    system = platform.system()\n    try:\n        return PLATFORM_SPECS[system]\n    except KeyError as ex:\n        raise Exception(f\"Platform {system!r} not supported.\") from ex\n\n\nPIP_PACKAGES = (\"pip\", \"setuptools==57.5.0\", \"wheel\")\n\n\nclass VirtualEnv:\n    \"\"\"Info about a single virtual environment.\"\"\"\n\n    def __init__(self, root: Path):\n        \"\"\"Initialize the `VirtualEnv` instance.\n\n        Args:\n            root: The root directory of the virtual environment.\n        \"\"\"\n        self.root = root.resolve()\n        self.specs = venv_platform_specs()\n\n    def __getattr__(self, key: str):\n        \"\"\"Get a specific attribute from this instance.\n\n        Used to provide `VenvSpecs` attributes for this specific virtual environment.\n\n        Args:\n            key: The attribute name. Must be one of the `VenvSpecs` attributes.\n\n        Returns:\n            The root directory of this virtual environment joined to the requested\n            platform-specific path using this platform's `VenvSpecs` instance.\n        \"\"\"\n        return self.root / getattr(self.specs, key)\n\n    def __str__(self):\n        \"\"\"_summary_.\n\n        Returns:\n            _description_.\n        \"\"\"\n        return str(self.root)\n\n\nasync def exec_async(*args, **kwargs) -> Process:\n    \"\"\"Run an executable asyncronously in a subprocess.\n\n    Args:\n        args: Positional arguments for `asyncio.create_subprocess_exec`.\n        kwargs: Keyword arguments for `asyncio.create_subprocess_exec`.\n\n    Raises:\n        AsyncSubprocessError: The command failed.\n\n    Returns:\n        The subprocess.\n    \"\"\"\n    run = await asyncio.create_subprocess_exec(\n        *args,\n        stdout=subprocess.PIPE,\n        stderr=subprocess.PIPE,\n        **kwargs,\n    )\n    await run.wait()\n\n    if run.returncode != 0:\n        raise AsyncSubprocessError(\"Command failed\", run)\n\n    return run\n\n\ndef fingerprint(pip_install_args: Iterable[str]) -> str:\n    \"\"\"Generate a hash identifying pip install args.\n\n    Arguments are sorted and deduplicated before the hash is generated.\n\n    Args:\n        pip_install_args: Arguments for `pip install`.\n\n    Returns:\n        The SHA256 hash hex digest of the sorted set of pip install args.\n    \"\"\"\n    return hashlib.sha256(\" \".join(sorted(set(pip_install_args))).encode()).hexdigest()\n\n\nclass VenvService:  # noqa: WPS214\n    \"\"\"Manages virtual environments.\n\n    The methods in this class are not threadsafe.\n    \"\"\"\n\n    def __init__(self, project: Project, namespace: str = \"\", name: str = \"\"):\n        \"\"\"Initialize the `VenvService`.\n\n        Args:\n            project: The Meltano project.\n            namespace: The namespace for the venv, e.g. a Plugin type.\n            name: The name of the venv, e.g. a Plugin name.\n        \"\"\"\n        self.project = project\n        self.namespace = namespace\n        self.name = name\n        self.venv = VirtualEnv(self.project.venvs_dir(namespace, name))\n        self.plugin_fingerprint_path = self.venv.root / \".meltano_plugin_fingerprint\"\n\n    async def install(self, pip_install_args: list[str], clean: bool = False) -> None:\n        \"\"\"Configure a virtual environment, then run pip install with the given args.\n\n        Args:\n            pip_install_args: Arguments passed to `pip install`.\n            clean: Whether to not attempt to use an existing virtual environment.\n        \"\"\"\n        if not clean and self.requires_clean_install(pip_install_args):\n            logger.debug(\n                f\"Packages for '{self.namespace}/{self.name}' have changed so performing a clean install.\"\n            )\n            clean = True\n\n        self.clean_run_files()\n        await self._pip_install(pip_install_args=pip_install_args, clean=clean)\n        self.write_fingerprint(pip_install_args)\n\n    def requires_clean_install(self, pip_install_args: list[str]) -> bool:\n        \"\"\"Determine whether a clean install is needed.\n\n        Args:\n            pip_install_args: The arguments being passed to `pip install`, used\n                for fingerprinting the installation.\n\n        Returns:\n            Whether virtual environment doesn't exist or can't be reused.\n        \"\"\"\n        # A generator function is used to perform the checks lazily\n        def checks():\n            # The Python installation used to create this venv no longer exists\n            yield not self.exec_path(\"python\").exists()\n            # The deprecated `meltano_venv.pth` feature is used by this venv\n            yield self.venv.site_packages_dir.joinpath(\"meltano_venv.pth\").exists()\n            # The fingerprint of the venv does not match the pip install args\n            existing_fingerprint = self.read_fingerprint()\n            yield existing_fingerprint is None\n            yield existing_fingerprint != fingerprint(pip_install_args)\n\n        return any(checks())\n\n    def clean_run_files(self) -> None:\n        \"\"\"Destroy cached configuration files, if they exist.\"\"\"\n        try:\n            shutil.rmtree(self.project.run_dir(self.name, make_dirs=False))\n        except FileNotFoundError:\n            logger.debug(\"No cached configuration files to remove\")\n\n    def clean(self) -> None:\n        \"\"\"Destroy the virtual environment, if it exists.\"\"\"\n        try:\n            shutil.rmtree(self.venv.root)\n            logger.debug(\n                \"Removed old virtual environment for '%s/%s'\",  # noqa: WPS323\n                self.namespace,\n                self.name,\n            )\n        except FileNotFoundError:\n            # If the VirtualEnv has never been created before do nothing\n            logger.debug(\"No old virtual environment to remove\")\n\n    async def create(self) -> Process:\n        \"\"\"Create a new virtual environment.\n\n        Raises:\n            AsyncSubprocessError: The virtual environment could not be created.\n\n        Returns:\n            The Python process creating the virtual environment.\n        \"\"\"\n        logger.debug(f\"Creating virtual environment for '{self.namespace}/{self.name}'\")\n        try:\n            return await exec_async(\n                str(Path(sys.executable).parent / \"virtualenv\"),\n                str(self.venv),\n            )\n        except AsyncSubprocessError as err:\n            raise AsyncSubprocessError(\n                f\"Could not create the virtualenv for '{self.namespace}/{self.name}'\",\n                err.process,\n            ) from err\n\n    async def upgrade_pip(self) -> Process:\n        \"\"\"Upgrade the `pip` package to the latest version in the virtual environment.\n\n        Raises:\n            AsyncSubprocessError: Failed to upgrade pip to the latest version.\n\n        Returns:\n            The process running `pip install --upgrade ...`.\n        \"\"\"\n        logger.debug(f\"Upgrading pip for '{self.namespace}/{self.name}'\")\n        try:\n            return await self._pip_install([\"--upgrade\", *PIP_PACKAGES])\n        except AsyncSubprocessError as err:\n            raise AsyncSubprocessError(\n                \"Failed to upgrade pip to the latest version.\", err.process\n            ) from err\n\n    def read_fingerprint(self) -> str | None:\n        \"\"\"Get the fingerprint of the existing virtual environment.\n\n        Returns:\n            The fingerprint of the existing virtual environment if it exists.\n            `None` otherwise.\n        \"\"\"\n        if not self.plugin_fingerprint_path.exists():\n            return None\n        with open(self.plugin_fingerprint_path) as fingerprint_file:\n            return fingerprint_file.read()\n\n    def write_fingerprint(self, pip_install_args: list[str]) -> None:\n        \"\"\"Save the fingerprint for this installation.\n\n        Args:\n            pip_install_args: The arguments being passed to `pip install`.\n        \"\"\"\n        with open(self.plugin_fingerprint_path, \"w\") as fingerprint_file:\n            fingerprint_file.write(fingerprint(pip_install_args))\n\n    def exec_path(self, executable: str) -> Path:\n        \"\"\"Return the absolute path for the given executable in the virtual environment.\n\n        Args:\n            executable: The path to the executable relative to the venv bin directory.\n\n        Returns:\n            The venv bin directory joined to the provided executable.\n        \"\"\"\n        absolute_executable = self.venv.bin_dir / executable\n        if platform.system() != \"Windows\":\n            return absolute_executable\n\n        # On Windows, try using the '.exe' suffixed version if it exists. Use the\n        # regular executable path as a fallback (and for backwards compatibility).\n        absolute_executable_windows = absolute_executable.with_suffix(\".exe\")\n        return (\n            absolute_executable_windows\n            if absolute_executable_windows.exists()\n            else absolute_executable\n        )\n\n    async def _pip_install(\n        self, pip_install_args: list[str], clean: bool = False\n    ) -> Process:\n        \"\"\"Install a package using `pip` in the proper virtual environment.\n\n        Args:\n            pip_install_args: The arguments to pass to `pip install`.\n            clean: Whether the installation should be done in a clean venv.\n\n        Raises:\n            AsyncSubprocessError: The command failed.\n\n        Returns:\n            The process running `pip install` with the provided args.\n        \"\"\"\n        if clean:\n            self.clean()\n            await self.create()\n            await self.upgrade_pip()\n\n        pip_install_args_str = \" \".join(pip_install_args)\n        log_msg_prefix = (\n            f\"Upgrading with args {pip_install_args_str!r} in existing\"\n            if \"--upgrade\" in pip_install_args\n            else f\"Installing with args {pip_install_args_str!r} into\"\n        )\n        logger.debug(\n            f\"{log_msg_prefix} virtual environment for '{self.namespace}/{self.name}'\"\n        )\n\n        try:\n            return await exec_async(\n                str(self.exec_path(\"python\")), \"-m\", \"pip\", \"install\", *pip_install_args\n            )\n        except AsyncSubprocessError as err:\n            raise AsyncSubprocessError(\n                f\"Failed to install plugin '{self.name}'.\", err.process\n            ) from err\n", "path": "src/meltano/core/venv_service.py"}]}