We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
RFE: Replace use of `pkg_resources` with `importlib.metadata`
See discussions:
[astropy/astropy#11091](https://github.com/astropy/astropy/pull/11091)
[pypa/pip#7413](https://github.com/pypa/pip/issues/7413)

```console
[tkloczko@devel-g2v lexicon-3.11.0]$ grep -r pkg_resources
lexicon/discovery.py:import pkg_resources
lexicon/discovery.py:        distribution = pkg_resources.get_distribution("dns-lexicon")
lexicon/discovery.py:    except pkg_resources.DistributionNotFound:
lexicon/discovery.py:        return pkg_resources.get_distribution("dns-lexicon").version
lexicon/discovery.py:    except pkg_resources.DistributionNotFound:
lexicon/discovery.py:    provider: str, distribution: pkg_resources.Distribution
lexicon/discovery.py:        requirements: List[pkg_resources.Requirement] = distribution.requires(
lexicon/discovery.py:    except pkg_resources.UnknownExtra:
lexicon/discovery.py:                    pkg_resources.get_distribution(requirement.name)  # type: ignore
lexicon/discovery.py:                    pkg_resources.get_distribution(requirement)
lexicon/discovery.py:        except (pkg_resources.DistributionNotFound, pkg_resources.VersionConflict):
t:| * 90642ca - (origin/fix-provider-disco-legacy, jamin/fix-provider-disco-legacy, giuse/fix-provider-disco-legacy) Support old versions of pkg_resources (2 years, 10 months ago) <Adrien Ferrand>
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `lexicon/client.py`
Content:
```
1 """Main module of Lexicon. Defines the Client class, that holds all Lexicon logic."""
2 import importlib
3 import logging
4 import os
5 from typing import Dict, List, Optional, Type, Union, cast
6 
7 import tldextract  # type: ignore
8 
9 from lexicon import config as helper_config
10 from lexicon import discovery
11 from lexicon.exceptions import ProviderNotAvailableError
12 from lexicon.providers.base import Provider
13 
14 
15 class Client(object):
16     """This is the Lexicon client, that will execute all the logic."""
17 
18     def __init__(
19         self, config: Optional[Union[helper_config.ConfigResolver, Dict]] = None
20     ):
21         if not config:
22             # If there is not config specified, we load a non-interactive configuration.
23             self.config = helper_config.non_interactive_config_resolver()
24         elif not isinstance(config, helper_config.ConfigResolver):
25             # If config is not a ConfigResolver, we are in a legacy situation.
26             # We protect this part of the Client API.
27             self.config = helper_config.legacy_config_resolver(config)
28         else:
29             self.config = config
30 
31         # Validate configuration
32         self._validate_config()
33 
34         runtime_config = {}
35 
36         # Process domain, strip subdomain
37         try:
38             domain_extractor = tldextract.TLDExtract(
39                 cache_dir=_get_tldextract_cache_path(), include_psl_private_domains=True
40             )
41         except TypeError:
42             domain_extractor = tldextract.TLDExtract(
43                 cache_file=_get_tldextract_cache_path(), include_psl_private_domains=True  # type: ignore
44             )
45         domain_parts = domain_extractor(
46             cast(str, self.config.resolve("lexicon:domain"))
47         )
48         runtime_config["domain"] = f"{domain_parts.domain}.{domain_parts.suffix}"
49 
50         delegated = self.config.resolve("lexicon:delegated")
51         if delegated:
52             # handle delegated domain
53             delegated = str(delegated).rstrip(".")
54             initial_domain = str(runtime_config.get("domain"))
55             if delegated != initial_domain:
56                 # convert to relative name
57                 if delegated.endswith(initial_domain):
58                     delegated = delegated[: -len(initial_domain)]
59                     delegated = delegated.rstrip(".")
60                 # update domain
61                 runtime_config["domain"] = f"{delegated}.{initial_domain}"
62 
63         self.action = self.config.resolve("lexicon:action")
64         self.provider_name = self.config.resolve(
65             "lexicon:provider_name"
66         ) or self.config.resolve("lexicon:provider")
67 
68         if not self.provider_name:
69             raise ValueError("Could not resolve provider name.")
70 
71         self.config.add_config_source(helper_config.DictConfigSource(runtime_config), 0)
72 
73         provider_module = importlib.import_module(
74             "lexicon.providers." + self.provider_name
75         )
76         provider_class: Type[Provider] = getattr(provider_module, "Provider")
77         self.provider = provider_class(self.config)
78 
79     def execute(self) -> Union[bool, List[Dict]]:
80         """Execute provided configuration in class constructor to the DNS records"""
81         self.provider.authenticate()
82         identifier = self.config.resolve("lexicon:identifier")
83         record_type = self.config.resolve("lexicon:type")
84         name = self.config.resolve("lexicon:name")
85         content = self.config.resolve("lexicon:content")
86 
87         if self.action == "create":
88             if not record_type or not name or not content:
89                 raise ValueError("Missing record_type, name or content parameters.")
90             return self.provider.create_record(record_type, name, content)
91 
92         if self.action == "list":
93             return self.provider.list_records(record_type, name, content)
94 
95         if self.action == "update":
96             return self.provider.update_record(identifier, record_type, name, content)
97 
98         if self.action == "delete":
99             return self.provider.delete_record(identifier, record_type, name, content)
100 
101         raise ValueError(f"Invalid action statement: {self.action}")
102 
103     def _validate_config(self) -> None:
104         provider_name = self.config.resolve("lexicon:provider_name")
105         if not provider_name:
106             raise AttributeError("provider_name")
107 
108         try:
109             available = discovery.find_providers()[provider_name]
110         except KeyError:
111             raise ProviderNotAvailableError(
112                 f"This provider ({provider_name}) is not supported by Lexicon."
113             )
114         else:
115             if not available:
116                 raise ProviderNotAvailableError(
117                     f"This provider ({provider_name}) has required dependencies that are missing. "
118                     f"Please install lexicon[{provider_name}] first."
119                 )
120 
121         if not self.config.resolve("lexicon:action"):
122             raise AttributeError("action")
123         if not self.config.resolve("lexicon:domain"):
124             raise AttributeError("domain")
125         if not self.config.resolve("lexicon:type"):
126             raise AttributeError("type")
127 
128 
129 def _get_tldextract_cache_path() -> str:
130     if os.environ.get("TLDEXTRACT_CACHE_FILE"):
131         logging.warning(
132             "TLD_EXTRACT_CACHE_FILE environment variable is deprecated, please use TLDEXTRACT_CACHE_PATH instead."
133         )
134         os.environ["TLDEXTRACT_CACHE_PATH"] = os.environ["TLDEXTRACT_CACHE_FILE"]
135 
136     return os.path.expanduser(
137         os.environ.get("TLDEXTRACT_CACHE_PATH", os.path.join("~", ".lexicon_tld_set"))
138     )
139
```

Path: `lexicon/parser.py`
Content:
```
1 """Parsers definition for the Lexicon command-line interface"""
2 import argparse
3 import importlib
4 import os
5 
6 from lexicon import discovery
7 
8 
9 def generate_base_provider_parser() -> argparse.ArgumentParser:
10     """Function that generates the base provider to be used by all dns providers."""
11     parser = argparse.ArgumentParser(add_help=False)
12     parser.add_argument(
13         "action",
14         help="specify the action to take",
15         default="list",
16         choices=["create", "list", "update", "delete"],
17     )
18     parser.add_argument(
19         "domain", help="specify the domain, supports subdomains as well"
20     )
21     parser.add_argument(
22         "type",
23         help="specify the entry type",
24         default="TXT",
25         choices=["A", "AAAA", "CNAME", "MX", "NS", "SOA", "TXT", "SRV", "LOC"],
26     )
27 
28     parser.add_argument("--name", help="specify the record name")
29     parser.add_argument("--content", help="specify the record content")
30     parser.add_argument("--ttl", type=int, help="specify the record time-to-live")
31     parser.add_argument("--priority", help="specify the record priority")
32     parser.add_argument(
33         "--identifier", help="specify the record for update or delete actions"
34     )
35     parser.add_argument(
36         "--log_level",
37         help="specify the log level",
38         default="ERROR",
39         choices=["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG", "NOTSET"],
40     )
41     parser.add_argument(
42         "--output",
43         help=(
44             "specify the type of output: by default a formatted table (TABLE), "
45             "a formatted table without header (TABLE-NO-HEADER), "
46             "a JSON string (JSON) or no output (QUIET)"
47         ),
48         default="TABLE",
49         choices=["TABLE", "TABLE-NO-HEADER", "JSON", "QUIET"],
50     )
51     return parser
52 
53 
54 def generate_cli_main_parser() -> argparse.ArgumentParser:
55     """Using all providers available, generate a parser that will be used by Lexicon CLI"""
56     parser = argparse.ArgumentParser(
57         description="Create, Update, Delete, List DNS entries"
58     )
59 
60     parser.add_argument(
61         "--version",
62         help="show the current version of lexicon",
63         action="version",
64         version=f"%(prog)s {discovery.lexicon_version()}",
65     )
66     parser.add_argument("--delegated", help="specify the delegated domain")
67     parser.add_argument(
68         "--config-dir",
69         default=os.getcwd(),
70         help="specify the directory where to search lexicon.yml and "
71         "lexicon_[provider].yml configuration files "
72         "(default: current directory).",
73     )
74     subparsers = parser.add_subparsers(
75         dest="provider_name", help="specify the DNS provider to use"
76     )
77     subparsers.required = True
78 
79     for provider, available in discovery.find_providers().items():
80         provider_module = importlib.import_module("lexicon.providers." + provider)
81         provider_parser = getattr(provider_module, "provider_parser")
82 
83         subparser = subparsers.add_parser(
84             provider,
85             help=f"{provider} provider",
86             parents=[generate_base_provider_parser()],
87         )
88         provider_parser(subparser)
89 
90         if not available:
91             subparser.epilog = (
92                 "WARNING: some required dependencies for this provider are not "
93                 f"installed. Please install lexicon[{provider}] first before using it."
94             )
95 
96     return parser
97
```

Path: `lexicon/discovery.py`
Content:
```
1 """
2 This module takes care of finding information about the runtime of Lexicon:
3 * what are the providers installed, and available
4 * what is the version of Lexicon
5 """
6 import pkgutil
7 from typing import Dict, List
8 
9 import pkg_resources
10 
11 from lexicon import providers
12 
13 
14 def find_providers() -> Dict[str, bool]:
15     """Find all providers registered in Lexicon, and their availability"""
16     providers_list = sorted(
17         {
18             modname
19             for (_, modname, _) in pkgutil.iter_modules(providers.__path__)  # type: ignore
20             if modname != "base"
21         }
22     )
23 
24     try:
25         distribution = pkg_resources.get_distribution("dns-lexicon")
26     except pkg_resources.DistributionNotFound:
27         return {provider: True for provider in providers_list}
28     else:
29         return {
30             provider: _resolve_requirements(provider, distribution)
31             for provider in providers_list
32         }
33 
34 
35 def lexicon_version() -> str:
36     """Retrieve current Lexicon version"""
37     try:
38         return pkg_resources.get_distribution("dns-lexicon").version
39     except pkg_resources.DistributionNotFound:
40         return "unknown"
41 
42 
43 def _resolve_requirements(
44     provider: str, distribution: pkg_resources.Distribution
45 ) -> bool:
46     try:
47         requirements: List[pkg_resources.Requirement] = distribution.requires(
48             extras=(provider,)
49         )
50     except pkg_resources.UnknownExtra:
51         # No extra for this provider
52         return True
53     else:
54         # Extra is defined
55         try:
56             for requirement in requirements:
57                 if hasattr(requirement, "name"):
58                     pkg_resources.get_distribution(requirement.name)  # type: ignore
59                 else:
60                     pkg_resources.get_distribution(requirement)
61         except (pkg_resources.DistributionNotFound, pkg_resources.VersionConflict):
62             # At least one extra requirement is not fulfilled
63             return False
64 
65     return True
66
```
--- 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/lexicon/client.py b/lexicon/client.py
--- a/lexicon/client.py
+++ b/lexicon/client.py
@@ -114,8 +114,8 @@
         else:
             if not available:
                 raise ProviderNotAvailableError(
-                    f"This provider ({provider_name}) has required dependencies that are missing. "
-                    f"Please install lexicon[{provider_name}] first."
+                    f"This provider ({provider_name}) has required extra dependencies that are missing. "
+                    f"Please run `pip install lexicon[{provider_name}]` first before using it."
                 )
 
         if not self.config.resolve("lexicon:action"):
diff --git a/lexicon/discovery.py b/lexicon/discovery.py
--- a/lexicon/discovery.py
+++ b/lexicon/discovery.py
@@ -4,9 +4,13 @@
 * what is the version of Lexicon
 """
 import pkgutil
-from typing import Dict, List
+import re
+from typing import Dict
 
-import pkg_resources
+try:
+    from importlib.metadata import Distribution, PackageNotFoundError
+except ModuleNotFoundError:
+    from importlib_metadata import Distribution, PackageNotFoundError  # type: ignore[misc]
 
 from lexicon import providers
 
@@ -22,8 +26,8 @@
     )
 
     try:
-        distribution = pkg_resources.get_distribution("dns-lexicon")
-    except pkg_resources.DistributionNotFound:
+        distribution = Distribution.from_name("dns-lexicon")
+    except PackageNotFoundError:
         return {provider: True for provider in providers_list}
     else:
         return {
@@ -35,31 +39,32 @@
 def lexicon_version() -> str:
     """Retrieve current Lexicon version"""
     try:
-        return pkg_resources.get_distribution("dns-lexicon").version
-    except pkg_resources.DistributionNotFound:
+        return Distribution.from_name("dns-lexicon").version
+    except PackageNotFoundError:
         return "unknown"
 
 
-def _resolve_requirements(
-    provider: str, distribution: pkg_resources.Distribution
-) -> bool:
-    try:
-        requirements: List[pkg_resources.Requirement] = distribution.requires(
-            extras=(provider,)
-        )
-    except pkg_resources.UnknownExtra:
+def _resolve_requirements(provider: str, distribution: Distribution) -> bool:
+    requires = distribution.requires
+    if requires is None:
+        raise ValueError("Error while trying finding requirements.")
+
+    requirements = [
+        re.sub(r"^(.*)\s\(.*\)(?:;.*|)$", r"\1", requirement)
+        for requirement in requires
+        if f'extra == "{provider}"' in requirement
+    ]
+
+    if not requirements:
         # No extra for this provider
         return True
-    else:
-        # Extra is defined
+
+    for requirement in requirements:
         try:
-            for requirement in requirements:
-                if hasattr(requirement, "name"):
-                    pkg_resources.get_distribution(requirement.name)  # type: ignore
-                else:
-                    pkg_resources.get_distribution(requirement)
-        except (pkg_resources.DistributionNotFound, pkg_resources.VersionConflict):
+            Distribution.from_name(requirement)
+        except PackageNotFoundError:
             # At least one extra requirement is not fulfilled
             return False
 
+    # All extra requirements are fulfilled
     return True
diff --git a/lexicon/parser.py b/lexicon/parser.py
--- a/lexicon/parser.py
+++ b/lexicon/parser.py
@@ -90,7 +90,7 @@
         if not available:
             subparser.epilog = (
                 "WARNING: some required dependencies for this provider are not "
-                f"installed. Please install lexicon[{provider}] first before using it."
+                f"installed. Please run `pip install lexicon[{provider}]` first before using it."
             )
 
     return parser

{"golden_diff": "diff --git a/lexicon/client.py b/lexicon/client.py\n--- a/lexicon/client.py\n+++ b/lexicon/client.py\n@@ -114,8 +114,8 @@\n         else:\n             if not available:\n                 raise ProviderNotAvailableError(\n-                    f\"This provider ({provider_name}) has required dependencies that are missing. \"\n-                    f\"Please install lexicon[{provider_name}] first.\"\n+                    f\"This provider ({provider_name}) has required extra dependencies that are missing. \"\n+                    f\"Please run `pip install lexicon[{provider_name}]` first before using it.\"\n                 )\n \n         if not self.config.resolve(\"lexicon:action\"):\ndiff --git a/lexicon/discovery.py b/lexicon/discovery.py\n--- a/lexicon/discovery.py\n+++ b/lexicon/discovery.py\n@@ -4,9 +4,13 @@\n * what is the version of Lexicon\n \"\"\"\n import pkgutil\n-from typing import Dict, List\n+import re\n+from typing import Dict\n \n-import pkg_resources\n+try:\n+    from importlib.metadata import Distribution, PackageNotFoundError\n+except ModuleNotFoundError:\n+    from importlib_metadata import Distribution, PackageNotFoundError  # type: ignore[misc]\n \n from lexicon import providers\n \n@@ -22,8 +26,8 @@\n     )\n \n     try:\n-        distribution = pkg_resources.get_distribution(\"dns-lexicon\")\n-    except pkg_resources.DistributionNotFound:\n+        distribution = Distribution.from_name(\"dns-lexicon\")\n+    except PackageNotFoundError:\n         return {provider: True for provider in providers_list}\n     else:\n         return {\n@@ -35,31 +39,32 @@\n def lexicon_version() -> str:\n     \"\"\"Retrieve current Lexicon version\"\"\"\n     try:\n-        return pkg_resources.get_distribution(\"dns-lexicon\").version\n-    except pkg_resources.DistributionNotFound:\n+        return Distribution.from_name(\"dns-lexicon\").version\n+    except PackageNotFoundError:\n         return \"unknown\"\n \n \n-def _resolve_requirements(\n-    provider: str, distribution: pkg_resources.Distribution\n-) -> bool:\n-    try:\n-        requirements: List[pkg_resources.Requirement] = distribution.requires(\n-            extras=(provider,)\n-        )\n-    except pkg_resources.UnknownExtra:\n+def _resolve_requirements(provider: str, distribution: Distribution) -> bool:\n+    requires = distribution.requires\n+    if requires is None:\n+        raise ValueError(\"Error while trying finding requirements.\")\n+\n+    requirements = [\n+        re.sub(r\"^(.*)\\s\\(.*\\)(?:;.*|)$\", r\"\\1\", requirement)\n+        for requirement in requires\n+        if f'extra == \"{provider}\"' in requirement\n+    ]\n+\n+    if not requirements:\n         # No extra for this provider\n         return True\n-    else:\n-        # Extra is defined\n+\n+    for requirement in requirements:\n         try:\n-            for requirement in requirements:\n-                if hasattr(requirement, \"name\"):\n-                    pkg_resources.get_distribution(requirement.name)  # type: ignore\n-                else:\n-                    pkg_resources.get_distribution(requirement)\n-        except (pkg_resources.DistributionNotFound, pkg_resources.VersionConflict):\n+            Distribution.from_name(requirement)\n+        except PackageNotFoundError:\n             # At least one extra requirement is not fulfilled\n             return False\n \n+    # All extra requirements are fulfilled\n     return True\ndiff --git a/lexicon/parser.py b/lexicon/parser.py\n--- a/lexicon/parser.py\n+++ b/lexicon/parser.py\n@@ -90,7 +90,7 @@\n         if not available:\n             subparser.epilog = (\n                 \"WARNING: some required dependencies for this provider are not \"\n-                f\"installed. Please install lexicon[{provider}] first before using it.\"\n+                f\"installed. Please run `pip install lexicon[{provider}]` first before using it.\"\n             )\n \n     return parser\n", "issue": "RFE: Replace use of `pkg_resources` with `importlib.metadata`\nSee discussions:\r\n[astropy/astropy#11091](https://github.com/astropy/astropy/pull/11091)\r\n[pypa/pip#7413](https://github.com/pypa/pip/issues/7413)\r\n\r\n```console\r\n[tkloczko@devel-g2v lexicon-3.11.0]$ grep -r pkg_resources\r\nlexicon/discovery.py:import pkg_resources\r\nlexicon/discovery.py:        distribution = pkg_resources.get_distribution(\"dns-lexicon\")\r\nlexicon/discovery.py:    except pkg_resources.DistributionNotFound:\r\nlexicon/discovery.py:        return pkg_resources.get_distribution(\"dns-lexicon\").version\r\nlexicon/discovery.py:    except pkg_resources.DistributionNotFound:\r\nlexicon/discovery.py:    provider: str, distribution: pkg_resources.Distribution\r\nlexicon/discovery.py:        requirements: List[pkg_resources.Requirement] = distribution.requires(\r\nlexicon/discovery.py:    except pkg_resources.UnknownExtra:\r\nlexicon/discovery.py:                    pkg_resources.get_distribution(requirement.name)  # type: ignore\r\nlexicon/discovery.py:                    pkg_resources.get_distribution(requirement)\r\nlexicon/discovery.py:        except (pkg_resources.DistributionNotFound, pkg_resources.VersionConflict):\r\nt:| * 90642ca - (origin/fix-provider-disco-legacy, jamin/fix-provider-disco-legacy, giuse/fix-provider-disco-legacy) Support old versions of pkg_resources (2 years, 10 months ago) <Adrien Ferrand>\r\n```\r\n\n", "before_files": [{"content": "\"\"\"Main module of Lexicon. Defines the Client class, that holds all Lexicon logic.\"\"\"\nimport importlib\nimport logging\nimport os\nfrom typing import Dict, List, Optional, Type, Union, cast\n\nimport tldextract  # type: ignore\n\nfrom lexicon import config as helper_config\nfrom lexicon import discovery\nfrom lexicon.exceptions import ProviderNotAvailableError\nfrom lexicon.providers.base import Provider\n\n\nclass Client(object):\n    \"\"\"This is the Lexicon client, that will execute all the logic.\"\"\"\n\n    def __init__(\n        self, config: Optional[Union[helper_config.ConfigResolver, Dict]] = None\n    ):\n        if not config:\n            # If there is not config specified, we load a non-interactive configuration.\n            self.config = helper_config.non_interactive_config_resolver()\n        elif not isinstance(config, helper_config.ConfigResolver):\n            # If config is not a ConfigResolver, we are in a legacy situation.\n            # We protect this part of the Client API.\n            self.config = helper_config.legacy_config_resolver(config)\n        else:\n            self.config = config\n\n        # Validate configuration\n        self._validate_config()\n\n        runtime_config = {}\n\n        # Process domain, strip subdomain\n        try:\n            domain_extractor = tldextract.TLDExtract(\n                cache_dir=_get_tldextract_cache_path(), include_psl_private_domains=True\n            )\n        except TypeError:\n            domain_extractor = tldextract.TLDExtract(\n                cache_file=_get_tldextract_cache_path(), include_psl_private_domains=True  # type: ignore\n            )\n        domain_parts = domain_extractor(\n            cast(str, self.config.resolve(\"lexicon:domain\"))\n        )\n        runtime_config[\"domain\"] = f\"{domain_parts.domain}.{domain_parts.suffix}\"\n\n        delegated = self.config.resolve(\"lexicon:delegated\")\n        if delegated:\n            # handle delegated domain\n            delegated = str(delegated).rstrip(\".\")\n            initial_domain = str(runtime_config.get(\"domain\"))\n            if delegated != initial_domain:\n                # convert to relative name\n                if delegated.endswith(initial_domain):\n                    delegated = delegated[: -len(initial_domain)]\n                    delegated = delegated.rstrip(\".\")\n                # update domain\n                runtime_config[\"domain\"] = f\"{delegated}.{initial_domain}\"\n\n        self.action = self.config.resolve(\"lexicon:action\")\n        self.provider_name = self.config.resolve(\n            \"lexicon:provider_name\"\n        ) or self.config.resolve(\"lexicon:provider\")\n\n        if not self.provider_name:\n            raise ValueError(\"Could not resolve provider name.\")\n\n        self.config.add_config_source(helper_config.DictConfigSource(runtime_config), 0)\n\n        provider_module = importlib.import_module(\n            \"lexicon.providers.\" + self.provider_name\n        )\n        provider_class: Type[Provider] = getattr(provider_module, \"Provider\")\n        self.provider = provider_class(self.config)\n\n    def execute(self) -> Union[bool, List[Dict]]:\n        \"\"\"Execute provided configuration in class constructor to the DNS records\"\"\"\n        self.provider.authenticate()\n        identifier = self.config.resolve(\"lexicon:identifier\")\n        record_type = self.config.resolve(\"lexicon:type\")\n        name = self.config.resolve(\"lexicon:name\")\n        content = self.config.resolve(\"lexicon:content\")\n\n        if self.action == \"create\":\n            if not record_type or not name or not content:\n                raise ValueError(\"Missing record_type, name or content parameters.\")\n            return self.provider.create_record(record_type, name, content)\n\n        if self.action == \"list\":\n            return self.provider.list_records(record_type, name, content)\n\n        if self.action == \"update\":\n            return self.provider.update_record(identifier, record_type, name, content)\n\n        if self.action == \"delete\":\n            return self.provider.delete_record(identifier, record_type, name, content)\n\n        raise ValueError(f\"Invalid action statement: {self.action}\")\n\n    def _validate_config(self) -> None:\n        provider_name = self.config.resolve(\"lexicon:provider_name\")\n        if not provider_name:\n            raise AttributeError(\"provider_name\")\n\n        try:\n            available = discovery.find_providers()[provider_name]\n        except KeyError:\n            raise ProviderNotAvailableError(\n                f\"This provider ({provider_name}) is not supported by Lexicon.\"\n            )\n        else:\n            if not available:\n                raise ProviderNotAvailableError(\n                    f\"This provider ({provider_name}) has required dependencies that are missing. \"\n                    f\"Please install lexicon[{provider_name}] first.\"\n                )\n\n        if not self.config.resolve(\"lexicon:action\"):\n            raise AttributeError(\"action\")\n        if not self.config.resolve(\"lexicon:domain\"):\n            raise AttributeError(\"domain\")\n        if not self.config.resolve(\"lexicon:type\"):\n            raise AttributeError(\"type\")\n\n\ndef _get_tldextract_cache_path() -> str:\n    if os.environ.get(\"TLDEXTRACT_CACHE_FILE\"):\n        logging.warning(\n            \"TLD_EXTRACT_CACHE_FILE environment variable is deprecated, please use TLDEXTRACT_CACHE_PATH instead.\"\n        )\n        os.environ[\"TLDEXTRACT_CACHE_PATH\"] = os.environ[\"TLDEXTRACT_CACHE_FILE\"]\n\n    return os.path.expanduser(\n        os.environ.get(\"TLDEXTRACT_CACHE_PATH\", os.path.join(\"~\", \".lexicon_tld_set\"))\n    )\n", "path": "lexicon/client.py"}, {"content": "\"\"\"Parsers definition for the Lexicon command-line interface\"\"\"\nimport argparse\nimport importlib\nimport os\n\nfrom lexicon import discovery\n\n\ndef generate_base_provider_parser() -> argparse.ArgumentParser:\n    \"\"\"Function that generates the base provider to be used by all dns providers.\"\"\"\n    parser = argparse.ArgumentParser(add_help=False)\n    parser.add_argument(\n        \"action\",\n        help=\"specify the action to take\",\n        default=\"list\",\n        choices=[\"create\", \"list\", \"update\", \"delete\"],\n    )\n    parser.add_argument(\n        \"domain\", help=\"specify the domain, supports subdomains as well\"\n    )\n    parser.add_argument(\n        \"type\",\n        help=\"specify the entry type\",\n        default=\"TXT\",\n        choices=[\"A\", \"AAAA\", \"CNAME\", \"MX\", \"NS\", \"SOA\", \"TXT\", \"SRV\", \"LOC\"],\n    )\n\n    parser.add_argument(\"--name\", help=\"specify the record name\")\n    parser.add_argument(\"--content\", help=\"specify the record content\")\n    parser.add_argument(\"--ttl\", type=int, help=\"specify the record time-to-live\")\n    parser.add_argument(\"--priority\", help=\"specify the record priority\")\n    parser.add_argument(\n        \"--identifier\", help=\"specify the record for update or delete actions\"\n    )\n    parser.add_argument(\n        \"--log_level\",\n        help=\"specify the log level\",\n        default=\"ERROR\",\n        choices=[\"CRITICAL\", \"ERROR\", \"WARNING\", \"INFO\", \"DEBUG\", \"NOTSET\"],\n    )\n    parser.add_argument(\n        \"--output\",\n        help=(\n            \"specify the type of output: by default a formatted table (TABLE), \"\n            \"a formatted table without header (TABLE-NO-HEADER), \"\n            \"a JSON string (JSON) or no output (QUIET)\"\n        ),\n        default=\"TABLE\",\n        choices=[\"TABLE\", \"TABLE-NO-HEADER\", \"JSON\", \"QUIET\"],\n    )\n    return parser\n\n\ndef generate_cli_main_parser() -> argparse.ArgumentParser:\n    \"\"\"Using all providers available, generate a parser that will be used by Lexicon CLI\"\"\"\n    parser = argparse.ArgumentParser(\n        description=\"Create, Update, Delete, List DNS entries\"\n    )\n\n    parser.add_argument(\n        \"--version\",\n        help=\"show the current version of lexicon\",\n        action=\"version\",\n        version=f\"%(prog)s {discovery.lexicon_version()}\",\n    )\n    parser.add_argument(\"--delegated\", help=\"specify the delegated domain\")\n    parser.add_argument(\n        \"--config-dir\",\n        default=os.getcwd(),\n        help=\"specify the directory where to search lexicon.yml and \"\n        \"lexicon_[provider].yml configuration files \"\n        \"(default: current directory).\",\n    )\n    subparsers = parser.add_subparsers(\n        dest=\"provider_name\", help=\"specify the DNS provider to use\"\n    )\n    subparsers.required = True\n\n    for provider, available in discovery.find_providers().items():\n        provider_module = importlib.import_module(\"lexicon.providers.\" + provider)\n        provider_parser = getattr(provider_module, \"provider_parser\")\n\n        subparser = subparsers.add_parser(\n            provider,\n            help=f\"{provider} provider\",\n            parents=[generate_base_provider_parser()],\n        )\n        provider_parser(subparser)\n\n        if not available:\n            subparser.epilog = (\n                \"WARNING: some required dependencies for this provider are not \"\n                f\"installed. Please install lexicon[{provider}] first before using it.\"\n            )\n\n    return parser\n", "path": "lexicon/parser.py"}, {"content": "\"\"\"\nThis module takes care of finding information about the runtime of Lexicon:\n* what are the providers installed, and available\n* what is the version of Lexicon\n\"\"\"\nimport pkgutil\nfrom typing import Dict, List\n\nimport pkg_resources\n\nfrom lexicon import providers\n\n\ndef find_providers() -> Dict[str, bool]:\n    \"\"\"Find all providers registered in Lexicon, and their availability\"\"\"\n    providers_list = sorted(\n        {\n            modname\n            for (_, modname, _) in pkgutil.iter_modules(providers.__path__)  # type: ignore\n            if modname != \"base\"\n        }\n    )\n\n    try:\n        distribution = pkg_resources.get_distribution(\"dns-lexicon\")\n    except pkg_resources.DistributionNotFound:\n        return {provider: True for provider in providers_list}\n    else:\n        return {\n            provider: _resolve_requirements(provider, distribution)\n            for provider in providers_list\n        }\n\n\ndef lexicon_version() -> str:\n    \"\"\"Retrieve current Lexicon version\"\"\"\n    try:\n        return pkg_resources.get_distribution(\"dns-lexicon\").version\n    except pkg_resources.DistributionNotFound:\n        return \"unknown\"\n\n\ndef _resolve_requirements(\n    provider: str, distribution: pkg_resources.Distribution\n) -> bool:\n    try:\n        requirements: List[pkg_resources.Requirement] = distribution.requires(\n            extras=(provider,)\n        )\n    except pkg_resources.UnknownExtra:\n        # No extra for this provider\n        return True\n    else:\n        # Extra is defined\n        try:\n            for requirement in requirements:\n                if hasattr(requirement, \"name\"):\n                    pkg_resources.get_distribution(requirement.name)  # type: ignore\n                else:\n                    pkg_resources.get_distribution(requirement)\n        except (pkg_resources.DistributionNotFound, pkg_resources.VersionConflict):\n            # At least one extra requirement is not fulfilled\n            return False\n\n    return True\n", "path": "lexicon/discovery.py"}], "after_files": [{"content": "\"\"\"Main module of Lexicon. Defines the Client class, that holds all Lexicon logic.\"\"\"\nimport importlib\nimport logging\nimport os\nfrom typing import Dict, List, Optional, Type, Union, cast\n\nimport tldextract  # type: ignore\n\nfrom lexicon import config as helper_config\nfrom lexicon import discovery\nfrom lexicon.exceptions import ProviderNotAvailableError\nfrom lexicon.providers.base import Provider\n\n\nclass Client(object):\n    \"\"\"This is the Lexicon client, that will execute all the logic.\"\"\"\n\n    def __init__(\n        self, config: Optional[Union[helper_config.ConfigResolver, Dict]] = None\n    ):\n        if not config:\n            # If there is not config specified, we load a non-interactive configuration.\n            self.config = helper_config.non_interactive_config_resolver()\n        elif not isinstance(config, helper_config.ConfigResolver):\n            # If config is not a ConfigResolver, we are in a legacy situation.\n            # We protect this part of the Client API.\n            self.config = helper_config.legacy_config_resolver(config)\n        else:\n            self.config = config\n\n        # Validate configuration\n        self._validate_config()\n\n        runtime_config = {}\n\n        # Process domain, strip subdomain\n        try:\n            domain_extractor = tldextract.TLDExtract(\n                cache_dir=_get_tldextract_cache_path(), include_psl_private_domains=True\n            )\n        except TypeError:\n            domain_extractor = tldextract.TLDExtract(\n                cache_file=_get_tldextract_cache_path(), include_psl_private_domains=True  # type: ignore\n            )\n        domain_parts = domain_extractor(\n            cast(str, self.config.resolve(\"lexicon:domain\"))\n        )\n        runtime_config[\"domain\"] = f\"{domain_parts.domain}.{domain_parts.suffix}\"\n\n        delegated = self.config.resolve(\"lexicon:delegated\")\n        if delegated:\n            # handle delegated domain\n            delegated = str(delegated).rstrip(\".\")\n            initial_domain = str(runtime_config.get(\"domain\"))\n            if delegated != initial_domain:\n                # convert to relative name\n                if delegated.endswith(initial_domain):\n                    delegated = delegated[: -len(initial_domain)]\n                    delegated = delegated.rstrip(\".\")\n                # update domain\n                runtime_config[\"domain\"] = f\"{delegated}.{initial_domain}\"\n\n        self.action = self.config.resolve(\"lexicon:action\")\n        self.provider_name = self.config.resolve(\n            \"lexicon:provider_name\"\n        ) or self.config.resolve(\"lexicon:provider\")\n\n        if not self.provider_name:\n            raise ValueError(\"Could not resolve provider name.\")\n\n        self.config.add_config_source(helper_config.DictConfigSource(runtime_config), 0)\n\n        provider_module = importlib.import_module(\n            \"lexicon.providers.\" + self.provider_name\n        )\n        provider_class: Type[Provider] = getattr(provider_module, \"Provider\")\n        self.provider = provider_class(self.config)\n\n    def execute(self) -> Union[bool, List[Dict]]:\n        \"\"\"Execute provided configuration in class constructor to the DNS records\"\"\"\n        self.provider.authenticate()\n        identifier = self.config.resolve(\"lexicon:identifier\")\n        record_type = self.config.resolve(\"lexicon:type\")\n        name = self.config.resolve(\"lexicon:name\")\n        content = self.config.resolve(\"lexicon:content\")\n\n        if self.action == \"create\":\n            if not record_type or not name or not content:\n                raise ValueError(\"Missing record_type, name or content parameters.\")\n            return self.provider.create_record(record_type, name, content)\n\n        if self.action == \"list\":\n            return self.provider.list_records(record_type, name, content)\n\n        if self.action == \"update\":\n            return self.provider.update_record(identifier, record_type, name, content)\n\n        if self.action == \"delete\":\n            return self.provider.delete_record(identifier, record_type, name, content)\n\n        raise ValueError(f\"Invalid action statement: {self.action}\")\n\n    def _validate_config(self) -> None:\n        provider_name = self.config.resolve(\"lexicon:provider_name\")\n        if not provider_name:\n            raise AttributeError(\"provider_name\")\n\n        try:\n            available = discovery.find_providers()[provider_name]\n        except KeyError:\n            raise ProviderNotAvailableError(\n                f\"This provider ({provider_name}) is not supported by Lexicon.\"\n            )\n        else:\n            if not available:\n                raise ProviderNotAvailableError(\n                    f\"This provider ({provider_name}) has required extra dependencies that are missing. \"\n                    f\"Please run `pip install lexicon[{provider_name}]` first before using it.\"\n                )\n\n        if not self.config.resolve(\"lexicon:action\"):\n            raise AttributeError(\"action\")\n        if not self.config.resolve(\"lexicon:domain\"):\n            raise AttributeError(\"domain\")\n        if not self.config.resolve(\"lexicon:type\"):\n            raise AttributeError(\"type\")\n\n\ndef _get_tldextract_cache_path() -> str:\n    if os.environ.get(\"TLDEXTRACT_CACHE_FILE\"):\n        logging.warning(\n            \"TLD_EXTRACT_CACHE_FILE environment variable is deprecated, please use TLDEXTRACT_CACHE_PATH instead.\"\n        )\n        os.environ[\"TLDEXTRACT_CACHE_PATH\"] = os.environ[\"TLDEXTRACT_CACHE_FILE\"]\n\n    return os.path.expanduser(\n        os.environ.get(\"TLDEXTRACT_CACHE_PATH\", os.path.join(\"~\", \".lexicon_tld_set\"))\n    )\n", "path": "lexicon/client.py"}, {"content": "\"\"\"Parsers definition for the Lexicon command-line interface\"\"\"\nimport argparse\nimport importlib\nimport os\n\nfrom lexicon import discovery\n\n\ndef generate_base_provider_parser() -> argparse.ArgumentParser:\n    \"\"\"Function that generates the base provider to be used by all dns providers.\"\"\"\n    parser = argparse.ArgumentParser(add_help=False)\n    parser.add_argument(\n        \"action\",\n        help=\"specify the action to take\",\n        default=\"list\",\n        choices=[\"create\", \"list\", \"update\", \"delete\"],\n    )\n    parser.add_argument(\n        \"domain\", help=\"specify the domain, supports subdomains as well\"\n    )\n    parser.add_argument(\n        \"type\",\n        help=\"specify the entry type\",\n        default=\"TXT\",\n        choices=[\"A\", \"AAAA\", \"CNAME\", \"MX\", \"NS\", \"SOA\", \"TXT\", \"SRV\", \"LOC\"],\n    )\n\n    parser.add_argument(\"--name\", help=\"specify the record name\")\n    parser.add_argument(\"--content\", help=\"specify the record content\")\n    parser.add_argument(\"--ttl\", type=int, help=\"specify the record time-to-live\")\n    parser.add_argument(\"--priority\", help=\"specify the record priority\")\n    parser.add_argument(\n        \"--identifier\", help=\"specify the record for update or delete actions\"\n    )\n    parser.add_argument(\n        \"--log_level\",\n        help=\"specify the log level\",\n        default=\"ERROR\",\n        choices=[\"CRITICAL\", \"ERROR\", \"WARNING\", \"INFO\", \"DEBUG\", \"NOTSET\"],\n    )\n    parser.add_argument(\n        \"--output\",\n        help=(\n            \"specify the type of output: by default a formatted table (TABLE), \"\n            \"a formatted table without header (TABLE-NO-HEADER), \"\n            \"a JSON string (JSON) or no output (QUIET)\"\n        ),\n        default=\"TABLE\",\n        choices=[\"TABLE\", \"TABLE-NO-HEADER\", \"JSON\", \"QUIET\"],\n    )\n    return parser\n\n\ndef generate_cli_main_parser() -> argparse.ArgumentParser:\n    \"\"\"Using all providers available, generate a parser that will be used by Lexicon CLI\"\"\"\n    parser = argparse.ArgumentParser(\n        description=\"Create, Update, Delete, List DNS entries\"\n    )\n\n    parser.add_argument(\n        \"--version\",\n        help=\"show the current version of lexicon\",\n        action=\"version\",\n        version=f\"%(prog)s {discovery.lexicon_version()}\",\n    )\n    parser.add_argument(\"--delegated\", help=\"specify the delegated domain\")\n    parser.add_argument(\n        \"--config-dir\",\n        default=os.getcwd(),\n        help=\"specify the directory where to search lexicon.yml and \"\n        \"lexicon_[provider].yml configuration files \"\n        \"(default: current directory).\",\n    )\n    subparsers = parser.add_subparsers(\n        dest=\"provider_name\", help=\"specify the DNS provider to use\"\n    )\n    subparsers.required = True\n\n    for provider, available in discovery.find_providers().items():\n        provider_module = importlib.import_module(\"lexicon.providers.\" + provider)\n        provider_parser = getattr(provider_module, \"provider_parser\")\n\n        subparser = subparsers.add_parser(\n            provider,\n            help=f\"{provider} provider\",\n            parents=[generate_base_provider_parser()],\n        )\n        provider_parser(subparser)\n\n        if not available:\n            subparser.epilog = (\n                \"WARNING: some required dependencies for this provider are not \"\n                f\"installed. Please run `pip install lexicon[{provider}]` first before using it.\"\n            )\n\n    return parser\n", "path": "lexicon/parser.py"}, {"content": "\"\"\"\nThis module takes care of finding information about the runtime of Lexicon:\n* what are the providers installed, and available\n* what is the version of Lexicon\n\"\"\"\nimport pkgutil\nimport re\nfrom typing import Dict\n\ntry:\n    from importlib.metadata import Distribution, PackageNotFoundError\nexcept ModuleNotFoundError:\n    from importlib_metadata import Distribution, PackageNotFoundError  # type: ignore[misc]\n\nfrom lexicon import providers\n\n\ndef find_providers() -> Dict[str, bool]:\n    \"\"\"Find all providers registered in Lexicon, and their availability\"\"\"\n    providers_list = sorted(\n        {\n            modname\n            for (_, modname, _) in pkgutil.iter_modules(providers.__path__)  # type: ignore\n            if modname != \"base\"\n        }\n    )\n\n    try:\n        distribution = Distribution.from_name(\"dns-lexicon\")\n    except PackageNotFoundError:\n        return {provider: True for provider in providers_list}\n    else:\n        return {\n            provider: _resolve_requirements(provider, distribution)\n            for provider in providers_list\n        }\n\n\ndef lexicon_version() -> str:\n    \"\"\"Retrieve current Lexicon version\"\"\"\n    try:\n        return Distribution.from_name(\"dns-lexicon\").version\n    except PackageNotFoundError:\n        return \"unknown\"\n\n\ndef _resolve_requirements(provider: str, distribution: Distribution) -> bool:\n    requires = distribution.requires\n    if requires is None:\n        raise ValueError(\"Error while trying finding requirements.\")\n\n    requirements = [\n        re.sub(r\"^(.*)\\s\\(.*\\)(?:;.*|)$\", r\"\\1\", requirement)\n        for requirement in requires\n        if f'extra == \"{provider}\"' in requirement\n    ]\n\n    if not requirements:\n        # No extra for this provider\n        return True\n\n    for requirement in requirements:\n        try:\n            Distribution.from_name(requirement)\n        except PackageNotFoundError:\n            # At least one extra requirement is not fulfilled\n            return False\n\n    # All extra requirements are fulfilled\n    return True\n", "path": "lexicon/discovery.py"}]}