We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
bug: Missing Python executable from plugin venv causes error during `meltano install`
### Meltano Version

2.10.0

### Python Version

3.9

### Bug scope

CLI (options, error messages, logging, etc.)

### Operating System

Linux Mint 20.3

### Description

1. Install Python (e.g. using `pyenv`).
2. Use that version of Python to run `meltano install`.
3. Uninstall that version of Python.
4. Install a new version of Python (e.g. going from version 3.9.X to 3.9.Y)
5. Use that version of Python to run `meltano install` again:

```
Installing X plugins...
[...]

Need help fixing this problem? Visit http://melta.no/ for troubleshooting steps, or to
join our friendly Slack community.

[Errno 2] No such file or directory: '<project directory>/.meltano/<installed plugin>/venv/bin/python'
```

The venvs managed by Meltano assume that the underlying Python executable that `venv/bin/python` links to will not be deleted. In reality it may be, and if it is we should detect this and automatically replace the venv (as if the `--clean` flag has been provided).

### Code

_No response_

--- 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", "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.python_path = self.venv.bin_dir / "python"
158         self.plugin_fingerprint_path = self.venv.root / ".meltano_plugin_fingerprint"
159 
160     async def install(self, pip_install_args: list[str], clean: bool = False) -> None:
161         """Configure a virtual environment, then run pip install with the given args.
162 
163         Args:
164             pip_install_args: Arguments passed to `pip install`.
165             clean: Whether to not attempt to use an existing virtual environment.
166         """
167         if not clean and self.requires_clean_install(pip_install_args):
168             logger.debug(
169                 f"Packages for '{self.namespace}/{self.name}' have changed so performing a clean install."
170             )
171             clean = True
172 
173         self.clean_run_files()
174         await self._pip_install(pip_install_args=pip_install_args, clean=clean)
175         self.write_fingerprint(pip_install_args)
176 
177     def requires_clean_install(self, pip_install_args: list[str]) -> bool:
178         """Determine whether a clean install is needed.
179 
180         Args:
181             pip_install_args: The arguments being passed to `pip install`, used
182                 for fingerprinting the installation.
183 
184         Returns:
185             Whether virtual environment doesn't exist or can't be reused.
186         """
187         if self.venv.site_packages_dir.joinpath("meltano_venv.pth").exists():
188             # clean up deprecated feature
189             return True
190         existing_fingerprint = self.read_fingerprint()
191         return (
192             existing_fingerprint != fingerprint(pip_install_args)
193             if existing_fingerprint
194             else True
195         )
196 
197     def clean_run_files(self) -> None:
198         """Destroy cached configuration files, if they exist."""
199         try:
200             shutil.rmtree(self.project.run_dir(self.name, make_dirs=False))
201         except FileNotFoundError:
202             logger.debug("No cached configuration files to remove")
203 
204     def clean(self) -> None:
205         """Destroy the virtual environment, if it exists."""
206         try:
207             shutil.rmtree(self.venv.root)
208             logger.debug(
209                 "Removed old virtual environment for '%s/%s'",  # noqa: WPS323
210                 self.namespace,
211                 self.name,
212             )
213         except FileNotFoundError:
214             # If the VirtualEnv has never been created before do nothing
215             logger.debug("No old virtual environment to remove")
216 
217     async def create(self) -> Process:
218         """Create a new virtual environment.
219 
220         Raises:
221             AsyncSubprocessError: The virtual environment could not be created.
222 
223         Returns:
224             The Python process creating the virtual environment.
225         """
226         logger.debug(f"Creating virtual environment for '{self.namespace}/{self.name}'")
227         try:
228             return await exec_async(sys.executable, "-m", "venv", str(self.venv))
229         except AsyncSubprocessError as err:
230             raise AsyncSubprocessError(
231                 f"Could not create the virtualenv for '{self.namespace}/{self.name}'",
232                 err.process,
233             ) from err
234 
235     async def upgrade_pip(self) -> Process:
236         """Upgrade the `pip` package to the latest version in the virtual environment.
237 
238         Raises:
239             AsyncSubprocessError: Failed to upgrade pip to the latest version.
240 
241         Returns:
242             The process running `pip install --upgrade ...`.
243         """
244         logger.debug(f"Upgrading pip for '{self.namespace}/{self.name}'")
245         try:
246             return await self._pip_install(["--upgrade", *PIP_PACKAGES])
247         except AsyncSubprocessError as err:
248             raise AsyncSubprocessError(
249                 "Failed to upgrade pip to the latest version.", err.process
250             ) from err
251 
252     def read_fingerprint(self) -> str | None:
253         """Get the fingerprint of the existing virtual environment.
254 
255         Returns:
256             The fingerprint of the existing virtual environment if it exists.
257             `None` otherwise.
258         """
259         if not self.plugin_fingerprint_path.exists():
260             return None
261         with open(self.plugin_fingerprint_path) as fingerprint_file:
262             return fingerprint_file.read()
263 
264     def write_fingerprint(self, pip_install_args: list[str]) -> None:
265         """Save the fingerprint for this installation.
266 
267         Args:
268             pip_install_args: The arguments being passed to `pip install`.
269         """
270         with open(self.plugin_fingerprint_path, "wt") as fingerprint_file:
271             fingerprint_file.write(fingerprint(pip_install_args))
272 
273     def exec_path(self, executable: str) -> Path:
274         """Return the absolute path for the given executable in the virtual environment.
275 
276         Args:
277             executable: The path to the executable relative to the venv bin directory.
278 
279         Returns:
280             The venv bin directory joined to the provided executable.
281         """
282         return self.venv.bin_dir / executable
283 
284     async def _pip_install(
285         self, pip_install_args: list[str], clean: bool = False
286     ) -> Process:
287         """Install a package using `pip` in the proper virtual environment.
288 
289         Args:
290             pip_install_args: The arguments to pass to `pip install`.
291             clean: Whether the installation should be done in a clean venv.
292 
293         Raises:
294             AsyncSubprocessError: The command failed.
295 
296         Returns:
297             The process running `pip install` with the provided args.
298         """
299         if clean:
300             self.clean()
301             await self.create()
302             await self.upgrade_pip()
303 
304         pip_install_args_str = " ".join(pip_install_args)
305         log_msg_prefix = (
306             f"Upgrading with args {pip_install_args_str!r} in existing"
307             if "--upgrade" in pip_install_args
308             else f"Installing with args {pip_install_args_str!r} into"
309         )
310         logger.debug(
311             f"{log_msg_prefix} virtual environment for '{self.namespace}/{self.name}'"
312         )
313 
314         try:
315             return await exec_async(
316                 str(self.python_path), "-m", "pip", "install", *pip_install_args
317             )
318         except AsyncSubprocessError as err:
319             raise AsyncSubprocessError(
320                 f"Failed to install plugin '{self.name}'.", err.process
321             ) from err
322
```
--- 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
@@ -154,7 +154,6 @@
         self.namespace = namespace
         self.name = name
         self.venv = VirtualEnv(self.project.venvs_dir(namespace, name))
-        self.python_path = self.venv.bin_dir / "python"
         self.plugin_fingerprint_path = self.venv.root / ".meltano_plugin_fingerprint"
 
     async def install(self, pip_install_args: list[str], clean: bool = False) -> None:
@@ -184,15 +183,18 @@
         Returns:
             Whether virtual environment doesn't exist or can't be reused.
         """
-        if self.venv.site_packages_dir.joinpath("meltano_venv.pth").exists():
-            # clean up deprecated feature
-            return True
-        existing_fingerprint = self.read_fingerprint()
-        return (
-            existing_fingerprint != fingerprint(pip_install_args)
-            if existing_fingerprint
-            else True
-        )
+        # A generator function is used to perform the checks lazily
+        def checks():
+            # The Python installation used to create this venv no longer exists
+            yield not self.exec_path("python").exists()
+            # The deprecated `meltano_venv.pth` feature is used by this venv
+            yield self.venv.site_packages_dir.joinpath("meltano_venv.pth").exists()
+            # The fingerprint of the venv does not match the pip install args
+            existing_fingerprint = self.read_fingerprint()
+            yield existing_fingerprint is None
+            yield existing_fingerprint != fingerprint(pip_install_args)
+
+        return any(checks())
 
     def clean_run_files(self) -> None:
         """Destroy cached configuration files, if they exist."""
@@ -279,7 +281,18 @@
         Returns:
             The venv bin directory joined to the provided executable.
         """
-        return self.venv.bin_dir / executable
+        absolute_executable = self.venv.bin_dir / executable
+        if platform.system() != "Windows":
+            return absolute_executable
+
+        # On Windows, try using the '.exe' suffixed version if it exists. Use the
+        # regular executable path as a fallback (and for backwards compatibility).
+        absolute_executable_windows = absolute_executable.with_suffix(".exe")
+        return (
+            absolute_executable_windows
+            if absolute_executable_windows.exists()
+            else absolute_executable
+        )
 
     async def _pip_install(
         self, pip_install_args: list[str], clean: bool = False
@@ -313,7 +326,7 @@
 
         try:
             return await exec_async(
-                str(self.python_path), "-m", "pip", "install", *pip_install_args
+                str(self.exec_path("python")), "-m", "pip", "install", *pip_install_args
             )
         except AsyncSubprocessError as err:
             raise AsyncSubprocessError(

{"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@@ -154,7 +154,6 @@\n         self.namespace = namespace\n         self.name = name\n         self.venv = VirtualEnv(self.project.venvs_dir(namespace, name))\n-        self.python_path = self.venv.bin_dir / \"python\"\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@@ -184,15 +183,18 @@\n         Returns:\n             Whether virtual environment doesn't exist or can't be reused.\n         \"\"\"\n-        if self.venv.site_packages_dir.joinpath(\"meltano_venv.pth\").exists():\n-            # clean up deprecated feature\n-            return True\n-        existing_fingerprint = self.read_fingerprint()\n-        return (\n-            existing_fingerprint != fingerprint(pip_install_args)\n-            if existing_fingerprint\n-            else True\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@@ -279,7 +281,18 @@\n         Returns:\n             The venv bin directory joined to the provided executable.\n         \"\"\"\n-        return self.venv.bin_dir / executable\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@@ -313,7 +326,7 @@\n \n         try:\n             return await exec_async(\n-                str(self.python_path), \"-m\", \"pip\", \"install\", *pip_install_args\n+                str(self.exec_path(\"python\")), \"-m\", \"pip\", \"install\", *pip_install_args\n             )\n         except AsyncSubprocessError as err:\n             raise AsyncSubprocessError(\n", "issue": "bug: Missing Python executable from plugin venv causes error during `meltano install`\n### Meltano Version\n\n2.10.0\n\n### Python Version\n\n3.9\n\n### Bug scope\n\nCLI (options, error messages, logging, etc.)\n\n### Operating System\n\nLinux Mint 20.3\n\n### Description\n\n1. Install Python (e.g. using `pyenv`).\r\n2. Use that version of Python to run `meltano install`.\r\n3. Uninstall that version of Python.\r\n4. Install a new version of Python (e.g. going from version 3.9.X to 3.9.Y)\r\n5. Use that version of Python to run `meltano install` again:\r\n\r\n```\r\nInstalling X plugins...\r\n[...]\r\n\r\nNeed help fixing this problem? Visit http://melta.no/ for troubleshooting steps, or to\r\njoin our friendly Slack community.\r\n\r\n[Errno 2] No such file or directory: '<project directory>/.meltano/<installed plugin>/venv/bin/python'\r\n```\r\n\r\nThe venvs managed by Meltano assume that the underlying Python executable that `venv/bin/python` links to will not be deleted. In reality it may be, and if it is we should detect this and automatically replace the venv (as if the `--clean` flag has been provided).\n\n### Code\n\n_No response_\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\", \"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.python_path = self.venv.bin_dir / \"python\"\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        if self.venv.site_packages_dir.joinpath(\"meltano_venv.pth\").exists():\n            # clean up deprecated feature\n            return True\n        existing_fingerprint = self.read_fingerprint()\n        return (\n            existing_fingerprint != fingerprint(pip_install_args)\n            if existing_fingerprint\n            else True\n        )\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, \"wt\") 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        return self.venv.bin_dir / executable\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.python_path), \"-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\", \"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, \"wt\") 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"}]}