IDEV-2011: Move list of feed products in constants.py

Author: jbabac

domaintools/api.py

@@ -3,7 +3,7 @@
 from hmac import new as hmac
 import re
 
-from domaintools.constants import Endpoint, ENDPOINT_TO_SOURCE_MAP, OutputFormat
+from domaintools.constants import Endpoint, ENDPOINT_TO_SOURCE_MAP, FEEDS_PRODUCTS_LIST, OutputFormat
 from domaintools._version import current as version
 from domaintools.results import (
     GroupedIterable,
@@ -19,7 +19,7 @@
     filter_by_field,
     DTResultFilter,
 )
-from domaintools.utils import get_feeds_products_list, validate_feeds_parameters
+from domaintools.utils import validate_feeds_parameters
 
 
 AVAILABLE_KEY_SIGN_HASHES = ["sha1", "sha256", "md5"]
@@ -133,7 +133,7 @@ def _results(self, product, path, cls=Results, **kwargs):
 
     def handle_api_key(self, product, path, parameters, header_authentication):
         if self.https and not self.always_sign_api_key:
-            if product in get_feeds_products_list() and header_authentication:
+            if product in FEEDS_PRODUCTS_LIST and header_authentication:
                 parameters["X-Api-Key"] = self.key
             else:
                 parameters["api_key"] = self.key

domaintools/base_results.py

@@ -9,7 +9,7 @@
 from datetime import datetime
 from httpx import Client
 
-from domaintools.constants import OutputFormat, HEADER_ACCEPT_KEY_CSV_FORMAT
+from domaintools.constants import FEEDS_PRODUCTS_LIST, OutputFormat, HEADER_ACCEPT_KEY_CSV_FORMAT
 from domaintools.exceptions import (
     BadRequestException,
     InternalServerErrorException,
@@ -20,7 +20,6 @@
     IncompleteResponseException,
     RequestUriTooLongException,
 )
-from domaintools.utils import get_feeds_products_list
 
 
 try:  # pragma: no cover
@@ -93,7 +92,7 @@ def _make_request(self):
                 patch_data = self.kwargs.copy()
                 patch_data.update(self.api.extra_request_params)
                 return session.patch(url=self.url, json=patch_data)
-            elif self.product in get_feeds_products_list():
+            elif self.product in FEEDS_PRODUCTS_LIST:
                 parameters = deepcopy(self.kwargs)
                 parameters.pop("output_format", None)
                 parameters.pop(
@@ -139,8 +138,7 @@ def data(self):
             self.setStatus(results.status_code, results)
             if (
                 self.kwargs.get("format", "json") == "json"
-                and self.product
-                not in get_feeds_products_list()  # Special handling of feeds products' data to preserve the result in jsonline format
+                and self.product not in FEEDS_PRODUCTS_LIST  # Special handling of feeds products' data to preserve the result in jsonline format
             ):
                 self._data = results.json()
             else:
@@ -157,7 +155,7 @@ def data(self):
             return self._data
 
     def check_limit_exceeded(self):
-        if self.kwargs.get("format", "json") == "json" and self.product not in get_feeds_products_list():
+        if self.kwargs.get("format", "json") == "json" and self.product not in FEEDS_PRODUCTS_LIST:
             if "response" in self._data and "limit_exceeded" in self._data["response"] and self._data["response"]["limit_exceeded"] is True:
                 return True, self._data["response"]["message"]
         # TODO: handle html, xml response errors better.

domaintools/constants.py

@@ -22,3 +22,14 @@ class OutputFormat(Enum):
     Endpoint.FEED.value: Source.API,
     Endpoint.DOWNLOAD.value: Source.S3,
 }
+
+FEEDS_PRODUCTS_LIST = [
+    "newly-active-domains-feed-(api)",
+    "newly-active-domains-feed-(s3)",
+    "newly-observed-domains-feed-(api)",
+    "newly-observed-domains-feed-(s3)",
+    "domain-registration-data-access-protocol-feed-(api)",
+    "domain-registration-data-access-protocol-feed-(s3)",
+    "real-time-domain-discovery-feed-(api)",
+    "real-time-domain-discovery-feed-(s3)",
+]

domaintools/utils.py

@@ -172,19 +172,6 @@ def convert_str_to_dateobj(string_date: str, date_format: Optional[str] = "%Y-%m
     return datetime.strptime(string_date, date_format)
 
 
-def get_feeds_products_list():
-    return [
-        "newly-active-domains-feed-(api)",
-        "newly-active-domains-feed-(s3)",
-        "newly-observed-domains-feed-(api)",
-        "newly-observed-domains-feed-(s3)",
-        "domain-registration-data-access-protocol-feed-(api)",
-        "domain-registration-data-access-protocol-feed-(s3)",
-        "real-time-domain-discovery-feed-(api)",
-        "real-time-domain-discovery-feed-(s3)",
-    ]
-
-
 def validate_feeds_parameters(params):
     sessionID = params.get("sessionID")
     after = params.get("after")

domaintools_async/__init__.py

@@ -6,9 +6,8 @@
 from httpx import AsyncClient
 
 from domaintools.base_results import Results
-from domaintools.constants import OutputFormat, HEADER_ACCEPT_KEY_CSV_FORMAT
+from domaintools.constants import FEEDS_PRODUCTS_LIST, OutputFormat, HEADER_ACCEPT_KEY_CSV_FORMAT
 from domaintools.exceptions import ServiceUnavailableException
-from domaintools.utils import get_feeds_products_list
 
 
 class _AIter(object):
@@ -52,7 +51,7 @@ async def _make_async_request(self, session):
             patch_data = self.kwargs.copy()
             patch_data.update(self.api.extra_request_params)
             results = await session.patch(url=self.url, json=patch_data)
-        elif self.product in get_feeds_products_list():
+        elif self.product in FEEDS_PRODUCTS_LIST:
             parameters = deepcopy(self.kwargs)
             parameters.pop("output_format", None)
             parameters.pop(