Source code for PyFunceble.query.platform
"""
The tool to check the availability or syntax of domain, IP or URL.
::
██████╗ ██╗ ██╗███████╗██╗ ██╗███╗ ██╗ ██████╗███████╗██████╗ ██╗ ███████╗
██╔══██╗╚██╗ ██╔╝██╔════╝██║ ██║████╗ ██║██╔════╝██╔════╝██╔══██╗██║ ██╔════╝
██████╔╝ ╚████╔╝ █████╗ ██║ ██║██╔██╗ ██║██║ █████╗ ██████╔╝██║ █████╗
██╔═══╝ ╚██╔╝ ██╔══╝ ██║ ██║██║╚██╗██║██║ ██╔══╝ ██╔══██╗██║ ██╔══╝
██║ ██║ ██║ ╚██████╔╝██║ ╚████║╚██████╗███████╗██████╔╝███████╗███████╗
╚═╝ ╚═╝ ╚═╝ ╚═════╝ ╚═╝ ╚═══╝ ╚═════╝╚══════╝╚═════╝ ╚══════╝╚══════╝
Provides ans interface which let us interact with the platform API.
Author:
Nissar Chababy, @funilrys, contactTATAfunilrysTODTODcom
Special thanks:
https://pyfunceble.github.io/#/special-thanks
Contributors:
https://pyfunceble.github.io/#/contributors
Project link:
https://github.com/funilrys/PyFunceble
Project documentation:
https://pyfunceble.readthedocs.io/en/dev/
Project homepage:
https://pyfunceble.github.io/
License:
::
Copyright 2017, 2018, 2019, 2020, 2022, 2023, 2024 Nissar Chababy
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
# pylint: disable=too-many-lines
import json
import os
from typing import Generator, List, Optional, Union
import requests
import requests.exceptions
import PyFunceble.facility
import PyFunceble.storage
from PyFunceble.checker.availability.status import AvailabilityCheckerStatus
from PyFunceble.checker.reputation.status import ReputationCheckerStatus
from PyFunceble.checker.syntax.status import SyntaxCheckerStatus
from PyFunceble.helpers.environment_variable import EnvironmentVariableHelper
[docs]
class PlatformQueryTool:
"""
Provides the interface to interact with the platform.
:param token:
The token to use to communicate with the API.
.. warning::
If :code:`None` is given, the class constructor will try to load the
:code:`PYFUNCEBLE_COLLECTION_API_TOKEN` or
:code:`PYFUNCEBLE_PLATFORM_API_TOKEN` environment variable.
:param url_base:
The base of the URL to communicate with.
:param preferred_status_origin:
The preferred data origin.
It can be :code:`frequent`, :code:`latest` or :code:`recommended`.
"""
SUPPORTED_CHECKERS: List[str] = ["syntax", "reputation", "availability"]
SUPPORTED_STATUS_ORIGIN: List[str] = ["frequent", "latest", "recommended"]
SUBJECT: str = (
"10927294711127294799272947462729471152729471162729471152729471112729"
"4710427294745272947100272947972729471012729471002729474627294797272947"
"116272947101272947982729474627294710527294711227294797272947472729474"
"727294758272947115272947112272947116272947116272947104"
)
STD_PREFERRED_STATUS_ORIGIN: str = "frequent"
STD_CHECKER_PRIORITY: str = ["none"]
STD_CHECKER_EXCLUDE: str = ["none"]
STD_TIMEOUT: float = 5.0
_token: Optional[str] = None
"""
The token to use while communicating with the platform API.
"""
_url_base: Optional[str] = None
"""
The base of the URL to communicate with.
"""
_preferred_status_origin: Optional[str] = None
"""
The preferred data origin
"""
_checker_priority: Optional[List[str]] = []
"""
The checker to prioritize.
"""
_checker_exclude: Optional[List[str]] = []
"""
The checker to exclude.
"""
_is_modern_api: Optional[bool] = None
"""
Whether we are working with the modern or legacy API.
"""
_timeout: float = 5.0
"""
The timeout to use while communicating with the API.
"""
session: Optional[requests.Session] = None
def __init__(
self,
*,
token: Optional[str] = None,
preferred_status_origin: Optional[str] = None,
timeout: Optional[float] = None,
checker_priority: Optional[List[str]] = None,
checker_exclude: Optional[List[str]] = None,
) -> None:
if token is not None:
self.token = token
else:
self.token = EnvironmentVariableHelper(
"PYFUNCEBLE_COLLECTION_API_TOKEN"
).get_value(default="") or EnvironmentVariableHelper(
"PYFUNCEBLE_PLATFORM_API_TOKEN"
).get_value(
default=""
)
if preferred_status_origin is not None:
self.preferred_status_origin = preferred_status_origin
else:
self.guess_and_set_preferred_status_origin()
if checker_priority is not None:
self.checker_priority = checker_priority
else:
self.guess_and_set_checker_priority()
if checker_exclude is not None:
self.checker_exclude = checker_exclude
else:
self.guess_and_set_checker_exclude()
if timeout is not None:
self.timeout = timeout
else:
self.guess_and_set_timeout()
self._url_base = EnvironmentVariableHelper(
"PYFUNCEBLE_COLLECTION_API_URL"
).get_value(default=None) or EnvironmentVariableHelper(
"PYFUNCEBLE_PLATFORM_API_URL"
).get_value(
default=None
)
self.session = requests.Session()
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",
}
)
def __contains__(self, value: str) -> bool:
"""
Checks if the given value is in the platform.
:param value:
The value to check.
"""
return self.pull(value) is not None
def __getitem__(self, value: str) -> Optional[dict]:
"""
Gets the information about the given value.
:param value:
The value to get the information about.
"""
return self.pull(value)
@property
def token(self) -> Optional[str]:
"""
Provides the currently set token.
"""
return self._token
@token.setter
def token(self, value: str) -> None:
"""
Sets the value of the :code:`_token` attribute.
:param value:
The value to set.
:raise TypeError:
When the given :code:`value` is not a :py:class:`str`
"""
if not isinstance(value, str):
raise TypeError(f"<value> should be {str}, {type(value)} given.")
self._token = value
[docs]
def set_token(self, value: str) -> "PlatformQueryTool":
"""
Sets the value of the :code:`_token` attribute.
:param value:
The value to set.
"""
self.token = value
return self
@property
def url_base(self) -> Optional[str]:
"""
Provides the value of the :code:`_url_base` attribute.
"""
return self._url_base or "".join(
reversed([chr(int(x)) for x in self.SUBJECT.split("272947")])
)
@url_base.setter
def url_base(self, value: str) -> None:
"""
Sets the base of the URL to work with.
:param value:
The value to set.
:raise TypeError:
When the given :code:`value` is not a :py:class:`str`.
"""
if not isinstance(value, str):
raise TypeError(f"<value> should be {str}, {type(value)} given.")
if not value.startswith(("http", "https")):
raise ValueError(
f"<value> is missing the scheme (http/https), {value} given."
)
self._url_base = value.rstrip("/")
[docs]
def set_url_base(self, value: str) -> "PlatformQueryTool":
"""
Sets the base of the URL to work with.
:parma value:
The value to set.
"""
self.url_base = value
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
[docs]
def set_is_modern_api(self, value: bool) -> "PlatformQueryTool":
"""
Sets the value of the :code:`_is_modern_api` attribute.
:param value:
The value to set.
"""
self.is_modern_api = value
return self
@property
def timeout(self) -> float:
"""
Provides the value of the :code:`_timeout` attribute.
"""
return self._timeout
@timeout.setter
def timeout(self, value: float) -> None:
"""
Sets the value of the :code:`_timeout` attribute.
:param value:
The value to set.
:raise TypeError:
When the given :code:`value` is not a :py:class:`float`.
"""
if not isinstance(value, (int, float)):
raise TypeError(f"<value> should be {float}, {type(value)} given.")
self._timeout = value
[docs]
def set_timeout(self, value: float) -> "PlatformQueryTool":
"""
Sets the value of the :code:`_timeout` attribute.
:param value:
The value to set.
"""
self.timeout = value
return self
[docs]
def guess_and_set_is_modern_api(self) -> "PlatformQueryTool":
"""
Try to guess if we are working with a legacy version.
"""
if self.token:
try:
response = self.session.get(
f"{self.url_base}/v1/stats/subject",
timeout=self.timeout,
)
response.raise_for_status()
self.is_modern_api = False
except (requests.RequestException, json.decoder.JSONDecodeError):
self.is_modern_api = True
else:
self.is_modern_api = False
return self
@property
def preferred_status_origin(self) -> Optional[str]:
"""
Provides the value of the :code:`_preferred_status_origin` attribute.
"""
return self._preferred_status_origin
@preferred_status_origin.setter
def preferred_status_origin(self, value: str) -> None:
"""
Sets the preferred status origin.
:param value:
The value to set.
:raise TypeError:
When the given :code:`value` is not a :py:class:`str`.
:raise ValueError:
When the given :code:`value` is not supported.
"""
if not isinstance(value, str):
raise TypeError(f"<value> should be {str}, {type(value)} given.")
if value not in self.SUPPORTED_STATUS_ORIGIN:
raise ValueError(f"<value> ({value}) is not supported.")
self._preferred_status_origin = value
[docs]
def set_preferred_status_origin(self, value: str) -> "PlatformQueryTool":
"""
Sets the preferred status origin.
:parma value:
The value to set.
"""
self.preferred_status_origin = value
return self
[docs]
def guess_and_set_preferred_status_origin(self) -> "PlatformQueryTool":
"""
Try to guess the preferred status origin.
"""
if PyFunceble.facility.ConfigLoader.is_already_loaded():
if isinstance(
PyFunceble.storage.CONFIGURATION.platform.preferred_status_origin, str
):
self.preferred_status_origin = (
PyFunceble.storage.CONFIGURATION.platform.preferred_status_origin
)
else:
self.preferred_status_origin = self.STD_PREFERRED_STATUS_ORIGIN
else:
self.preferred_status_origin = self.STD_PREFERRED_STATUS_ORIGIN
return self
@property
def checker_priority(self) -> Optional[List[str]]:
"""
Provides the value of the :code:`_checker_priority` attribute.
"""
return self._checker_priority
@checker_priority.setter
def checker_priority(self, value: List[str]) -> None:
"""
Sets the checker priority to set - order matters.
:param value:
The value to set.
:raise TypeError:
When the given :code:`value` is not a :py:class:`str`.
:raise ValueError:
When the given :code:`value` is not supported.
"""
accepted = []
for checker_type in value:
if not isinstance(checker_type, str):
raise TypeError(
f"<checker_type> ({checker_type}) should be {str}, "
"{type(checker_type)} given."
)
if checker_type.lower() not in self.SUPPORTED_CHECKERS + ["none"]:
raise ValueError(f"<checker_type> ({checker_type}) is not supported.")
accepted.append(checker_type.lower())
self._checker_priority = accepted
[docs]
def set_checker_priority(self, value: List[str]) -> "PlatformQueryTool":
"""
Sets the checker priority.
:parma value:
The value to set.
"""
self.checker_priority = value
return self
[docs]
def guess_and_set_checker_priority(self) -> "PlatformQueryTool":
"""
Try to guess the checker priority to use.
"""
if "PYFUNCEBLE_PLATFORM_CHECKER_PRIORITY" in os.environ:
self.checker_priority = os.environ[
"PYFUNCEBLE_PLATFORM_CHECKER_PRIORITY"
].split(",")
elif PyFunceble.facility.ConfigLoader.is_already_loaded():
if isinstance(PyFunceble.storage.PLATFORM.checker_priority, list):
self.checker_priority = PyFunceble.storage.PLATFORM.checker_priority
else:
self.checker_priority = self.STD_CHECKER_PRIORITY
else:
self.checker_priority = self.STD_CHECKER_PRIORITY
return self
@property
def checker_exclude(self) -> Optional[List[str]]:
"""
Provides the value of the :code:`_checker_exclude` attribute.
"""
return self._checker_exclude
@checker_exclude.setter
def checker_exclude(self, value: List[str]) -> None:
"""
Sets the checker exclude.
:param value:
The value to set.
:raise TypeError:
When the given :code:`value` is not a :py:class:`str`.
:raise ValueError:
When the given :code:`value` is not supported.
"""
accepted = []
for checker_type in value:
if not isinstance(checker_type, str):
raise TypeError(
f"<checker_type> ({checker_type}) should be {str}, "
"{type(checker_type)} given."
)
if checker_type.lower() not in self.SUPPORTED_CHECKERS + ["none"]:
raise ValueError(f"<checker_type> ({checker_type}) is not supported.")
accepted.append(checker_type.lower())
self._checker_exclude = accepted
[docs]
def set_checker_exclude(self, value: List[str]) -> "PlatformQueryTool":
"""
Sets the checker to exclude.
:parma value:
The value to set.
"""
self.checker_exclude = value
return self
[docs]
def guess_and_set_checker_exclude(self) -> "PlatformQueryTool":
"""
Try to guess the checker to exclude.
"""
if "PYFUNCEBLE_PLATFORM_CHECKER_EXCLUDE" in os.environ:
self.checker_exclude = os.environ[
"PYFUNCEBLE_PLATFORM_CHECKER_EXCLUDE"
].split(",")
elif PyFunceble.facility.ConfigLoader.is_already_loaded():
if isinstance(PyFunceble.storage.PLATFORM.checker_exclude, list):
self.checker_exclude = PyFunceble.storage.PLATFORM.checker_exclude
else:
self.checker_exclude = self.STD_CHECKER_EXCLUDE
else:
self.checker_exclude = self.STD_CHECKER_EXCLUDE
return self
[docs]
def guess_and_set_timeout(self) -> "PlatformQueryTool":
"""
Try to guess the timeout to use.
"""
if PyFunceble.facility.ConfigLoader.is_already_loaded():
self.timeout = PyFunceble.storage.CONFIGURATION.lookup.timeout
else:
self.timeout = self.STD_TIMEOUT
return self
[docs]
def ensure_modern_api(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
[docs]
@ensure_modern_api
def pull(self, subject: str) -> Optional[dict]:
"""
Pulls all data related to the subject or :py:class:`None`
:param subject:
The subject to search for.
:raise TypeError:
When the given :code:`subject` is not a :py:class:`str`.
:return:
The response of the search.
"""
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.")
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(
url,
json={"subject": subject},
timeout=self.timeout,
)
response_json = response.json()
if response.status_code == 200:
PyFunceble.facility.Logger.debug(
"Successfully search subject: %r. Response: %r",
subject,
response_json,
)
PyFunceble.facility.Logger.info(
"Finished to search subject: %r", subject
)
return response_json
except (requests.RequestException, json.decoder.JSONDecodeError):
response_json = {}
PyFunceble.facility.Logger.debug(
"Failed to search subject: %r. Response: %r", subject, response_json
)
PyFunceble.facility.Logger.info("Finished to search subject: %r", subject)
return None
[docs]
@ensure_modern_api
def pull_contract(self, amount: int = 1) -> Generator[dict, None, None]:
"""
Pulls the next amount of contracts.
:param int amount:
The amount of data to pull.
:return:
The response of the query.
"""
PyFunceble.facility.Logger.info("Starting to pull next contract")
if not isinstance(amount, int) or amount < 1:
amount = 1
url = f"{self.url_base}/v1/contracts/next"
params = {
"limit": amount,
"shuffle": True,
}
if "none" in self.checker_priority:
params["shuffle"] = True
else:
params["checker_type_priority"] = ",".join(self.checker_priority)
if "none" not in self.checker_exclude:
params["checker_type_exclude"] = ",".join(self.checker_exclude)
try:
response = self.session.get(
url,
params=params,
timeout=self.timeout * 10,
)
response_json = response.json()
if response.status_code == 200:
PyFunceble.facility.Logger.debug(
"Successfully pulled next %r contracts. Response: %r", response_json
)
PyFunceble.facility.Logger.info("Finished to pull next contract")
yield response_json
else:
response_json = []
except (requests.RequestException, json.decoder.JSONDecodeError):
response_json = []
PyFunceble.facility.Logger.debug(
"Failed to pull next contract. Response: %r", response_json
)
PyFunceble.facility.Logger.info("Finished to pull next contracts")
yield response_json
[docs]
@ensure_modern_api
def deliver_contract(self, contract: dict, contract_data: dict) -> Optional[dict]:
"""
Delivers the given contract data.
:param contract:
The contract to deliver.
:param contract_data:
The data to deliver.
:return:
The response of the query.
"""
PyFunceble.facility.Logger.info(
"Starting to deliver contract data: %r", contract
)
contract_id = contract["id"]
contract_data = (
contract_data.to_json()
if not isinstance(contract_data, dict)
else contract_data
)
url = f"{self.url_base}/v1/contracts/{contract_id}/delivery"
try:
response = self.session.post(
url,
data=contract_data.encode("utf-8"),
timeout=self.timeout * 10,
)
response_json = response.json()
if response.status_code == 200:
PyFunceble.facility.Logger.debug(
"Successfully delivered contract: %r. Response: %r",
contract_data,
response_json,
)
PyFunceble.facility.Logger.info(
"Finished to deliver contract: %r", contract_data
)
return response_json
except (requests.RequestException, json.decoder.JSONDecodeError):
response_json = {}
PyFunceble.facility.Logger.debug(
"Failed to deliver contract: %r. Response: %r", contract_data, response_json
)
PyFunceble.facility.Logger.info(
"Finished to deliver contract: %r", contract_data
)
return None
[docs]
@ensure_modern_api
def push(
self,
checker_status: Union[
AvailabilityCheckerStatus, SyntaxCheckerStatus, ReputationCheckerStatus
],
) -> Optional[dict]:
"""
Push the given status to the platform.
:param checker_status:
The status to push.
:raise TypeError:
- When the given :code:`checker_status` is not a
:py:class:`AvailabilityCheckerStatus`,
:py:class:`SyntaxCheckerStatus` or
:py:class:`ReputationCheckerStatus`.
- When the given :code:`checker_status.subject` is not a
:py:class:`str`.
:raise ValueError:
When the given :code:`checker_status.subject` is empty.
"""
if not isinstance(
checker_status,
(AvailabilityCheckerStatus, SyntaxCheckerStatus, ReputationCheckerStatus),
):
raise TypeError(
f"<checker_status> should be {AvailabilityCheckerStatus}, "
f"{SyntaxCheckerStatus} or {ReputationCheckerStatus}, "
f"{type(checker_status)} given."
)
if not isinstance(checker_status.subject, str):
raise TypeError(
f"<checker_status.subject> should be {str}, "
f"{type(checker_status.subject)} given."
)
if not isinstance(checker_status.checker_type, str):
raise TypeError(
f"<checker_status_checker_type> should be {str}, "
f"{type(checker_status.subject)} given."
)
if checker_status.subject == "":
raise ValueError("<checker_status.subject> cannot be empty.")
if (
not self.is_modern_api
and hasattr(checker_status, "expiration_date")
and checker_status.expiration_date
):
self.__push_whois(checker_status)
data = self.__push_status(
checker_status.checker_type.lower(), checker_status.to_json()
)
return data
[docs]
def guess_all_settings(
self,
) -> "PlatformQueryTool": # pragma: no cover ## Underlying tested
"""
Try to guess all settings.
"""
to_ignore = ["guess_all_settings"]
for method in dir(self):
if method in to_ignore or not method.startswith("guess_"):
continue
getattr(self, method)()
return self
def __push_status(
self, checker_type: str, data: Union[dict, str]
) -> Optional[dict]:
"""
Submits the given status to the platform.
:param checker_type:
The type of the checker.
:param data:
The data to submit.
:raise TypeError:
- When :code:`checker_type` is not a :py:class:`str`.
- When :code:`data` is not a :py:class:`dict`.
:raise ValueError:
When the given :code:`checker_type` is not a subject checker type.
"""
if not self.token:
return None
if checker_type not in self.SUPPORTED_CHECKERS:
raise ValueError(f"<checker_type> ({checker_type}) is not supported.")
PyFunceble.facility.Logger.info("Starting to submit status: %r", data)
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:
if isinstance(data, dict):
response = self.session.post(
url,
json=data,
timeout=self.timeout * 10,
)
elif isinstance(
data,
(
AvailabilityCheckerStatus,
SyntaxCheckerStatus,
ReputationCheckerStatus,
),
):
response = self.session.post(
url,
json=data.to_dict(),
timeout=self.timeout * 10,
)
else:
response = self.session.post(
url,
data=data,
timeout=self.timeout * 10,
)
response_json = response.json()
if response.status_code == 200:
PyFunceble.facility.Logger.debug(
"Successfully submitted data: %r to %s", data, url
)
PyFunceble.facility.Logger.info("Finished to submit status: %r", data)
return response_json
except (requests.RequestException, json.decoder.JSONDecodeError):
response_json = {}
PyFunceble.facility.Logger.debug(
"Failed to submit data: %r to %s. Response: %r", data, url, response_json
)
PyFunceble.facility.Logger.info("Finished to submit status: %r", data)
return None
def __push_whois(self, data: dict) -> Optional[dict]:
"""
Submits the given WHOIS data into the given subject.
:param checker_type:
The type of the checker.
:param data:
The data to submit.
:raise TypeError:
- When :code:`checker_type` is not a :py:class:`str`.
- When :code:`data` is not a :py:class:`dict`.
:raise ValueError:
When the given :code:`checker_type` is not a subject checker type.
"""
if not self.token:
return None
PyFunceble.facility.Logger.info("Starting to submit WHOIS: %r", data)
url = f"{self.url_base}/v1/status/whois"
try:
if isinstance(data, dict):
response = self.session.post(
url,
json=data,
timeout=self.timeout * 10,
)
elif isinstance(
data,
(
AvailabilityCheckerStatus,
SyntaxCheckerStatus,
ReputationCheckerStatus,
),
):
response = self.session.post(
url,
data=data.to_json(),
timeout=self.timeout * 10,
)
else:
response = self.session.post(
url,
data=data,
timeout=self.timeout * 10,
)
response_json = response.json()
if response.status_code == 200:
PyFunceble.facility.Logger.debug(
"Successfully submitted WHOIS data: %r to %s", data, url
)
PyFunceble.facility.Logger.info("Finished to submit WHOIS: %r", data)
return response_json
except (requests.RequestException, json.decoder.JSONDecodeError):
response_json = {}
PyFunceble.facility.Logger.debug(
"Failed to WHOIS data: %r to %s. Response: %r", data, url, response_json
)
PyFunceble.facility.Logger.info("Finished to submit WHOIS: %r", data)
return None