Implement cleanup(), fully document deprecation

pyproject.toml

@@ -87,6 +87,10 @@ lexicon = "lexicon.cli:main"
 
 [tool.pytest.ini_options]
 junit_family = "xunit2"
+filterwarnings = [
+    "error",
+    "ignore:Method execute\\(\\) is deprecated:DeprecationWarning",
+]
 
 [tool.mypy]
 show_error_codes = true

src/lexicon/cli.py

@@ -5,6 +5,7 @@
 import os
 import sys
 from typing import Dict, List, Optional, Union
+import warnings
 
 from lexicon.client import Client
 from lexicon.config import ConfigResolver
@@ -129,7 +130,8 @@ def main() -> None:
 
     client = Client(config)
 
-    results = client.execute()
+    with warnings.simplefilter("ignore", category=DeprecationWarning):
+        results = client.execute()
 
     action = config.resolve("lexicon:action")
     if not action:

src/lexicon/client.py

@@ -6,7 +6,7 @@
 from contextlib import AbstractContextManager
 from types import TracebackType
 from typing import Type, Any
-from warnings import warn
+import warnings
 from threading import local
 
 import tldextract  # type: ignore
@@ -153,20 +153,23 @@ def __exit__(self, __exc_type: type[BaseException] | None, __exc_value: BaseExce
 
         return None
 
-    @staticmethod
-    def test():
-        warn("""
-Method execute() is deprecated. Please remove action/type/name/content from the
-config, and use the dedicated action methods in the context manager itself.
+    def execute(self) -> bool | list[dict[str, Any]]:
+        """(deprecated) Execute provided configuration in class constructor to the DNS records"""
+        message = """\
+Method execute() is deprecated and will be removed in Lexicon 4>=.
+
+Please remove action/type/name/content fields from Lexicon config,
+and use the methods dedicated for each action (*_record()/list_records()).
+These methods are available within the Lexicon client context manager.
 
 Example for creating a record:
 
 with Client(config) as operations:
     operations.create_record("TXT", "foo", "bar")
-""")
+        """
+
+        warnings.warn(message, DeprecationWarning, stacklevel=2)
 
-    def execute(self) -> bool | list[dict[str, Any]]:
-        """Execute provided configuration in class constructor to the DNS records"""
         action = self.config.resolve("lexicon:action")
         identifier = self.config.resolve("lexicon:identifier")
         rtype = self.config.resolve("lexicon:type")

src/lexicon/providers/aliyun.py

@@ -53,6 +53,9 @@ def authenticate(self):
 
         return self
 
+    def cleanup(self) -> None:
+        pass
+
     def create_record(self, rtype, name, content):
         if not self.list_records(rtype, name, content):
             query_params = {

src/lexicon/providers/aurora.py

@@ -45,6 +45,9 @@ def authenticate(self):
         else:
             raise AuthenticationError("No domain found")
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If record already exists with the same content, do nothing'
     def create_record(self, rtype, name, content):
         data = {"type": rtype, "name": self._relative_name(name), "content": content}

src/lexicon/providers/azure.py

@@ -294,6 +294,9 @@ def authenticate(self):
 
         self.domain_id = our_data[0]["id"]
 
+    def cleanup(self) -> None:
+        pass
+
     def _request(self, action="GET", url="/", data=None, query_params=None):
         query_params = {} if not query_params else query_params.copy()
         query_params["api-version"] = API_VERSION

src/lexicon/providers/base.py

@@ -76,7 +76,9 @@ def authenticate(self) -> None:
         """
 
     def cleanup(self) -> None:
-        pass
+        """
+        Clean any relevant resource before this provider instance is closed.
+        """
 
     @abstractmethod
     def create_record(self, rtype: str, name: str, content: str) -> bool:

src/lexicon/providers/cloudflare.py

@@ -66,6 +66,9 @@ def authenticate(self):
 
             self.domain_id = zone_id
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If record already exists with the same content, do nothing'
     def create_record(self, rtype, name, content):
         content, cf_data = self._format_content(rtype, content)

src/lexicon/providers/cloudns.py

@@ -45,6 +45,9 @@ def authenticate(self):
         self.domain_id = payload["name"]
         LOGGER.debug("authenticate: %s", payload)
 
+    def cleanup(self) -> None:
+        pass
+
     def create_record(self, rtype, name, content):
         # Skip execution if such a record already exists
         existing_records = self.list_records(rtype, name, content)

src/lexicon/providers/cloudxns.py

@@ -45,6 +45,9 @@ def authenticate(self):
         else:
             raise AuthenticationError("No domain found")
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If record already exists with the same content, do nothing'
     def create_record(self, rtype, name, content):
         record = {

src/lexicon/providers/conoha.py

@@ -96,6 +96,9 @@ def authenticate(self):
 
         self.domain_id = payload["domains"][0]["id"]
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If record already exists with the same content, do nothing'
     def create_record(self, rtype, name, content):
         if not rtype:

src/lexicon/providers/constellix.py

@@ -69,6 +69,9 @@ def authenticate(self):
         else:
             raise AuthenticationError("No domain found")
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If record already exists with the same content, do nothing'
 
     def create_record(self, rtype, name, content):

src/lexicon/providers/ddns.py

@@ -55,6 +55,9 @@ def authenticate(self):
             raise AuthenticationError("No DDNS server provided, use --ddns-server")
         pass
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If record already exists with the same content, do nothing
     def create_record(self, rtype, name, content):
         if self._get_lexicon_option("ttl"):

src/lexicon/providers/digitalocean.py

@@ -31,6 +31,9 @@ def authenticate(self):
         self._get(f"/domains/{self.domain}")
         self.domain_id = self.domain
 
+    def cleanup(self) -> None:
+        pass
+
     def create_record(self, rtype, name, content):
         # check if record already exists
         ttl = self._get_lexicon_option("ttl")

src/lexicon/providers/dinahosting.py

@@ -75,6 +75,9 @@ def authenticate(self):
 
         self.domain_id = self.domain
 
+    def cleanup(self) -> None:
+        pass
+
     def create_record(self, rtype, name, content):
         Provider._check_unsupported_type(rtype)
 

src/lexicon/providers/directadmin.py

@@ -56,6 +56,9 @@ def authenticate(self):
         except ValueError:
             raise AuthenticationError(f"Domain {self.domain} not found")
 
+    def cleanup(self) -> None:
+        pass
+
     def create_record(self, rtype, name, content):
         # Refuse to create duplicate records
         existing_records = self.list_records(rtype, name, content)

src/lexicon/providers/dnsimple.py

@@ -67,6 +67,9 @@ def authenticate(self):
         else:
             raise AuthenticationError(f"No domain found like {self.domain}")
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If record already exists with the same content, do nothing
 
     def create_record(self, rtype, name, content):

src/lexicon/providers/dnsmadeeasy.py

@@ -77,6 +77,9 @@ def authenticate(self):
 
         self.domain_id = payload["id"]
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If record already exists with the same content, do nothing'
 
     def create_record(self, rtype, name, content):

src/lexicon/providers/dnspark.py

@@ -39,7 +39,10 @@ def authenticate(self):
 
         self.domain_id = payload["additional"]["domain_id"]
 
-    # Create record. If record already exists with the same content, do nothing'
+    def cleanup(self) -> None:
+        pass
+
+    # Create record. If record already exists with the same content, do nothing.
 
     def create_record(self, rtype, name, content):
         record = {"rname": self._relative_name(name), "rtype": rtype, "rdata": content}

src/lexicon/providers/dnspod.py

@@ -36,6 +36,9 @@ def authenticate(self):
 
         self.domain_id = payload["domain"]["id"]
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If record already exists with the same content, do nothing'
     def create_record(self, rtype, name, content):
         record = {

src/lexicon/providers/dnsservices.py

@@ -55,6 +55,9 @@ def authenticate(self):
                 return
         raise AuthenticationError("No domain found")
 
+    def cleanup(self) -> None:
+        pass
+
     def create_record(self, rtype, name, content):
         print("create_record")
         # check if record already exists

src/lexicon/providers/dreamhost.py

@@ -124,6 +124,9 @@ def authenticate(self):
         if self.domain_id is None:
             raise AuthenticationError("Domain not found")
 
+    def cleanup(self) -> None:
+        pass
+
     def create_record(self, rtype, name, content):
         name = self._full_name(name)
 

src/lexicon/providers/duckdns.py

@@ -123,6 +123,9 @@ def authenticate(self):
         if self._get_provider_option("auth_token") is None:
             raise AuthenticationError("Must provide account token")
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If the record already exists with the same content, do nothing"
     def create_record(self, rtype, name, content):
         if self._get_lexicon_option("ttl"):

src/lexicon/providers/dynu.py

@@ -41,6 +41,9 @@ def authenticate(self):
         else:
             raise AuthenticationError("No matching domain found")
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If record already exists with the same content, do nothing.
     def create_record(self, rtype, name, content):
         record = self._to_dynu_record(rtype, name, content)

src/lexicon/providers/easydns.py

@@ -45,6 +45,9 @@ def authenticate(self):
 
         self.domain_id = payload["data"]["id"]
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If record already exists with the same content, do nothing'
 
     def create_record(self, rtype, name, content):

src/lexicon/providers/easyname.py

@@ -98,6 +98,9 @@ def authenticate(self):
 
         return True
 
+    def cleanup(self) -> None:
+        pass
+
     def create_record(self, rtype, name, content):
         return self._create_record_internal(rtype=rtype, name=name, content=content)
 

src/lexicon/providers/euserv.py

@@ -103,6 +103,9 @@ def authenticate(self):
         else:
             raise AuthenticationError("Domain not found in DNS records")
 
+    def cleanup(self) -> None:
+        pass
+
     def create_record(self, rtype, name, content):
         records = self.list_records(rtype, name, content)
 

src/lexicon/providers/exoscale.py

@@ -36,6 +36,9 @@ def authenticate(self):
 
         self.domain_id = response["domain"]["id"]
 
+    def cleanup(self) -> None:
+        pass
+
     def create_record(self, rtype, name, content):
         """Create record if doesnt already exist with same content"""
         # check if record already exists

src/lexicon/providers/flexibleengine.py

@@ -49,6 +49,9 @@ def authenticate(self):
             self.domain_id = zone_id
             self.domain = payload["zones"][0]["name"].rstrip(".")
 
+    def cleanup(self) -> None:
+        pass
+
     def create_record(self, rtype, name, content):
         # put string in array
         tmp = content

src/lexicon/providers/gandi.py

@@ -92,6 +92,9 @@ def authenticate(self):
             self._get(f"/domains/{self.domain}")
             self.domain_id = self.domain.lower()
 
+    def cleanup(self) -> None:
+        pass
+
     def create_record(self, rtype, name, content):
         if self.protocol == "rpc":
             return self.rpc_helper.create_record(

src/lexicon/providers/gehirn.py

@@ -69,6 +69,9 @@ def authenticate(self):
         self.domain_id = domains[0]["id"]
         self.version_id = domains[0]["current_version_id"]
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If record already exists with the same content, do nothing'
     def create_record(self, rtype, name, content):
         name = self._full_name(name)

src/lexicon/providers/glesys.py

@@ -38,6 +38,9 @@ def authenticate(self):
         else:
             raise AuthenticationError("No domain found")
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If record already exists with the same content, do nothing.
     def create_record(self, rtype, name, content):
         existing = self.list_records(rtype, name, content)

src/lexicon/providers/godaddy.py

@@ -62,6 +62,9 @@ def authenticate(self):
         result = self._get(f"/domains/{domain}")
         self.domain_id = result["domainId"]
 
+    def cleanup(self) -> None:
+        pass
+
     def list_records(self, rtype=None, name=None, content=None):
         domain = self.domain
 

src/lexicon/providers/googleclouddns.py

@@ -200,6 +200,9 @@ def authenticate(self):
 
         self.domain_id = targeted_managed_zone_ids[0]
 
+    def cleanup(self) -> None:
+        pass
+
     # List all records for the given type/name/content.
     # It is quite straight forward to request data, the biggest operation is to convert
     # the stacked multivalued RecordSets into Lexicon monovalued entries.

src/lexicon/providers/gransy.py

@@ -72,6 +72,9 @@ def authenticate(self):
         else:
             raise AuthenticationError("No SSID provided by server")
 
+    def cleanup(self) -> None:
+        pass
+
     # Create record. If record already exists with the same content, do nothing.
     def create_record(self, rtype, name, content):
         """Creates a new unique record"""