We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Meltano lock fails on migrating to >2.0.0 when variant didnt previously exist
We should update the log message and add a section to the migration docs to help users get variants added for plugins that didnt have a variant previously. For example Airflow locking fails because a variant isnt set for all installations prior to 2.0.0. The error message isnt totally clear in saying this and users will need to manually update their meltano.yml.

Originally discussed in https://github.com/meltano/meltano/issues/6359#issuecomment-1175432513

> Additionally I'm getting Orchestrator 'airflow' variant 'original' is not known to Meltano. Variants: ['apache (default)'] from a default installation of Airflow. I know that its hard because when I originally installed it there wasnt a variant name for airflow but now there is. Is there a way for us to resolve that its the default variant? Maybe that wont work because the default could change, maybe using the discovery.yml when we think were migrating to >2.0.0?
> 
> Or we could log it better and say something like "I notice youre migrating and you dont have a variant, go to the Migration guide and learn how to define a variant" and we write up the steps for locking post migration and adding a variant name:
> 
> remove executable/namespace if defined
> add variant using one available on MeltanoHub. We can list the mapping like airflow -> apache, dbt -> dbt-labs, etc. for plugins that just got variant names
> run meltano lock --all

cc @edgarrmondragon 

--- 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/hub/client.py`
Content:
```
1 """Meltano Hub Client."""
2 
3 from __future__ import annotations
4 
5 from typing import Any
6 
7 import requests
8 from structlog.stdlib import get_logger
9 
10 import meltano
11 from meltano.core.plugin import (
12     BasePlugin,
13     PluginDefinition,
14     PluginRef,
15     PluginType,
16     Variant,
17 )
18 from meltano.core.plugin.error import PluginNotFoundError
19 from meltano.core.plugin.factory import base_plugin_factory
20 from meltano.core.plugin_discovery_service import PluginRepository
21 from meltano.core.project import Project
22 from meltano.core.project_settings_service import ProjectSettingsService
23 
24 from .schema import IndexedPlugin, VariantRef
25 
26 logger = get_logger(__name__)
27 
28 
29 class HubPluginTypeNotFound(Exception):
30     """Raised when a Hub plugin type is not found."""
31 
32     def __init__(self, plugin_type: PluginType):
33         """Create a new HubPluginVariantNotFound.
34 
35         Args:
36             plugin_type: The type of the plugin.
37         """
38         self.plugin_type = plugin_type
39 
40     def __str__(self) -> str:
41         """Return a string representation of the error.
42 
43         Returns:
44             The string representation of the error.
45         """
46         return "{type} is not supported in Meltano Hub. Available plugin types: {types}".format(
47             type=self.plugin_type.descriptor.capitalize(),
48             types=list(PluginType),
49         )
50 
51 
52 class HubPluginVariantNotFound(Exception):
53     """Raised when a Hub plugin variant is not found."""
54 
55     def __init__(
56         self,
57         plugin_type: PluginType,
58         plugin: IndexedPlugin,
59         variant_name: str,
60     ):
61         """Create a new HubPluginVariantNotFound.
62 
63         Args:
64             plugin_type: The type of the plugin.
65             plugin: The indexed plugin.
66             variant_name: The name of the variant that was not found.
67         """
68         self.plugin_type = plugin_type
69         self.plugin = plugin
70         self.variant_name = variant_name
71 
72     def __str__(self) -> str:
73         """Return a string representation of the error.
74 
75         Returns:
76             The string representation of the error.
77         """
78         return "{type} '{name}' variant '{variant}' is not known to Meltano. Variants: {variant_labels}".format(
79             type=self.plugin_type.descriptor.capitalize(),
80             name=self.plugin.name,
81             variant=self.variant_name,
82             variant_labels=self.plugin.variant_labels,
83         )
84 
85 
86 class MeltanoHubService(PluginRepository):
87     """PluginRepository implementation for the Meltano Hub."""
88 
89     def __init__(self, project: Project) -> None:
90         """Initialize the service.
91 
92         Args:
93             project: The Meltano project.
94         """
95         self.project = project
96         self.session = requests.Session()
97         self.session.headers.update(
98             {
99                 "Accept": "application/json",
100                 "User-Agent": f"Meltano/{meltano.__version__}",
101             }
102         )
103 
104         self.settings_service = ProjectSettingsService(self.project)
105 
106         if self.settings_service.get("send_anonymous_usage_stats"):
107             project_id = self.settings_service.get("project_id")
108 
109             self.session.headers["X-Project-ID"] = project_id
110 
111     @property
112     def hub_api_url(self):
113         """Return the URL of the Hub API.
114 
115         Returns:
116             The URL of the Hub API.
117         """
118         hub_url = self.settings_service.get("hub_url")
119         return f"{hub_url}/meltano/api/v1"
120 
121     def plugin_type_endpoint(self, plugin_type: PluginType) -> str:
122         """Return the list endpoint for the given plugin type.
123 
124         Args:
125             plugin_type: The plugin type.
126 
127         Returns:
128             The endpoint for the given plugin type.
129         """
130         return f"{self.hub_api_url}/plugins/{plugin_type.value}/index"
131 
132     def plugin_endpoint(
133         self,
134         plugin_type: PluginType,
135         plugin_name: str,
136         variant_name: str | None = None,
137     ) -> str:
138         """Return the resource endpoint for the given plugin.
139 
140         Args:
141             plugin_type: The plugin type.
142             plugin_name: The plugin name.
143             variant_name: The plugin variant name.
144 
145         Returns:
146             The endpoint for the given plugin type.
147         """
148         url = f"{self.hub_api_url}/plugins/{plugin_type.value}/{plugin_name}"
149         if variant_name:
150             url = f"{url}--{variant_name}"
151 
152         return url
153 
154     def find_definition(
155         self,
156         plugin_type: PluginType,
157         plugin_name: str,
158         variant_name: str | None = None,
159     ) -> PluginDefinition:
160         """Find a locked plugin definition.
161 
162         Args:
163             plugin_type: The plugin type.
164             plugin_name: The plugin name.
165             variant_name: The plugin variant name.
166 
167         Returns:
168             The plugin definition.
169 
170         Raises:
171             PluginNotFoundError: If the plugin definition could not be found.
172             HubPluginVariantNotFound: If the plugin variant could not be found.
173         """
174         plugins = self.get_plugins_of_type(plugin_type)
175 
176         try:
177             plugin = plugins[plugin_name]
178         except KeyError as plugins_key_err:
179             raise PluginNotFoundError(
180                 PluginRef(plugin_type, plugin_name)
181             ) from plugins_key_err
182 
183         if variant_name is None or variant_name == Variant.DEFAULT_NAME:
184             variant_name = plugin.default_variant
185 
186         try:
187             url = plugin.variants[variant_name].ref
188         except KeyError as variant_key_err:
189             raise HubPluginVariantNotFound(
190                 plugin_type, plugin, variant_name
191             ) from variant_key_err
192 
193         response = self.session.get(url)
194 
195         try:
196             response.raise_for_status()
197         except requests.HTTPError as http_err:
198             logger.error(
199                 "Can not retrieve plugin",
200                 status_code=http_err.response.status_code,
201                 error=http_err,
202             )
203             raise PluginNotFoundError(PluginRef(plugin_type, plugin_name)) from http_err
204 
205         return PluginDefinition(**response.json(), plugin_type=plugin_type)
206 
207     def find_base_plugin(
208         self,
209         plugin_type: PluginType,
210         plugin_name: str,
211         variant: str | None = None,
212     ) -> BasePlugin:
213         """Get the base plugin for a project plugin.
214 
215         Args:
216             plugin_type: The plugin type.
217             plugin_name: The plugin name.
218             variant: The plugin variant.
219 
220         Returns:
221             The base plugin.
222         """
223         plugin = self.find_definition(
224             plugin_type,
225             plugin_name,
226             variant_name=variant,
227         )
228 
229         return base_plugin_factory(plugin, plugin.variants[0])
230 
231     def get_plugins_of_type(self, plugin_type: PluginType) -> dict[str, IndexedPlugin]:
232         """Get all plugins of a given type.
233 
234         Args:
235             plugin_type: The plugin type.
236 
237         Returns:
238             The plugin definitions.
239 
240         Raises:
241             HubPluginTypeNotFound: If the plugin type is not supported.
242         """
243         if not plugin_type.discoverable:
244             return {}
245 
246         url = self.plugin_type_endpoint(plugin_type)
247         response = self.session.get(url)
248 
249         try:
250             response.raise_for_status()
251         except requests.HTTPError as err:
252             logger.error(
253                 "Can not retrieve plugin type",
254                 status_code=err.response.status_code,
255                 error=err,
256             )
257             raise HubPluginTypeNotFound(plugin_type) from err
258 
259         plugins: dict[str, dict[str, Any]] = response.json()
260         return {
261             name: IndexedPlugin(
262                 name,
263                 logo_url=plugin["logo_url"],
264                 default_variant=plugin["default_variant"],
265                 variants={
266                     variant_name: VariantRef(variant_name, ref=variant["ref"])
267                     for variant_name, variant in plugin["variants"].items()
268                 },
269             )
270             for name, plugin in plugins.items()
271         }
272
```
--- 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/hub/client.py b/src/meltano/core/hub/client.py
--- a/src/meltano/core/hub/client.py
+++ b/src/meltano/core/hub/client.py
@@ -26,7 +26,7 @@
 logger = get_logger(__name__)
 
 
-class HubPluginTypeNotFound(Exception):
+class HubPluginTypeNotFoundError(Exception):
     """Raised when a Hub plugin type is not found."""
 
     def __init__(self, plugin_type: PluginType):
@@ -49,7 +49,7 @@
         )
 
 
-class HubPluginVariantNotFound(Exception):
+class HubPluginVariantNotFoundError(Exception):
     """Raised when a Hub plugin variant is not found."""
 
     def __init__(
@@ -169,7 +169,7 @@
 
         Raises:
             PluginNotFoundError: If the plugin definition could not be found.
-            HubPluginVariantNotFound: If the plugin variant could not be found.
+            HubPluginVariantNotFoundError: If the plugin variant could not be found.
         """
         plugins = self.get_plugins_of_type(plugin_type)
 
@@ -180,13 +180,16 @@
                 PluginRef(plugin_type, plugin_name)
             ) from plugins_key_err
 
-        if variant_name is None or variant_name == Variant.DEFAULT_NAME:
+        if variant_name is None or variant_name in {
+            Variant.DEFAULT_NAME,
+            Variant.ORIGINAL_NAME,
+        }:
             variant_name = plugin.default_variant
 
         try:
             url = plugin.variants[variant_name].ref
         except KeyError as variant_key_err:
-            raise HubPluginVariantNotFound(
+            raise HubPluginVariantNotFoundError(
                 plugin_type, plugin, variant_name
             ) from variant_key_err
 
@@ -238,7 +241,7 @@
             The plugin definitions.
 
         Raises:
-            HubPluginTypeNotFound: If the plugin type is not supported.
+            HubPluginTypeNotFoundError: If the plugin type is not supported.
         """
         if not plugin_type.discoverable:
             return {}
@@ -254,7 +257,7 @@
                 status_code=err.response.status_code,
                 error=err,
             )
-            raise HubPluginTypeNotFound(plugin_type) from err
+            raise HubPluginTypeNotFoundError(plugin_type) from err
 
         plugins: dict[str, dict[str, Any]] = response.json()
         return {

{"golden_diff": "diff --git a/src/meltano/core/hub/client.py b/src/meltano/core/hub/client.py\n--- a/src/meltano/core/hub/client.py\n+++ b/src/meltano/core/hub/client.py\n@@ -26,7 +26,7 @@\n logger = get_logger(__name__)\n \n \n-class HubPluginTypeNotFound(Exception):\n+class HubPluginTypeNotFoundError(Exception):\n     \"\"\"Raised when a Hub plugin type is not found.\"\"\"\n \n     def __init__(self, plugin_type: PluginType):\n@@ -49,7 +49,7 @@\n         )\n \n \n-class HubPluginVariantNotFound(Exception):\n+class HubPluginVariantNotFoundError(Exception):\n     \"\"\"Raised when a Hub plugin variant is not found.\"\"\"\n \n     def __init__(\n@@ -169,7 +169,7 @@\n \n         Raises:\n             PluginNotFoundError: If the plugin definition could not be found.\n-            HubPluginVariantNotFound: If the plugin variant could not be found.\n+            HubPluginVariantNotFoundError: If the plugin variant could not be found.\n         \"\"\"\n         plugins = self.get_plugins_of_type(plugin_type)\n \n@@ -180,13 +180,16 @@\n                 PluginRef(plugin_type, plugin_name)\n             ) from plugins_key_err\n \n-        if variant_name is None or variant_name == Variant.DEFAULT_NAME:\n+        if variant_name is None or variant_name in {\n+            Variant.DEFAULT_NAME,\n+            Variant.ORIGINAL_NAME,\n+        }:\n             variant_name = plugin.default_variant\n \n         try:\n             url = plugin.variants[variant_name].ref\n         except KeyError as variant_key_err:\n-            raise HubPluginVariantNotFound(\n+            raise HubPluginVariantNotFoundError(\n                 plugin_type, plugin, variant_name\n             ) from variant_key_err\n \n@@ -238,7 +241,7 @@\n             The plugin definitions.\n \n         Raises:\n-            HubPluginTypeNotFound: If the plugin type is not supported.\n+            HubPluginTypeNotFoundError: If the plugin type is not supported.\n         \"\"\"\n         if not plugin_type.discoverable:\n             return {}\n@@ -254,7 +257,7 @@\n                 status_code=err.response.status_code,\n                 error=err,\n             )\n-            raise HubPluginTypeNotFound(plugin_type) from err\n+            raise HubPluginTypeNotFoundError(plugin_type) from err\n \n         plugins: dict[str, dict[str, Any]] = response.json()\n         return {\n", "issue": "Meltano lock fails on migrating to >2.0.0 when variant didnt previously exist\nWe should update the log message and add a section to the migration docs to help users get variants added for plugins that didnt have a variant previously. For example Airflow locking fails because a variant isnt set for all installations prior to 2.0.0. The error message isnt totally clear in saying this and users will need to manually update their meltano.yml.\r\n\r\nOriginally discussed in https://github.com/meltano/meltano/issues/6359#issuecomment-1175432513\r\n\r\n> Additionally I'm getting Orchestrator 'airflow' variant 'original' is not known to Meltano. Variants: ['apache (default)'] from a default installation of Airflow. I know that its hard because when I originally installed it there wasnt a variant name for airflow but now there is. Is there a way for us to resolve that its the default variant? Maybe that wont work because the default could change, maybe using the discovery.yml when we think were migrating to >2.0.0?\r\n> \r\n> Or we could log it better and say something like \"I notice youre migrating and you dont have a variant, go to the Migration guide and learn how to define a variant\" and we write up the steps for locking post migration and adding a variant name:\r\n> \r\n> remove executable/namespace if defined\r\n> add variant using one available on MeltanoHub. We can list the mapping like airflow -> apache, dbt -> dbt-labs, etc. for plugins that just got variant names\r\n> run meltano lock --all\r\n\r\ncc @edgarrmondragon \n", "before_files": [{"content": "\"\"\"Meltano Hub Client.\"\"\"\n\nfrom __future__ import annotations\n\nfrom typing import Any\n\nimport requests\nfrom structlog.stdlib import get_logger\n\nimport meltano\nfrom meltano.core.plugin import (\n    BasePlugin,\n    PluginDefinition,\n    PluginRef,\n    PluginType,\n    Variant,\n)\nfrom meltano.core.plugin.error import PluginNotFoundError\nfrom meltano.core.plugin.factory import base_plugin_factory\nfrom meltano.core.plugin_discovery_service import PluginRepository\nfrom meltano.core.project import Project\nfrom meltano.core.project_settings_service import ProjectSettingsService\n\nfrom .schema import IndexedPlugin, VariantRef\n\nlogger = get_logger(__name__)\n\n\nclass HubPluginTypeNotFound(Exception):\n    \"\"\"Raised when a Hub plugin type is not found.\"\"\"\n\n    def __init__(self, plugin_type: PluginType):\n        \"\"\"Create a new HubPluginVariantNotFound.\n\n        Args:\n            plugin_type: The type of the plugin.\n        \"\"\"\n        self.plugin_type = plugin_type\n\n    def __str__(self) -> str:\n        \"\"\"Return a string representation of the error.\n\n        Returns:\n            The string representation of the error.\n        \"\"\"\n        return \"{type} is not supported in Meltano Hub. Available plugin types: {types}\".format(\n            type=self.plugin_type.descriptor.capitalize(),\n            types=list(PluginType),\n        )\n\n\nclass HubPluginVariantNotFound(Exception):\n    \"\"\"Raised when a Hub plugin variant is not found.\"\"\"\n\n    def __init__(\n        self,\n        plugin_type: PluginType,\n        plugin: IndexedPlugin,\n        variant_name: str,\n    ):\n        \"\"\"Create a new HubPluginVariantNotFound.\n\n        Args:\n            plugin_type: The type of the plugin.\n            plugin: The indexed plugin.\n            variant_name: The name of the variant that was not found.\n        \"\"\"\n        self.plugin_type = plugin_type\n        self.plugin = plugin\n        self.variant_name = variant_name\n\n    def __str__(self) -> str:\n        \"\"\"Return a string representation of the error.\n\n        Returns:\n            The string representation of the error.\n        \"\"\"\n        return \"{type} '{name}' variant '{variant}' is not known to Meltano. Variants: {variant_labels}\".format(\n            type=self.plugin_type.descriptor.capitalize(),\n            name=self.plugin.name,\n            variant=self.variant_name,\n            variant_labels=self.plugin.variant_labels,\n        )\n\n\nclass MeltanoHubService(PluginRepository):\n    \"\"\"PluginRepository implementation for the Meltano Hub.\"\"\"\n\n    def __init__(self, project: Project) -> None:\n        \"\"\"Initialize the service.\n\n        Args:\n            project: The Meltano project.\n        \"\"\"\n        self.project = project\n        self.session = requests.Session()\n        self.session.headers.update(\n            {\n                \"Accept\": \"application/json\",\n                \"User-Agent\": f\"Meltano/{meltano.__version__}\",\n            }\n        )\n\n        self.settings_service = ProjectSettingsService(self.project)\n\n        if self.settings_service.get(\"send_anonymous_usage_stats\"):\n            project_id = self.settings_service.get(\"project_id\")\n\n            self.session.headers[\"X-Project-ID\"] = project_id\n\n    @property\n    def hub_api_url(self):\n        \"\"\"Return the URL of the Hub API.\n\n        Returns:\n            The URL of the Hub API.\n        \"\"\"\n        hub_url = self.settings_service.get(\"hub_url\")\n        return f\"{hub_url}/meltano/api/v1\"\n\n    def plugin_type_endpoint(self, plugin_type: PluginType) -> str:\n        \"\"\"Return the list endpoint for the given plugin type.\n\n        Args:\n            plugin_type: The plugin type.\n\n        Returns:\n            The endpoint for the given plugin type.\n        \"\"\"\n        return f\"{self.hub_api_url}/plugins/{plugin_type.value}/index\"\n\n    def plugin_endpoint(\n        self,\n        plugin_type: PluginType,\n        plugin_name: str,\n        variant_name: str | None = None,\n    ) -> str:\n        \"\"\"Return the resource endpoint for the given plugin.\n\n        Args:\n            plugin_type: The plugin type.\n            plugin_name: The plugin name.\n            variant_name: The plugin variant name.\n\n        Returns:\n            The endpoint for the given plugin type.\n        \"\"\"\n        url = f\"{self.hub_api_url}/plugins/{plugin_type.value}/{plugin_name}\"\n        if variant_name:\n            url = f\"{url}--{variant_name}\"\n\n        return url\n\n    def find_definition(\n        self,\n        plugin_type: PluginType,\n        plugin_name: str,\n        variant_name: str | None = None,\n    ) -> PluginDefinition:\n        \"\"\"Find a locked plugin definition.\n\n        Args:\n            plugin_type: The plugin type.\n            plugin_name: The plugin name.\n            variant_name: The plugin variant name.\n\n        Returns:\n            The plugin definition.\n\n        Raises:\n            PluginNotFoundError: If the plugin definition could not be found.\n            HubPluginVariantNotFound: If the plugin variant could not be found.\n        \"\"\"\n        plugins = self.get_plugins_of_type(plugin_type)\n\n        try:\n            plugin = plugins[plugin_name]\n        except KeyError as plugins_key_err:\n            raise PluginNotFoundError(\n                PluginRef(plugin_type, plugin_name)\n            ) from plugins_key_err\n\n        if variant_name is None or variant_name == Variant.DEFAULT_NAME:\n            variant_name = plugin.default_variant\n\n        try:\n            url = plugin.variants[variant_name].ref\n        except KeyError as variant_key_err:\n            raise HubPluginVariantNotFound(\n                plugin_type, plugin, variant_name\n            ) from variant_key_err\n\n        response = self.session.get(url)\n\n        try:\n            response.raise_for_status()\n        except requests.HTTPError as http_err:\n            logger.error(\n                \"Can not retrieve plugin\",\n                status_code=http_err.response.status_code,\n                error=http_err,\n            )\n            raise PluginNotFoundError(PluginRef(plugin_type, plugin_name)) from http_err\n\n        return PluginDefinition(**response.json(), plugin_type=plugin_type)\n\n    def find_base_plugin(\n        self,\n        plugin_type: PluginType,\n        plugin_name: str,\n        variant: str | None = None,\n    ) -> BasePlugin:\n        \"\"\"Get the base plugin for a project plugin.\n\n        Args:\n            plugin_type: The plugin type.\n            plugin_name: The plugin name.\n            variant: The plugin variant.\n\n        Returns:\n            The base plugin.\n        \"\"\"\n        plugin = self.find_definition(\n            plugin_type,\n            plugin_name,\n            variant_name=variant,\n        )\n\n        return base_plugin_factory(plugin, plugin.variants[0])\n\n    def get_plugins_of_type(self, plugin_type: PluginType) -> dict[str, IndexedPlugin]:\n        \"\"\"Get all plugins of a given type.\n\n        Args:\n            plugin_type: The plugin type.\n\n        Returns:\n            The plugin definitions.\n\n        Raises:\n            HubPluginTypeNotFound: If the plugin type is not supported.\n        \"\"\"\n        if not plugin_type.discoverable:\n            return {}\n\n        url = self.plugin_type_endpoint(plugin_type)\n        response = self.session.get(url)\n\n        try:\n            response.raise_for_status()\n        except requests.HTTPError as err:\n            logger.error(\n                \"Can not retrieve plugin type\",\n                status_code=err.response.status_code,\n                error=err,\n            )\n            raise HubPluginTypeNotFound(plugin_type) from err\n\n        plugins: dict[str, dict[str, Any]] = response.json()\n        return {\n            name: IndexedPlugin(\n                name,\n                logo_url=plugin[\"logo_url\"],\n                default_variant=plugin[\"default_variant\"],\n                variants={\n                    variant_name: VariantRef(variant_name, ref=variant[\"ref\"])\n                    for variant_name, variant in plugin[\"variants\"].items()\n                },\n            )\n            for name, plugin in plugins.items()\n        }\n", "path": "src/meltano/core/hub/client.py"}], "after_files": [{"content": "\"\"\"Meltano Hub Client.\"\"\"\n\nfrom __future__ import annotations\n\nfrom typing import Any\n\nimport requests\nfrom structlog.stdlib import get_logger\n\nimport meltano\nfrom meltano.core.plugin import (\n    BasePlugin,\n    PluginDefinition,\n    PluginRef,\n    PluginType,\n    Variant,\n)\nfrom meltano.core.plugin.error import PluginNotFoundError\nfrom meltano.core.plugin.factory import base_plugin_factory\nfrom meltano.core.plugin_discovery_service import PluginRepository\nfrom meltano.core.project import Project\nfrom meltano.core.project_settings_service import ProjectSettingsService\n\nfrom .schema import IndexedPlugin, VariantRef\n\nlogger = get_logger(__name__)\n\n\nclass HubPluginTypeNotFoundError(Exception):\n    \"\"\"Raised when a Hub plugin type is not found.\"\"\"\n\n    def __init__(self, plugin_type: PluginType):\n        \"\"\"Create a new HubPluginVariantNotFound.\n\n        Args:\n            plugin_type: The type of the plugin.\n        \"\"\"\n        self.plugin_type = plugin_type\n\n    def __str__(self) -> str:\n        \"\"\"Return a string representation of the error.\n\n        Returns:\n            The string representation of the error.\n        \"\"\"\n        return \"{type} is not supported in Meltano Hub. Available plugin types: {types}\".format(\n            type=self.plugin_type.descriptor.capitalize(),\n            types=list(PluginType),\n        )\n\n\nclass HubPluginVariantNotFoundError(Exception):\n    \"\"\"Raised when a Hub plugin variant is not found.\"\"\"\n\n    def __init__(\n        self,\n        plugin_type: PluginType,\n        plugin: IndexedPlugin,\n        variant_name: str,\n    ):\n        \"\"\"Create a new HubPluginVariantNotFound.\n\n        Args:\n            plugin_type: The type of the plugin.\n            plugin: The indexed plugin.\n            variant_name: The name of the variant that was not found.\n        \"\"\"\n        self.plugin_type = plugin_type\n        self.plugin = plugin\n        self.variant_name = variant_name\n\n    def __str__(self) -> str:\n        \"\"\"Return a string representation of the error.\n\n        Returns:\n            The string representation of the error.\n        \"\"\"\n        return \"{type} '{name}' variant '{variant}' is not known to Meltano. Variants: {variant_labels}\".format(\n            type=self.plugin_type.descriptor.capitalize(),\n            name=self.plugin.name,\n            variant=self.variant_name,\n            variant_labels=self.plugin.variant_labels,\n        )\n\n\nclass MeltanoHubService(PluginRepository):\n    \"\"\"PluginRepository implementation for the Meltano Hub.\"\"\"\n\n    def __init__(self, project: Project) -> None:\n        \"\"\"Initialize the service.\n\n        Args:\n            project: The Meltano project.\n        \"\"\"\n        self.project = project\n        self.session = requests.Session()\n        self.session.headers.update(\n            {\n                \"Accept\": \"application/json\",\n                \"User-Agent\": f\"Meltano/{meltano.__version__}\",\n            }\n        )\n\n        self.settings_service = ProjectSettingsService(self.project)\n\n        if self.settings_service.get(\"send_anonymous_usage_stats\"):\n            project_id = self.settings_service.get(\"project_id\")\n\n            self.session.headers[\"X-Project-ID\"] = project_id\n\n    @property\n    def hub_api_url(self):\n        \"\"\"Return the URL of the Hub API.\n\n        Returns:\n            The URL of the Hub API.\n        \"\"\"\n        hub_url = self.settings_service.get(\"hub_url\")\n        return f\"{hub_url}/meltano/api/v1\"\n\n    def plugin_type_endpoint(self, plugin_type: PluginType) -> str:\n        \"\"\"Return the list endpoint for the given plugin type.\n\n        Args:\n            plugin_type: The plugin type.\n\n        Returns:\n            The endpoint for the given plugin type.\n        \"\"\"\n        return f\"{self.hub_api_url}/plugins/{plugin_type.value}/index\"\n\n    def plugin_endpoint(\n        self,\n        plugin_type: PluginType,\n        plugin_name: str,\n        variant_name: str | None = None,\n    ) -> str:\n        \"\"\"Return the resource endpoint for the given plugin.\n\n        Args:\n            plugin_type: The plugin type.\n            plugin_name: The plugin name.\n            variant_name: The plugin variant name.\n\n        Returns:\n            The endpoint for the given plugin type.\n        \"\"\"\n        url = f\"{self.hub_api_url}/plugins/{plugin_type.value}/{plugin_name}\"\n        if variant_name:\n            url = f\"{url}--{variant_name}\"\n\n        return url\n\n    def find_definition(\n        self,\n        plugin_type: PluginType,\n        plugin_name: str,\n        variant_name: str | None = None,\n    ) -> PluginDefinition:\n        \"\"\"Find a locked plugin definition.\n\n        Args:\n            plugin_type: The plugin type.\n            plugin_name: The plugin name.\n            variant_name: The plugin variant name.\n\n        Returns:\n            The plugin definition.\n\n        Raises:\n            PluginNotFoundError: If the plugin definition could not be found.\n            HubPluginVariantNotFoundError: If the plugin variant could not be found.\n        \"\"\"\n        plugins = self.get_plugins_of_type(plugin_type)\n\n        try:\n            plugin = plugins[plugin_name]\n        except KeyError as plugins_key_err:\n            raise PluginNotFoundError(\n                PluginRef(plugin_type, plugin_name)\n            ) from plugins_key_err\n\n        if variant_name is None or variant_name in {\n            Variant.DEFAULT_NAME,\n            Variant.ORIGINAL_NAME,\n        }:\n            variant_name = plugin.default_variant\n\n        try:\n            url = plugin.variants[variant_name].ref\n        except KeyError as variant_key_err:\n            raise HubPluginVariantNotFoundError(\n                plugin_type, plugin, variant_name\n            ) from variant_key_err\n\n        response = self.session.get(url)\n\n        try:\n            response.raise_for_status()\n        except requests.HTTPError as http_err:\n            logger.error(\n                \"Can not retrieve plugin\",\n                status_code=http_err.response.status_code,\n                error=http_err,\n            )\n            raise PluginNotFoundError(PluginRef(plugin_type, plugin_name)) from http_err\n\n        return PluginDefinition(**response.json(), plugin_type=plugin_type)\n\n    def find_base_plugin(\n        self,\n        plugin_type: PluginType,\n        plugin_name: str,\n        variant: str | None = None,\n    ) -> BasePlugin:\n        \"\"\"Get the base plugin for a project plugin.\n\n        Args:\n            plugin_type: The plugin type.\n            plugin_name: The plugin name.\n            variant: The plugin variant.\n\n        Returns:\n            The base plugin.\n        \"\"\"\n        plugin = self.find_definition(\n            plugin_type,\n            plugin_name,\n            variant_name=variant,\n        )\n\n        return base_plugin_factory(plugin, plugin.variants[0])\n\n    def get_plugins_of_type(self, plugin_type: PluginType) -> dict[str, IndexedPlugin]:\n        \"\"\"Get all plugins of a given type.\n\n        Args:\n            plugin_type: The plugin type.\n\n        Returns:\n            The plugin definitions.\n\n        Raises:\n            HubPluginTypeNotFoundError: If the plugin type is not supported.\n        \"\"\"\n        if not plugin_type.discoverable:\n            return {}\n\n        url = self.plugin_type_endpoint(plugin_type)\n        response = self.session.get(url)\n\n        try:\n            response.raise_for_status()\n        except requests.HTTPError as err:\n            logger.error(\n                \"Can not retrieve plugin type\",\n                status_code=err.response.status_code,\n                error=err,\n            )\n            raise HubPluginTypeNotFoundError(plugin_type) from err\n\n        plugins: dict[str, dict[str, Any]] = response.json()\n        return {\n            name: IndexedPlugin(\n                name,\n                logo_url=plugin[\"logo_url\"],\n                default_variant=plugin[\"default_variant\"],\n                variants={\n                    variant_name: VariantRef(variant_name, ref=variant[\"ref\"])\n                    for variant_name, variant in plugin[\"variants\"].items()\n                },\n            )\n            for name, plugin in plugins.items()\n        }\n", "path": "src/meltano/core/hub/client.py"}]}