Skip to content

Commit

Permalink
Merge remote-tracking branch 'origin/switch-collection' into dev
Browse files Browse the repository at this point in the history
  • Loading branch information
@funilrys
funilrys committed Jan 14, 2024
2 parents 1893678 + 89ef643 commit 14ae6a9
Show file tree
Hide file tree
Showing 2 changed files with 146 additions and 44 deletions.
17 changes: 15 additions & 2 deletions PyFunceble/checker/availability/base.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -793,8 +793,21 @@ def try_to_query_status_from_whois(
}
)
else:
self.status.expiration_date = known_record["expiration_date"]
self.status.registrar = known_record["registrar"]
# We pass the known record to the lookup record - so that other
# methods can use it.
self.whois_query_tool.lookup_record.record = known_record.get(
"record", None
)
self.whois_query_tool.lookup_record.expiration_date = known_record.get(
"expiration_date", None
)
self.whois_query_tool.lookup_record.registrar = known_record.get(
"registrar", None
)
self.status.whois_lookup_record = self.whois_query_tool.lookup_record

self.status.expiration_date = known_record.get("expiration_date", None)
self.status.registrar = known_record.get("registrar", None)
self.status.whois_record = None
else:
self.status.expiration_date = self.whois_query_tool.expiration_date
Expand Down
173 changes: 131 additions & 42 deletions PyFunceble/query/collection.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,8 +51,6 @@
"""

import json
import logging
from datetime import datetime
from typing import List, Optional, Union

import requests
Expand Down Expand Up @@ -106,6 +104,11 @@ class CollectionQueryTool:
The preferred data origin
"""

_is_modern_api: Optional[bool] = None
"""
Whether we are working with the modern or legacy API.
"""

session: Optional[requests.Session] = None

