17046631407

Committed 18 Aug 2025 04:35PM UTC coverage: 96.546% (-0.1%) from 96.648%

Build # 17046631407

Build Type

push

github

Committed by

funilrys

Commit Message

Fix issue with version extraction.

Run Details

11963 of 12391 relevant lines covered (96.55%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.41

/PyFunceble/query/requests/requester.py

"""
The tool to check the availability or syntax of domain, IP or URL.

::


    ██████╗ ██╗   ██╗███████╗██╗   ██╗███╗   ██╗ ██████╗███████╗██████╗ ██╗     ███████╗
    ██╔══██╗╚██╗ ██╔╝██╔════╝██║   ██║████╗  ██║██╔════╝██╔════╝██╔══██╗██║     ██╔════╝
    ██████╔╝ ╚████╔╝ █████╗  ██║   ██║██╔██╗ ██║██║     █████╗  ██████╔╝██║     █████╗
    ██╔═══╝   ╚██╔╝  ██╔══╝  ██║   ██║██║╚██╗██║██║     ██╔══╝  ██╔══██╗██║     ██╔══╝
    ██║        ██║   ██║     ╚██████╔╝██║ ╚████║╚██████╗███████╗██████╔╝███████╗███████╗
    ╚═╝        ╚═╝   ╚═╝      ╚═════╝ ╚═╝  ╚═══╝ ╚═════╝╚══════╝╚═════╝ ╚══════╝╚══════╝

Provides our own requests handler

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://docs.pyfunceble.com

Project homepage:
    https://pyfunceble.github.io/

License:
::


    Copyright 2017, 2018, 2019, 2020, 2022, 2023, 2024, 2025 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.
"""

import functools
import logging
import warnings
from typing import Optional, Union

import requests
import requests.exceptions
import urllib3.exceptions
from box import Box

import PyFunceble.storage
from PyFunceble.dataset.user_agent import UserAgentDataset
from PyFunceble.query.dns.query_tool import DNSQueryTool
from PyFunceble.query.requests.adapter.http import RequestHTTPAdapter
from PyFunceble.query.requests.adapter.https import RequestHTTPSAdapter


class Requester:
    """
    Provides our very own requests handler.

    :param int max_retries:
        Optional, The maximum number of retries to perform.
    :param bool verify_certificate:
        Optional, Should we verify and validate the SSL/TLS certificate ?
    :param float timeout:
        Optional, The timeout to apply to the query.
    :param int max_redirects:
        Optional, The maximum number of redirects to allow.
    :param dns_query_tool:
        Optional, The DNS Query tool to use.
    :param proxy_pattern:
        Optional, The proxy pattern to apply to each query.

        Expected format:

            ::
                {
                    "global": {
                        # Everything under global will be used as default if no
                        # rule matched.

                        "http": "str" ## HTTP_PROXY
                        "https": "str" ## HTTPS_PROXY
                    },
                    "rules":[
                        # A set/list of rules to work with.

                        {
                            "http": "str" ## HTTP_PROXY when TLD is matched.
                            "https": "str" ## HTTPS_PROXY when TLD is matched.
                            "tld": [
                                "str",
                                "str",
                                str
                            ]
                        },
                        {
                            "http": "str" ## HTTP_PROXY when TLD is matched.
                            "https": "str" ## HTTPS_PROXY when TLD is matched.
                            "tld": [
                                "str",
                                "str"
                            ]
                        },
                    ]
                }
    """

    STD_VERIFY_CERTIFICATE: bool = False
    STD_TIMEOUT: float = 3.0
    STD_MAX_RETRIES: int = 3

    urllib3_exceptions = urllib3.exceptions
    exceptions = requests.exceptions

    _timeout: float = 5.0
    _max_retries: int = 3
    _verify_certificate: bool = True
    _max_redirects: int = 60
    _proxy_pattern: dict = {}

    config: Optional[Box] = None

    session: Optional[requests.Session] = None
    dns_query_tool: Optional[DNSQueryTool] = None

    def __init__(
        self,
        *,
        max_retries: Optional[int] = None,
        verify_certificate: Optional[bool] = None,
        timeout: Optional[float] = None,
        max_redirects: Optional[int] = None,
        dns_query_tool: Optional[DNSQueryTool] = None,
        proxy_pattern: Optional[dict] = None,
        config: Optional[Box] = None,
    ) -> None:
        if config is not None:
            self.config = config
        else:
            self.config = Box({}, default_box=True)

        if max_retries is not None:
            self.max_retries = max_retries
        else:
            self.guess_and_set_max_retries()

        if verify_certificate is not None:
            self.verify_certificate = verify_certificate
        else:
            self.guess_and_set_verify_certificate()

        if timeout is not None:
            self.timeout = timeout
        else:
            self.guess_and_set_timeout()

        if max_redirects is not None:
            self.max_redirects = max_redirects

        if dns_query_tool is not None:
            self.dns_query_tool = dns_query_tool
        else:
            self.dns_query_tool = DNSQueryTool()

        if proxy_pattern is not None:
            self.proxy_pattern = proxy_pattern
        else:
            self.guess_and_set_proxy_pattern()

        self.session = self.get_session()

        warnings.simplefilter("ignore", urllib3.exceptions.InsecureRequestWarning)
        logging.getLogger("requests.packages.urllib3").setLevel(logging.CRITICAL)
        logging.getLogger("urllib3").setLevel(logging.CRITICAL)

    def recreate_session(func):  # pylint: disable=no-self-argument
        """
        Recreate a new session after executing the decorated method.
        """

        @functools.wraps(func)
        def wrapper(self, *args, **kwargs):
            result = func(self, *args, **kwargs)  # pylint: disable=not-callable

            if hasattr(self, "session") and isinstance(self.session, requests.Session):
                self.session = self.get_session()

            return result

        return wrapper

    def request_factory(verb: str):  # pylint: disable=no-self-argument
        """
        Provides a universal request factory.

        :param verb:
            The HTTP Verb to apply.
        """

        def request_method(func):
            @functools.wraps(func)
            def wrapper(self, *args, **kwargs):
                # pylint: disable=no-member
                req = getattr(self.session, verb.lower())(*args, **kwargs)
                return req

            return wrapper

        return request_method

    @property
    def headers(self) -> dict:
        """
        Provides the headers to use.
        """

        return self.session.headers

    @headers.setter
    @recreate_session
    def headers(self, value: dict) -> None:
        """
        Sets the headers to use.

        :param value:
            The headers to set.
        """

        self.session.headers.update(value)

    def set_config(self, config: Box) -> "Requester":
        """
        Sets the configuration to work with.

        :param config:
            The configuration to work with.
        """

        self.config = config

        return self

    @property
    def max_retries(self) -> int:
        """
        Provides the current state of the :code:`_max_retries` attribute.
        """

        return self._max_retries

    @max_retries.setter
    @recreate_session
    def max_retries(self, value: int) -> None:
        """
        Sets the max retries value to apply to all subsequent requests.

        :param value:
            The value to set.

        :raise TypeError:
            When the given :code:`value` is not a :py:class:`int`.
        :raise ValueError:
            When the given :code:`value` is less than :code:`1`.
        """

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

        if value < 0:
            raise ValueError(f"<value> ({value!r}) should be positive.")

        self._max_retries = value

    def set_max_retries(self, value: int) -> "Requester":
        """
        Sets the max retries value to apply to all subsequent requests.

        :param value:
            The value to set.
        """

        self.max_retries = value

        return self

    def guess_and_set_max_retries(self) -> "Requester":
        """
        Try to guess the value from the configuration and set it.
        """

        try:
            if isinstance(self.config.max_http_retries, int):
                self.set_max_retries(self.config.max_http_retries)
            else:
                self.set_max_retries(self.STD_MAX_RETRIES)
        except:  # pylint: disable=bare-except
            self.set_max_retries(self.STD_MAX_RETRIES)

        return self

    @property
    def max_redirects(self) -> int:
        """
        Provides the current state of the :code:`_max_redirects` attribute.
        """

        return self._max_redirects

    @max_redirects.setter
    @recreate_session
    def max_redirects(self, value: int) -> None:
        """
        Sets the max redirects value to apply to all subsequent requests.

        :param value:
            The value to set.

        :raise TypeError:
            When the given :code:`value` is not a :py:class:`int`.
        :raise ValueError:
            When the given :code:`value` is less than :code:`1`.
        """

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

        if value < 1:
            raise ValueError(f"<value> ({value!r}) should not be less than 1.")

        self._max_redirects = value

    def set_max_redirects(self, value: int) -> "Requester":
        """
        Sets the max redirects value to apply to all subsequent requests.

        :param value:
            The value to set.
        """

        self.max_redirects = value

        return self

    @property
    def verify_certificate(self) -> bool:
        """
        Provides the current state of the :code:`_verify_certificate` attribute.
        """

        return self._verify_certificate

    @verify_certificate.setter
    @recreate_session
    def verify_certificate(self, value: bool) -> None:
        """
        Enable or disables the certificate validation.

        :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._verify_certificate = value

    def set_verify_certificate(self, value: bool) -> "Requester":
        """
        Enable or disables the certificate validation.

        :param value:
            The value to set.
        """

        self.verify_certificate = value

        return self

    def guess_and_set_verify_certificate(self) -> "Requester":
        """
        Try to guess the value from the configuration and set it.
        """

        try:
            if isinstance(self.config.verify_ssl_certificate, bool):
                self.set_verify_certificate(self.config.verify_ssl_certificate)
            else:
                self.set_verify_certificate(self.STD_VERIFY_CERTIFICATE)
        except:  # pylint: disable=bare-except
            self.set_max_retries(self.STD_MAX_RETRIES)

        return self

    @property
    def timeout(self) -> float:
        """
        Provides the current state of the :code:`_timeout` attribute.
        """

        return self._timeout

    @timeout.setter
    @recreate_session
    def timeout(self, value: Union[int, float]) -> None:
        """
        Enable or disables the certificate validation.

        :param value:
            The value to set.

        :raise TypeError:
            When the given :code:`value` is not a :py:class`int` nor
            :py:class:`float`.
        :raise ValueError:
            Went the given :code:`value` is less than `1`.
        """

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

        if value < 0:
            raise ValueError("<value> should not be less than 0.")

        self._timeout = float(value)

    def set_timeout(self, value: Union[int, float]) -> "Requester":
        """
        Enable or disables the certificate validation.

        :param value:
            The value to set.
        """

        self.timeout = value

        return self

    def guess_and_set_timeout(self) -> "Requester":
        """
        Try to guess the value from the configuration and set it.
        """

        try:
            if isinstance(self.config.lookup.timeout, (int, float)):
                self.set_timeout(self.config.lookup.timeout)
            else:
                self.set_timeout(self.STD_TIMEOUT)
        except:  # pylint: disable=bare-except
            self.set_timeout(self.STD_TIMEOUT)

        return self

    @property
    def proxy_pattern(self) -> Optional[dict]:
        """
        Provides the current state of the :code:`_proxy_pattern` attribute.
        """

        return self._proxy_pattern

    @proxy_pattern.setter
    @recreate_session
    def proxy_pattern(self, value: dict) -> None:
        """
        Overwrite the proxy pattern to use.

        :param value:
            The value to set.

        :raise TypeError:
            When the given :code:`value` is not a :py:class`dict`.
        """

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

        self._proxy_pattern = value

    def set_proxy_pattern(self, value: dict) -> "Requester":
        """
        Overwrite the proxy pattern.

        :param value:
            The value to set.
        """

        self.proxy_pattern = value

        return self

    def guess_and_set_proxy_pattern(self) -> "Requester":
        """
        Try to guess the value from the configuration and set it.
        """

        try:
            if self.config.proxy:
                self.set_proxy_pattern(self.config.proxy)
            else:
                self.set_proxy_pattern({})
        except:  # pylint: disable=bare-except
            self.set_proxy_pattern({})

        return self

    def guess_all_settings(self) -> "Requester":
        """
        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 get_verify_certificate(self) -> bool:
        """
        Provides the current value of the certificate validation.
        """

        return self.verify_certificate

    def get_timeout(self) -> float:
        """
        Provides the currently set timeout.
        """

        return self.timeout

    def get_session(self) -> requests.Session:
        """
        Provides a new session.
        """

        session = requests.Session()

        session.verify = self.verify_certificate
        session.max_redirects = self.max_redirects
        session.mount(
            "https://",
            RequestHTTPSAdapter(
                max_retries=self.max_retries,
                timeout=self.timeout,
                dns_query_tool=self.dns_query_tool,
                proxy_pattern=self.proxy_pattern,
            ),
        )
        session.mount(
            "http://",
            RequestHTTPAdapter(
                max_retries=self.max_retries,
                timeout=self.timeout,
                dns_query_tool=self.dns_query_tool,
                proxy_pattern=self.proxy_pattern,
            ),
        )

        if PyFunceble.storage.USER_AGENTS:
            custom_headers = {"User-Agent": UserAgentDataset().get_latest()}
        else:
            custom_headers = {}

        session.headers.update(custom_headers)

        return session

    @request_factory("GET")
    def get(self, *args, **kwargs) -> requests.Response:
        """
        Sends a GET request and get its response.
        """

    @request_factory("OPTIONS")
    def options(self, *args, **kwargs) -> requests.Response:
        """
        Sends an OPTIONS request and get its response.
        """

    @request_factory("HEAD")
    def head(self, *args, **kwargs) -> requests.Response:
        """
        Sends a HEAD request and get its response.
        """

    @request_factory("POST")
    def post(self, *args, **kwargs) -> requests.Response:
        """
        Sends a POST request and get its response.
        """

    @request_factory("PUT")
    def put(self, *args, **kwargs) -> requests.Response:
        """
        Sends a PUT request and get its response.
        """

    @request_factory("PATCH")
    def patch(self, *args, **kwargs) -> requests.Response:
        """
        Sends a PATCH request and get its response.
        """

    @request_factory("DELETE")
    def delete(self, *args, **kwargs) -> requests.Response:
        """
        Sends a DELETE request and get its response.
        """

1	"""
2	The tool to check the availability or syntax of domain, IP or URL.
3
4	::
5
6
7	██████╗ ██╗ ██╗███████╗██╗ ██╗███╗ ██╗ ██████╗███████╗██████╗ ██╗ ███████╗
8	██╔══██╗╚██╗ ██╔╝██╔════╝██║ ██║████╗ ██║██╔════╝██╔════╝██╔══██╗██║ ██╔════╝
9	██████╔╝ ╚████╔╝ █████╗ ██║ ██║██╔██╗ ██║██║ █████╗ ██████╔╝██║ █████╗
10	██╔═══╝ ╚██╔╝ ██╔══╝ ██║ ██║██║╚██╗██║██║ ██╔══╝ ██╔══██╗██║ ██╔══╝
11	██║ ██║ ██║ ╚██████╔╝██║ ╚████║╚██████╗███████╗██████╔╝███████╗███████╗
12	╚═╝ ╚═╝ ╚═╝ ╚═════╝ ╚═╝ ╚═══╝ ╚═════╝╚══════╝╚═════╝ ╚══════╝╚══════╝
13
14	Provides our own requests handler
15
16	Author:
17	Nissar Chababy, @funilrys, contactTATAfunilrysTODTODcom
18
19	Special thanks:
20	https://pyfunceble.github.io/#/special-thanks
21
22	Contributors:
23	https://pyfunceble.github.io/#/contributors
24
25	Project link:
26	https://github.com/funilrys/PyFunceble
27
28	Project documentation:
29	https://docs.pyfunceble.com
30
31	Project homepage:
32	https://pyfunceble.github.io/
33
34	License:
35	::
36
37
38	Copyright 2017, 2018, 2019, 2020, 2022, 2023, 2024, 2025 Nissar Chababy
39
40	Licensed under the Apache License, Version 2.0 (the "License");
41	you may not use this file except in compliance with the License.
42	You may obtain a copy of the License at
43
44	https://www.apache.org/licenses/LICENSE-2.0
45
46	Unless required by applicable law or agreed to in writing, software
47	distributed under the License is distributed on an "AS IS" BASIS,
48	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
49	See the License for the specific language governing permissions and
50	limitations under the License.
51	"""
52
53	import functools	1✔
54	import logging	1✔
55	import warnings	1✔
56	from typing import Optional, Union	1✔
57
58	import requests	1✔
59	import requests.exceptions	1✔
60	import urllib3.exceptions	1✔
61	from box import Box	1✔
62
63	import PyFunceble.storage	1✔
64	from PyFunceble.dataset.user_agent import UserAgentDataset	1✔
65	from PyFunceble.query.dns.query_tool import DNSQueryTool	1✔
66	from PyFunceble.query.requests.adapter.http import RequestHTTPAdapter	1✔
67	from PyFunceble.query.requests.adapter.https import RequestHTTPSAdapter	1✔
68
69
70	class Requester:	1✔
71	"""
72	Provides our very own requests handler.
73
74	:param int max_retries:
75	Optional, The maximum number of retries to perform.
76	:param bool verify_certificate:
77	Optional, Should we verify and validate the SSL/TLS certificate ?
78	:param float timeout:
79	Optional, The timeout to apply to the query.
80	:param int max_redirects:
81	Optional, The maximum number of redirects to allow.
82	:param dns_query_tool:
83	Optional, The DNS Query tool to use.
84	:param proxy_pattern:
85	Optional, The proxy pattern to apply to each query.
86
87	Expected format:
88
89	::
90	{
91	"global": {
92	# Everything under global will be used as default if no
93	# rule matched.
94
95	"http": "str" ## HTTP_PROXY
96	"https": "str" ## HTTPS_PROXY
97	},
98	"rules":[
99	# A set/list of rules to work with.
100
101	{
102	"http": "str" ## HTTP_PROXY when TLD is matched.
103	"https": "str" ## HTTPS_PROXY when TLD is matched.
104	"tld": [
105	"str",
106	"str",
107	str
108	]
109	},
110	{
111	"http": "str" ## HTTP_PROXY when TLD is matched.
112	"https": "str" ## HTTPS_PROXY when TLD is matched.
113	"tld": [
114	"str",
115	"str"
116	]
117	},
118	]
119	}
120	"""
121
122	STD_VERIFY_CERTIFICATE: bool = False	1✔
123	STD_TIMEOUT: float = 3.0	1✔
124	STD_MAX_RETRIES: int = 3	1✔
125
126	urllib3_exceptions = urllib3.exceptions	1✔
127	exceptions = requests.exceptions	1✔
128
129	_timeout: float = 5.0	1✔
130	_max_retries: int = 3	1✔
131	_verify_certificate: bool = True	1✔
132	_max_redirects: int = 60	1✔
133	_proxy_pattern: dict = {}	1✔
134
135	config: Optional[Box] = None	1✔
136
137	session: Optional[requests.Session] = None	1✔
138	dns_query_tool: Optional[DNSQueryTool] = None	1✔
139
140	def __init__(
141	self,
142	*,
143	max_retries: Optional[int] = None,
144	verify_certificate: Optional[bool] = None,
145	timeout: Optional[float] = None,
146	max_redirects: Optional[int] = None,
147	dns_query_tool: Optional[DNSQueryTool] = None,
148	proxy_pattern: Optional[dict] = None,
149	config: Optional[Box] = None,
150	) -> None:
151	if config is not None:	1✔
152	self.config = config	1✔
153	else:
154	self.config = Box({}, default_box=True)	1✔
155
156	if max_retries is not None:	1✔
157	self.max_retries = max_retries	×
158	else:
159	self.guess_and_set_max_retries()	1✔
160
161	if verify_certificate is not None:	1✔
162	self.verify_certificate = verify_certificate	×
163	else:
164	self.guess_and_set_verify_certificate()	1✔
165
166	if timeout is not None:	1✔
167	self.timeout = timeout	×
168	else:
169	self.guess_and_set_timeout()	1✔
170
171	if max_redirects is not None:	1✔
172	self.max_redirects = max_redirects	×
173
174	if dns_query_tool is not None:	1✔
175	self.dns_query_tool = dns_query_tool	×
176	else:
177	self.dns_query_tool = DNSQueryTool()	1✔
178
179	if proxy_pattern is not None:	1✔
180	self.proxy_pattern = proxy_pattern	1✔
181	else:
182	self.guess_and_set_proxy_pattern()	1✔
183
184	self.session = self.get_session()	1✔
185
186	warnings.simplefilter("ignore", urllib3.exceptions.InsecureRequestWarning)	1✔
187	logging.getLogger("requests.packages.urllib3").setLevel(logging.CRITICAL)	1✔
188	logging.getLogger("urllib3").setLevel(logging.CRITICAL)	1✔
189
190	def recreate_session(func): # pylint: disable=no-self-argument	1✔
191	"""
192	Recreate a new session after executing the decorated method.
193	"""
194
195	@functools.wraps(func)	1✔
196	def wrapper(self, args, *kwargs):	1✔
197	result = func(self, args, *kwargs) # pylint: disable=not-callable	1✔
198
199	if hasattr(self, "session") and isinstance(self.session, requests.Session):	1✔
200	self.session = self.get_session()	×
201
202	return result	1✔
203
204	return wrapper	1✔
205
206	def request_factory(verb: str): # pylint: disable=no-self-argument	1✔
207	"""
208	Provides a universal request factory.
209
210	:param verb:
211	The HTTP Verb to apply.
212	"""
213
214	def request_method(func):	1✔
215	@functools.wraps(func)	1✔
216	def wrapper(self, args, *kwargs):	1✔
217	# pylint: disable=no-member
218	req = getattr(self.session, verb.lower())(args, *kwargs)	1✔
219	return req	1✔
220
221	return wrapper	1✔
222
223	return request_method	1✔
224
225	@property	1✔
226	def headers(self) -> dict:	1✔
227	"""
228	Provides the headers to use.
229	"""
230
231	return self.session.headers	×
232
233	@headers.setter	1✔
234	@recreate_session	1✔
235	def headers(self, value: dict) -> None:	1✔
236	"""
237	Sets the headers to use.
238
239	:param value:
240	The headers to set.
241	"""
242
243	self.session.headers.update(value)	×
244
245	def set_config(self, config: Box) -> "Requester":	1✔
246	"""
247	Sets the configuration to work with.
248
249	:param config:
250	The configuration to work with.
251	"""
252
253	self.config = config	×
254
255	return self	×
256
257	@property	1✔
258	def max_retries(self) -> int:	1✔
259	"""
260	Provides the current state of the :code:`_max_retries` attribute.
261	"""
262
263	return self._max_retries	1✔
264
265	@max_retries.setter	1✔
266	@recreate_session	1✔
267	def max_retries(self, value: int) -> None:	1✔
268	"""
269	Sets the max retries value to apply to all subsequent requests.
270
271	:param value:
272	The value to set.
273
274	:raise TypeError:
275	When the given :code:`value` is not a :py:class:`int`.
276	:raise ValueError:
277	When the given :code:`value` is less than :code:`1`.
278	"""
279
280	if not isinstance(value, int):	1✔
281	raise TypeError(f"<value> should be {int}, {type(value)} given.")	×
282
283	if value < 0:	1✔
284	raise ValueError(f"<value> ({value!r}) should be positive.")	×
285
286	self._max_retries = value	1✔
287
288	def set_max_retries(self, value: int) -> "Requester":	1✔
289	"""
290	Sets the max retries value to apply to all subsequent requests.
291
292	:param value:
293	The value to set.
294	"""
295
296	self.max_retries = value	1✔
297
298	return self	1✔
299
300	def guess_and_set_max_retries(self) -> "Requester":	1✔
301	"""
302	Try to guess the value from the configuration and set it.
303	"""
304
305	try:	1✔
306	if isinstance(self.config.max_http_retries, int):	1✔
307	self.set_max_retries(self.config.max_http_retries)	×
308	else:
309	self.set_max_retries(self.STD_MAX_RETRIES)	1✔
310	except: # pylint: disable=bare-except	1✔
311	self.set_max_retries(self.STD_MAX_RETRIES)	1✔
312
313	return self	1✔
314
315	@property	1✔
316	def max_redirects(self) -> int:	1✔
317	"""
318	Provides the current state of the :code:`_max_redirects` attribute.
319	"""
320
321	return self._max_redirects	1✔
322
323	@max_redirects.setter	1✔
324	@recreate_session	1✔
325	def max_redirects(self, value: int) -> None:	1✔
326	"""
327	Sets the max redirects value to apply to all subsequent requests.
328
329	:param value:
330	The value to set.
331
332	:raise TypeError:
333	When the given :code:`value` is not a :py:class:`int`.
334	:raise ValueError:
335	When the given :code:`value` is less than :code:`1`.
336	"""
337
338	if not isinstance(value, int):	×
339	raise TypeError(f"<value> should be {int}, {type(value)} given.")	×
340
341	if value < 1:	×
342	raise ValueError(f"<value> ({value!r}) should not be less than 1.")	×
343
344	self._max_redirects = value	×
345
346	def set_max_redirects(self, value: int) -> "Requester":	1✔
347	"""
348	Sets the max redirects value to apply to all subsequent requests.
349
350	:param value:
351	The value to set.
352	"""
353
354	self.max_redirects = value	×
355
356	return self	×
357
358	@property	1✔
359	def verify_certificate(self) -> bool:	1✔
360	"""
361	Provides the current state of the :code:`_verify_certificate` attribute.
362	"""
363
364	return self._verify_certificate	1✔
365
366	@verify_certificate.setter	1✔
367	@recreate_session	1✔
368	def verify_certificate(self, value: bool) -> None:	1✔
369	"""
370	Enable or disables the certificate validation.
371
372	:param value:
373	The value to set.
374
375	:raise TypeError:
376	When the given :code:`value` is not a :py:class`bool`.
377	"""
378
379	if not isinstance(value, bool):	1✔
380	raise TypeError(f"<value> should be {bool}, {type(value)} given.")	×
381
382	self._verify_certificate = value	1✔
383
384	def set_verify_certificate(self, value: bool) -> "Requester":	1✔
385	"""
386	Enable or disables the certificate validation.
387
388	:param value:
389	The value to set.
390	"""
391
392	self.verify_certificate = value	1✔
393
394	return self	1✔
395
396	def guess_and_set_verify_certificate(self) -> "Requester":	1✔
397	"""
398	Try to guess the value from the configuration and set it.
399	"""
400
401	try:	1✔
402	if isinstance(self.config.verify_ssl_certificate, bool):	1✔
403	self.set_verify_certificate(self.config.verify_ssl_certificate)	×
404	else:
405	self.set_verify_certificate(self.STD_VERIFY_CERTIFICATE)	1✔
406	except: # pylint: disable=bare-except	1✔
407	self.set_max_retries(self.STD_MAX_RETRIES)	1✔
408
409	return self	1✔
410
411	@property	1✔
412	def timeout(self) -> float:	1✔
413	"""
414	Provides the current state of the :code:`_timeout` attribute.
415	"""
416
417	return self._timeout	1✔
418
419	@timeout.setter	1✔
420	@recreate_session	1✔
421	def timeout(self, value: Union[int, float]) -> None:	1✔
422	"""
423	Enable or disables the certificate validation.
424
425	:param value:
426	The value to set.
427
428	:raise TypeError:
429	When the given :code:`value` is not a :py:class`int` nor
430	:py:class:`float`.
431	:raise ValueError:
432	Went the given :code:`value` is less than `1`.
433	"""
434
435	if not isinstance(value, (int, float)):	1✔
436	raise TypeError(f"<value> should be {int} or {float}, {type(value)} given.")	×
437
438	if value < 0:	1✔
439	raise ValueError("<value> should not be less than 0.")	×
440
441	self._timeout = float(value)	1✔
442
443	def set_timeout(self, value: Union[int, float]) -> "Requester":	1✔
444	"""
445	Enable or disables the certificate validation.
446
447	:param value:
448	The value to set.
449	"""
450
451	self.timeout = value	1✔
452
453	return self	1✔
454
455	def guess_and_set_timeout(self) -> "Requester":	1✔
456	"""
457	Try to guess the value from the configuration and set it.
458	"""
459
460	try:	1✔
461	if isinstance(self.config.lookup.timeout, (int, float)):	1✔
462	self.set_timeout(self.config.lookup.timeout)	×
463	else:
464	self.set_timeout(self.STD_TIMEOUT)	1✔
465	except: # pylint: disable=bare-except	1✔
466	self.set_timeout(self.STD_TIMEOUT)	1✔
467
468	return self	1✔
469
470	@property	1✔
471	def proxy_pattern(self) -> Optional[dict]:	1✔
472	"""
473	Provides the current state of the :code:`_proxy_pattern` attribute.
474	"""
475
476	return self._proxy_pattern	1✔
477
478	@proxy_pattern.setter	1✔
479	@recreate_session	1✔
480	def proxy_pattern(self, value: dict) -> None:	1✔
481	"""
482	Overwrite the proxy pattern to use.
483
484	:param value:
485	The value to set.
486
487	:raise TypeError:
488	When the given :code:`value` is not a :py:class`dict`.
489	"""
490
491	if not isinstance(value, dict):	1✔
492	raise TypeError(f"<value> should be {dict}, {type(value)} given.")	×
493
494	self._proxy_pattern = value	1✔
495
496	def set_proxy_pattern(self, value: dict) -> "Requester":	1✔
497	"""
498	Overwrite the proxy pattern.
499
500	:param value:
501	The value to set.
502	"""
503
504	self.proxy_pattern = value	1✔
505
506	return self	1✔
507
508	def guess_and_set_proxy_pattern(self) -> "Requester":	1✔
509	"""
510	Try to guess the value from the configuration and set it.
511	"""
512
513	try:	1✔
514	if self.config.proxy:	1✔
515	self.set_proxy_pattern(self.config.proxy)	×
516	else:
517	self.set_proxy_pattern({})	1✔
518	except: # pylint: disable=bare-except	1✔
519	self.set_proxy_pattern({})	1✔
520
521	return self	1✔
522
523	def guess_all_settings(self) -> "Requester":	1✔
524	"""
525	Try to guess all settings.
526	"""
527
528	to_ignore = ["guess_all_settings"]	×
529
530	for method in dir(self):	×
531	if method in to_ignore or not method.startswith("guess_"):	×
532	continue	×
533
534	getattr(self, method)()	×
535
536	return self	×
537
538	def get_verify_certificate(self) -> bool:	1✔
539	"""
540	Provides the current value of the certificate validation.
541	"""
542
543	return self.verify_certificate	×
544
545	def get_timeout(self) -> float:	1✔
546	"""
547	Provides the currently set timeout.
548	"""
549
550	return self.timeout	×
551
552	def get_session(self) -> requests.Session:	1✔
553	"""
554	Provides a new session.
555	"""
556
557	session = requests.Session()	1✔
558
559	session.verify = self.verify_certificate	1✔
560	session.max_redirects = self.max_redirects	1✔
561	session.mount(	1✔
562	"https://",
563	RequestHTTPSAdapter(
564	max_retries=self.max_retries,
565	timeout=self.timeout,
566	dns_query_tool=self.dns_query_tool,
567	proxy_pattern=self.proxy_pattern,
568	),
569	)
570	session.mount(	1✔
571	"http://",
572	RequestHTTPAdapter(
573	max_retries=self.max_retries,
574	timeout=self.timeout,
575	dns_query_tool=self.dns_query_tool,
576	proxy_pattern=self.proxy_pattern,
577	),
578	)
579
580	if PyFunceble.storage.USER_AGENTS:	1✔
581	custom_headers = {"User-Agent": UserAgentDataset().get_latest()}	1✔
582	else:
583	custom_headers = {}	1✔
584
585	session.headers.update(custom_headers)	1✔
586
587	return session	1✔
588
589	@request_factory("GET")	1✔
590	def get(self, args, *kwargs) -> requests.Response:	1✔
591	"""
592	Sends a GET request and get its response.
593	"""
594
595	@request_factory("OPTIONS")	1✔
596	def options(self, args, *kwargs) -> requests.Response:	1✔
597	"""
598	Sends an OPTIONS request and get its response.
599	"""
600
601	@request_factory("HEAD")	1✔
602	def head(self, args, *kwargs) -> requests.Response:	1✔
603	"""
604	Sends a HEAD request and get its response.
605	"""
606
607	@request_factory("POST")	1✔
608	def post(self, args, *kwargs) -> requests.Response:	1✔
609	"""
610	Sends a POST request and get its response.
611	"""
612
613	@request_factory("PUT")	1✔
614	def put(self, args, *kwargs) -> requests.Response:	1✔
615	"""
616	Sends a PUT request and get its response.
617	"""
618
619	@request_factory("PATCH")	1✔
620	def patch(self, args, *kwargs) -> requests.Response:	1✔
621	"""
622	Sends a PATCH request and get its response.
623	"""
624
625	@request_factory("DELETE")	1✔
626	def delete(self, args, *kwargs) -> requests.Response:	1✔
627	"""
628	Sends a DELETE request and get its response.
629	"""

funilrys / PyFunceble / 17046631407

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous