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

95.65

/PyFunceble/dataset/db_base.py

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

::


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

Provides the base of all datasets which acts as database interface.

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
from typing import Any, Generator, List, Optional

import PyFunceble.storage
from PyFunceble.dataset.base import DatasetBase


class DBDatasetBase(DatasetBase):
    """
    Provides the base of all datasets which acts as database interface.
    """

    STD_REMOVE_UNNEEDED_FIELDS: bool = True
    STD_AUTHORIZED: bool = False

    FIELDS: List[str] = []
    COMPARISON_FIELDS: List[str] = []

    source_file: Optional[str] = None

    _remove_unneeded_fields: Optional[bool] = True
    _authorized: Optional[bool] = False

    def __init__(
        self,
        *,
        authorized: Optional[bool] = None,
        remove_unneeded_fields: Optional[bool] = None,
        config_dir: Optional[str] = None,
    ) -> None:
        if config_dir is not None:
            self.config_dir = config_dir
        else:
            self.config_dir = PyFunceble.storage.CONFIG_DIRECTORY

        if authorized is not None:
            self.set_authorized(authorized)

        if remove_unneeded_fields is not None:
            self.set_remove_unneeded_fields(remove_unneeded_fields)

        self.__post_init__()

    def __post_init__(self) -> None:
        """
        A method to be called (automatically) after __init__.
        """

    def execute_if_authorized(default: Any = None):  # pylint: disable=no-self-argument
        """
        Executes the decorated method only if we are authorized to process.
        Otherwise, apply the given :code:`default`.
        """

        def inner_metdhod(func):
            @functools.wraps(func)
            def wrapper(self, *args, **kwargs):
                if self.authorized:
                    return func(self, *args, **kwargs)  # pylint: disable=not-callable
                return default

            return wrapper

        return inner_metdhod

    @property
    def config_dir(self) -> Optional[str]:
        """
        Provides the current state of the :code:`_config_dir` attribute.
        """

        return self._config_dir

    @config_dir.setter
    def config_dir(self, value: str) -> None:
        """
        Sets the configuration directory.

        :param value:
            The value to set.

        :raise TypeError:
            When value is not a :py:class:`str`.
        """

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

        self._config_dir = value

    def set_config_dir(self, value: str) -> "DBDatasetBase":
        """
        Sets the configuration directory.

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

        self.config_dir = value

        return self

    @property
    def authorized(self) -> Optional[bool]:
        """
        Provides the current state of the :code:`_authorized` attribute.
        """

        return self._authorized

    @authorized.setter
    def authorized(self, value: bool) -> None:
        """
        Sets the value of the :code:`_authorized` 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._authorized = value

    def set_authorized(self, value: bool) -> "DBDatasetBase":
        """
        Sets the value of the :code:`_authorized` attribute.

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

        self.authorized = value

        return self

    @property
    def remove_unneeded_fields(self) -> Optional[bool]:
        """
        Provides the current state of the :code:`_remove_unneeded_fields`.
        """

        return self._remove_unneeded_fields

    @remove_unneeded_fields.setter
    def remove_unneeded_fields(self, value: bool) -> None:
        """
        Sets the value of the :code:`_remove_unneeded_fields` 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._remove_unneeded_fields = value

    def set_remove_unneeded_fields(self, value: bool) -> "DBDatasetBase":
        """
        Sets the value of the :code:`_remove_unneeded_fields` attribute.

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

        self.remove_unneeded_fields = value

        return self

    @execute_if_authorized(dict())  # pylint: disable=use-dict-literal
    def get_filtered_row(self, row: dict) -> dict:
        """
        Removes all unkowns fields (not declared) from the given row.

        :param row:
            The row to work with.
        """

        result = {}

        for key, value in row.items():
            if value is None:
                value = ""

            if key in self.FIELDS:
                result[key] = value

        for field in self.COMPARISON_FIELDS:
            if field not in result:
                result[field] = ""

        return result

    def add(self, row: dict) -> "DBDatasetBase":
        """
        Adds the given dataset into the database.

        :param row:
            The row or dataset to add.

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

        raise NotImplementedError()

    def remove(self, row: dict) -> "DBDatasetBase":
        """
        Removes the given dataset from the database.

        :param row:
            The row or dataset to remove.

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

        raise NotImplementedError()

    def update(self, row: dict, *, ignore_if_exist: bool = False) -> "DBDatasetBase":
        """
        Adds the given dataset into the database if it does not exists.
        Update otherwise.

        :param row:
            The row or dataset to manipulate.

        :param ignore_if_exist:
            Ignores the insertion/update if the row already exists.


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

        raise NotImplementedError()

    def get_content(self) -> Generator[Optional[dict], None, None]:
        """
        Provides a generator which provides the next line to read.
        """

        raise NotImplementedError()

    def cleanup(self) -> "DBDatasetBase":
        """
        Cleanups the dataset.
        """

        raise NotImplementedError()

    @execute_if_authorized(None)
    def get_filtered_content(
        self, filter_map: dict
    ) -> Generator[Optional[dict], None, None]:
        """
        Provides a generator which provides the next dataset. to read.

        :param filter_map:
            A dictionary representing what we need to filter.

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

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

        for row in self.get_content():
            for key, value in filter_map.items():
                if key not in row:
                    continue

                if row[key] == value:
                    yield row

    def exists(self, row: dict) -> bool:
        """
        Checks if the given dataset exists in our dataset.

        :param row:
            The row or dataset to add.
        """

        raise NotImplementedError()

    def are_equal(self, read_row: dict, row: dict) -> bool:
        """
        Compares the given :code:`read_row` to the `row`.

        :param read_row:
            The row read from the dataset infrastructure.
        :param row:
            The row given by the testing infrastructure.
        """

        raise NotImplementedError()

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 the base of all datasets which acts as database interface.
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	from typing import Any, Generator, List, Optional	1✔
55
56	import PyFunceble.storage	1✔
57	from PyFunceble.dataset.base import DatasetBase	1✔
58
59
60	class DBDatasetBase(DatasetBase):	1✔
61	"""
62	Provides the base of all datasets which acts as database interface.
63	"""
64
65	STD_REMOVE_UNNEEDED_FIELDS: bool = True	1✔
66	STD_AUTHORIZED: bool = False	1✔
67
68	FIELDS: List[str] = []	1✔
69	COMPARISON_FIELDS: List[str] = []	1✔
70
71	source_file: Optional[str] = None	1✔
72
73	_remove_unneeded_fields: Optional[bool] = True	1✔
74	_authorized: Optional[bool] = False	1✔
75
76	def __init__(
77	self,
78	*,
79	authorized: Optional[bool] = None,
80	remove_unneeded_fields: Optional[bool] = None,
81	config_dir: Optional[str] = None,
82	) -> None:
83	if config_dir is not None:	1✔
84	self.config_dir = config_dir	×
85	else:
86	self.config_dir = PyFunceble.storage.CONFIG_DIRECTORY	1✔
87
88	if authorized is not None:	1✔
89	self.set_authorized(authorized)	1✔
90
91	if remove_unneeded_fields is not None:	1✔
92	self.set_remove_unneeded_fields(remove_unneeded_fields)	1✔
93
94	self.__post_init__()	1✔
95
96	def __post_init__(self) -> None:	1✔
97	"""
98	A method to be called (automatically) after __init__.
99	"""
100
101	def execute_if_authorized(default: Any = None): # pylint: disable=no-self-argument	1✔
102	"""
103	Executes the decorated method only if we are authorized to process.
104	Otherwise, apply the given :code:`default`.
105	"""
106
107	def inner_metdhod(func):	1✔
108	@functools.wraps(func)	1✔
109	def wrapper(self, args, *kwargs):	1✔
110	if self.authorized:	1✔
111	return func(self, args, *kwargs) # pylint: disable=not-callable	1✔
112	return default	1✔
113
114	return wrapper	1✔
115
116	return inner_metdhod	1✔
117
118	@property	1✔
119	def config_dir(self) -> Optional[str]:	1✔
120	"""
121	Provides the current state of the :code:`_config_dir` attribute.
122	"""
123
124	return self._config_dir	1✔
125
126	@config_dir.setter	1✔
127	def config_dir(self, value: str) -> None:	1✔
128	"""
129	Sets the configuration directory.
130
131	:param value:
132	The value to set.
133
134	:raise TypeError:
135	When value is not a :py:class:`str`.
136	"""
137
138	if not isinstance(value, str):	1✔
139	raise TypeError(f"<value> should be {str}, {type(value)} given.")	×
140
141	self._config_dir = value	1✔
142
143	def set_config_dir(self, value: str) -> "DBDatasetBase":	1✔
144	"""
145	Sets the configuration directory.
146
147	:param value:
148	The value to set.
149	"""
150
151	self.config_dir = value	×
152
153	return self	×
154
155	@property	1✔
156	def authorized(self) -> Optional[bool]:	1✔
157	"""
158	Provides the current state of the :code:`_authorized` attribute.
159	"""
160
161	return self._authorized	1✔
162
163	@authorized.setter	1✔
164	def authorized(self, value: bool) -> None:	1✔
165	"""
166	Sets the value of the :code:`_authorized` attribute.
167
168	:param value:
169	The value to set.
170
171	:raise TypeError:
172	When the given :code:`value` is not a :py:class:`bool`
173	"""
174
175	if not isinstance(value, bool):	1✔
176	raise TypeError(f"<value> should be {bool}, {type(value)} given.")	1✔
177
178	self._authorized = value	1✔
179
180	def set_authorized(self, value: bool) -> "DBDatasetBase":	1✔
181	"""
182	Sets the value of the :code:`_authorized` attribute.
183
184	:param value:
185	The value to set.
186	"""
187
188	self.authorized = value	1✔
189
190	return self	1✔
191
192	@property	1✔
193	def remove_unneeded_fields(self) -> Optional[bool]:	1✔
194	"""
195	Provides the current state of the :code:`_remove_unneeded_fields`.
196	"""
197
198	return self._remove_unneeded_fields	1✔
199
200	@remove_unneeded_fields.setter	1✔
201	def remove_unneeded_fields(self, value: bool) -> None:	1✔
202	"""
203	Sets the value of the :code:`_remove_unneeded_fields` attribute.
204
205	:param value:
206	The value to set.
207
208	:raise TypeError:
209	When the given :code:`value` is not a :py:class:`bool`
210	"""
211
212	if not isinstance(value, bool):	1✔
213	raise TypeError(f"<value> should be {bool}, {type(value)} given.")	1✔
214
215	self._remove_unneeded_fields = value	1✔
216
217	def set_remove_unneeded_fields(self, value: bool) -> "DBDatasetBase":	1✔
218	"""
219	Sets the value of the :code:`_remove_unneeded_fields` attribute.
220
221	:param value:
222	The value to set.
223	"""
224
225	self.remove_unneeded_fields = value	1✔
226
227	return self	1✔
228
229	@execute_if_authorized(dict()) # pylint: disable=use-dict-literal	1✔
230	def get_filtered_row(self, row: dict) -> dict:	1✔
231	"""
232	Removes all unkowns fields (not declared) from the given row.
233
234	:param row:
235	The row to work with.
236	"""
237
238	result = {}	1✔
239
240	for key, value in row.items():	1✔
241	if value is None:	1✔
242	value = ""	1✔
243
244	if key in self.FIELDS:	1✔
245	result[key] = value	1✔
246
247	for field in self.COMPARISON_FIELDS:	1✔
248	if field not in result:	1✔
249	result[field] = ""	1✔
250
251	return result	1✔
252
253	def add(self, row: dict) -> "DBDatasetBase":	1✔
254	"""
255	Adds the given dataset into the database.
256
257	:param row:
258	The row or dataset to add.
259
260	:raise TypeError:
261	When the given :code:`row` is not a :py:class`dict`.
262	"""
263
264	raise NotImplementedError()
265
266	def remove(self, row: dict) -> "DBDatasetBase":	1✔
267	"""
268	Removes the given dataset from the database.
269
270	:param row:
271	The row or dataset to remove.
272
273	:raise TypeError:
274	When the given :code:`row` is not a :py:class`dict`.
275	"""
276
277	raise NotImplementedError()
278
279	def update(self, row: dict, *, ignore_if_exist: bool = False) -> "DBDatasetBase":	1✔
280	"""
281	Adds the given dataset into the database if it does not exists.
282	Update otherwise.
283
284	:param row:
285	The row or dataset to manipulate.
286
287	:param ignore_if_exist:
288	Ignores the insertion/update if the row already exists.
289
290
291	:raise TypeError:
292	When the given :code:`row` is not a :py:class`dict`.
293	"""
294
295	raise NotImplementedError()
296
297	def get_content(self) -> Generator[Optional[dict], None, None]:	1✔
298	"""
299	Provides a generator which provides the next line to read.
300	"""
301
302	raise NotImplementedError()
303
304	def cleanup(self) -> "DBDatasetBase":	1✔
305	"""
306	Cleanups the dataset.
307	"""
308
309	raise NotImplementedError()
310
311	@execute_if_authorized(None)	1✔
312	def get_filtered_content(	1✔
313	self, filter_map: dict
314	) -> Generator[Optional[dict], None, None]:
315	"""
316	Provides a generator which provides the next dataset. to read.
317
318	:param filter_map:
319	A dictionary representing what we need to filter.
320
321	:raise TypeError:
322	When the given :code:`filter_map` is not a :py:class:`dict`.
323	"""
324
325	if not isinstance(filter_map, dict):	1✔
326	raise TypeError(f"<filter_map> should be {dict}, {type(filter_map)} given.")	1✔
327
328	for row in self.get_content():	1✔
329	for key, value in filter_map.items():	1✔
330	if key not in row:	1✔
331	continue	1✔
332
333	if row[key] == value:	1✔
334	yield row	1✔
335
336	def exists(self, row: dict) -> bool:	1✔
337	"""
338	Checks if the given dataset exists in our dataset.
339
340	:param row:
341	The row or dataset to add.
342	"""
343
344	raise NotImplementedError()
345
346	def are_equal(self, read_row: dict, row: dict) -> bool:	1✔
347	"""
348	Compares the given :code:`read_row` to the `row`.
349
350	:param read_row:
351	The row read from the dataset infrastructure.
352	:param row:
353	The row given by the testing infrastructure.
354	"""
355
356	raise NotImplementedError()

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