def __init__(
Expand Down Expand Up @@ -136,6 +139,7 @@ def __init__(
self.session.headers.update(
{
"Authorization": f"Bearer {self.token}" if self.token else None,
"X-Pyfunceble-Version": PyFunceble.storage.PROJECT_VERSION,
"Content-Type": "application/json",
}
)
Expand Down Expand Up @@ -242,6 +246,43 @@ def set_url_base(self, value: str) -> "CollectionQueryTool":

return self

@property
def is_modern_api(self) -> bool:
"""
Provides the value of the :code:`_is_modern_api` attribute.
"""

return self._is_modern_api

@is_modern_api.setter
def is_modern_api(self, value: bool) -> None:
"""
Sets the value of the :code:`_is_modern_api` attribute.
:param value:
The value to set.
:raise TypeError:
When the given :code:`value` is not a :py:class:`bool`.
"""

if not isinstance(value, bool):
raise TypeError(f"<value> should be {bool}, {type(value)} given.")

self._is_modern_api = value

def set_is_modern_api(self, value: bool) -> "CollectionQueryTool":
"""
Sets the value of the :code:`_is_modern_api` attribute.
:param value:
The value to set.
"""

self.is_modern_api = value

return self

def guess_and_set_url_base(self) -> "CollectionQueryTool":
"""
Try to guess the URL base to work with.
Expand All @@ -252,11 +293,31 @@ def guess_and_set_url_base(self) -> "CollectionQueryTool":
self.url_base = PyFunceble.storage.CONFIGURATION.collection.url_base
else:
self.url_base = self.STD_URL_BASE
elif EnvironmentVariableHelper("PYFUNCEBLE_COLLECTION_API_URL").exists():
self.url_base = EnvironmentVariableHelper(
"PYFUNCEBLE_COLLECTION_API_URL"
).get_value()
else:
self.url_base = self.STD_URL_BASE

return self

def guess_and_set_is_modern_api(self) -> "CollectionQueryTool":
"""
Try to guess if we are working with a legacy version.
"""

try:
response = self.session.get(
f"{self.url_base}/v1/stats/subject",
)

response.raise_for_status()

self.is_modern_api = False
except (requests.RequestException, json.decoder.JSONDecodeError):
self.is_modern_api = True

@property
def preferred_status_origin(self) -> Optional[str]:
"""
Expand Down Expand Up @@ -319,6 +380,21 @@ def guess_and_set_preferred_status_origin(self) -> "CollectionQueryTool":

return self

def ensure_is_modern_api_is_set(func): # pylint: disable=no-self-argument
"""
Ensures that the :code:`is_modern_api` attribute is set before running
the decorated method.
"""

def wrapper(self, *args, **kwargs):
if self.is_modern_api is None:
self.guess_and_set_is_modern_api()

return func(self, *args, **kwargs) # pylint: disable=not-callable

return wrapper

@ensure_is_modern_api_is_set
def pull(self, subject: str) -> Optional[dict]:
"""
Pulls all data related to the subject or :py:class:`None`
Expand All @@ -333,12 +409,18 @@ def pull(self, subject: str) -> Optional[dict]:
The response of the search.
"""

logging.info("Starting to search subject: %r", subject)
PyFunceble.facility.Logger.info("Starting to search subject: %r", subject)

if not isinstance(subject, str):
raise TypeError(f"<subject> should be {str}, {type(subject)} given.")

url = f"{self.url_base}/v1/subject/search"
if self.is_modern_api:
if self.token:
url = f"{self.url_base}/v1/aggregation/subject/search"
else:
url = f"{self.url_base}/v1/hub/aggregation/subject/search"
else:
url = f"{self.url_base}/v1/subject/search"

try:
response = self.session.post(
Expand All @@ -349,25 +431,28 @@ def pull(self, subject: str) -> Optional[dict]:
response_json = response.json()

if response.status_code == 200:
logging.debug(
PyFunceble.facility.Logger.debug(
"Successfully search subject: %r. Response: %r",
subject,
response_json,
)

logging.info("Finished to search subject: %r", subject)
PyFunceble.facility.Logger.info(
"Finished to search subject: %r", subject
)

return response_json
except (requests.RequestException, json.decoder.JSONDecodeError):
response_json = {}

logging.debug(
PyFunceble.facility.Logger.debug(
"Failed to search subject: %r. Response: %r", subject, response_json
)
logging.info("Finished to search subject: %r", subject)
PyFunceble.facility.Logger.info("Finished to search subject: %r", subject)

return None

@ensure_is_modern_api_is_set
def push(
self,
checker_status: Union[
Expand Down Expand Up @@ -418,29 +503,15 @@ def push(
if checker_status.subject == "":
raise ValueError("<checker_status.subject> cannot be empty.")

status_to_subject = {
"status": checker_status.status,
"status_source": checker_status.status_source,
"tested_at": checker_status.tested_at.isoformat(),
"subject": checker_status.idna_subject,
}

if (
hasattr(checker_status, "expiration_date")
not self.is_modern_api
and hasattr(checker_status, "expiration_date")
and checker_status.expiration_date
):
self.__push_whois(
{
"subject": checker_status.idna_subject,
"expiration_date": datetime.strptime(
checker_status.expiration_date, "%d-%b-%Y"
).isoformat(),
"registrar": checker_status.registrar,
}
)
self.__push_whois(checker_status)

data = self.__push_status(
checker_status.checker_type.lower(), status_to_subject
checker_status.checker_type.lower(), checker_status.to_json()
)

return data
Expand All @@ -462,7 +533,9 @@ def guess_all_settings(

return self

def __push_status(self, checker_type: str, data: dict) -> Optional[dict]:
def __push_status(
self, checker_type: str, data: Union[dict, str]
) -> Optional[dict]:
"""
Submits the given status to the collection.
Expand All @@ -486,31 +559,45 @@ def __push_status(self, checker_type: str, data: dict) -> Optional[dict]:
if checker_type not in self.SUPPORTED_CHECKERS:
raise ValueError(f"<checker_type> ({checker_type}) is not supported.")

logging.info("Starting to submit status: %r", data)
PyFunceble.facility.Logger.info("Starting to submit status: %r", data)

url = f"{self.url_base}/v1/status/{checker_type}"
if self.is_modern_api:
if not self.token:
url = f"{self.url_base}/v1/hub/status/{checker_type}"
else:
url = f"{self.url_base}/v1/contracts/self-delivery"
else:
url = f"{self.url_base}/v1/status/{checker_type}"

try:
response = self.session.post(
url,
json=data,
)
if isinstance(data, dict):
response = self.session.post(
url,
json=data,
)
else:
response = self.session.post(
url,
data=data,
)

response_json = response.json()

if response.status_code == 200:
logging.debug("Successfully submitted data: %r to %s", data, url)
PyFunceble.facility.Logger.debug(
"Successfully submitted data: %r to %s", data, url
)

logging.info("Finished to submit status: %r", data)
PyFunceble.facility.Logger.info("Finished to submit status: %r", data)
return response_json
except (requests.RequestException, json.decoder.JSONDecodeError):
response_json = {}

logging.debug(
PyFunceble.facility.Logger.debug(
"Failed to submit data: %r to %s. Response: %r", data, url, response_json
)

logging.info("Finished to submit status: %r", data)
PyFunceble.facility.Logger.info("Finished to submit status: %r", data)

return None

Expand Down Expand Up @@ -538,7 +625,7 @@ def __push_whois(self, data: dict) -> Optional[dict]:
if not isinstance(data, dict): # pragma: no cover ## Should never happen
raise TypeError(f"<data> should be {dict}, {type(data)} given.")

logging.info("Starting to submit WHOIS: %r", data)
PyFunceble.facility.Logger.info("Starting to submit WHOIS: %r", data)

url = f"{self.url_base}/v1/status/whois"

Expand All @@ -551,16 +638,18 @@ def __push_whois(self, data: dict) -> Optional[dict]:
response_json = response.json()

if response.status_code == 200:
logging.debug("Successfully submitted WHOIS data: %r to %s", data, url)
PyFunceble.facility.Logger.debug(
"Successfully submitted WHOIS data: %r to %s", data, url
)

logging.info("Finished to submit WHOIS: %r", data)
PyFunceble.facility.Logger.info("Finished to submit WHOIS: %r", data)
return response_json
except (requests.RequestException, json.decoder.JSONDecodeError):
response_json = {}

logging.debug(
PyFunceble.facility.Logger.debug(
"Failed to WHOIS data: %r to %s. Response: %r", data, url, response_json
)

logging.info("Finished to submit WHOIS: %r", data)
PyFunceble.facility.Logger.info("Finished to submit WHOIS: %r", data)
return None

0 comments on commit 14ae6a9

Please sign in to comment.