18024309883

Committed 14 Sep 2025 10:06AM UTC coverage: 96.648%. Remained the same

Build # 18024309883

Build Type

push

github

Committed by

funilrys

Commit Message

Better handling of missing or empty columns.

This patch touches #430.

Contributor: @Yuki2718

Coverage Stats

0 of 2 new or added lines in 2 files covered. (0.0%)

11967 of 12382 relevant lines covered (96.65%)

14.23 hits per line

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

33.33

/PyFunceble/dataset/whois/csv.py

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

::


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

Provides the interface for the WHOIS DB CSV management.

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

import PyFunceble.cli.storage
import PyFunceble.storage
from PyFunceble.dataset.csv_base import CSVDatasetBase
from PyFunceble.dataset.whois.base import WhoisDatasetBase


class CSVWhoisDataset(CSVDatasetBase, WhoisDatasetBase):
    """
    Provides the interface for the management of the WHOIS (db)
    CSV file.
    """

    def __post_init__(self) -> None:
        self.source_file = os.path.join(
            self.config_dir, PyFunceble.cli.storage.WHOIS_DB_FILE
        )

        return super().__post_init__()

    def __contains__(self, value: str) -> bool:
        for row in self.get_content():
            try:
                if value in (row["subject"], row["idna_subject"]):
                    return True
            except KeyError:
                break

        return False

    def __getitem__(self, value: Any) -> Optional[dict]:
        try:
            for row in self.get_content():
                if value in (row["subject"], row["idna_subject"]):
                    return row
        except TypeError:
            pass

        return None

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

        for row in super().get_content():
            try:
                row["epoch"] = float(row["epoch"])
            except (TypeError, ValueError, KeyError):
                continue

            yield row

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

        ..note::
            This should be the prefered method for introduction of new dataset.

        ..warning::
            This method do nothing if the row is expired.

        :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`.
        """

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

        if not self.is_expired(row):
            if self.exists(row):
                if not ignore_if_exist and self[row["subject"]] != row:
                    self.remove(row)
                    self.add(row)
            else:
                self.add(row)

        return self

    @CSVDatasetBase.execute_if_authorized(None)
    def remove(self, row: dict) -> "CSVDatasetBase":
        """
        Removes the given dataset from the CSV file.

        :param row:
            The row or dataset to add.

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

        # If we don't do this, the comparison will fail. #
        # We don't want to overwrite the whole (remove) method just for what
        # we need.
        previous_remove_uneeded_fields = self.remove_unneeded_fields

        self.set_remove_unneeded_fields(False)

        super().remove(row)

        self.set_remove_unneeded_fields(previous_remove_uneeded_fields)

        return self

    def cleanup(self) -> "CSVWhoisDataset":
        """
        Cleanups the dataset. Meaning that we delete every entries which are
        in the past.
        """

        for row in self.get_content():
            if self.is_expired(row):
                self.remove(row)

        return self

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 interface for the WHOIS DB CSV management.
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 os	15✔
54	from typing import Any, Generator, Optional	15✔
55
56	import PyFunceble.cli.storage	15✔
57	import PyFunceble.storage	15✔
58	from PyFunceble.dataset.csv_base import CSVDatasetBase	15✔
59	from PyFunceble.dataset.whois.base import WhoisDatasetBase	15✔
60
61
62	class CSVWhoisDataset(CSVDatasetBase, WhoisDatasetBase):	15✔
63	"""
64	Provides the interface for the management of the WHOIS (db)
65	CSV file.
66	"""
67
68	def __post_init__(self) -> None:	15✔
69	self.source_file = os.path.join(	15✔
70	self.config_dir, PyFunceble.cli.storage.WHOIS_DB_FILE
71	)
72
73	return super().__post_init__()	15✔
74
75	def __contains__(self, value: str) -> bool:	15✔
76	for row in self.get_content():	×
77	try:	×
78	if value in (row["subject"], row["idna_subject"]):	×
79	return True	×
80	except KeyError:	×
81	break	×
82
83	return False	×
84
85	def __getitem__(self, value: Any) -> Optional[dict]:	15✔
86	try:	×
87	for row in self.get_content():	×
88	if value in (row["subject"], row["idna_subject"]):	×
89	return row	×
90	except TypeError:	×
91	pass	×
92
93	return None	×
94
95	@CSVDatasetBase.execute_if_authorized(None)	15✔
96	def get_content(self) -> Generator[Optional[dict], None, None]:	15✔
97	"""
98	Provides a generator which provides the next line to read.
99	"""
100
101	for row in super().get_content():	×
102	try:	×
103	row["epoch"] = float(row["epoch"])	×
NEW 104	except (TypeError, ValueError, KeyError):	×
105	continue	×
106
107	yield row	×
108
109	@CSVDatasetBase.execute_if_authorized(None)	15✔
110	def update(self, row: dict, *, ignore_if_exist: bool = False) -> "CSVWhoisDataset":	15✔
111	"""
112	Adds the given dataset into the database if it does not exists.
113	Update otherwise.
114
115	..note::
116	This should be the prefered method for introduction of new dataset.
117
118	..warning::
119	This method do nothing if the row is expired.
120
121	:param row:
122	The row or dataset to manipulate.
123
124	:param ignore_if_exist:
125	Ignores the insertion/update if the row already exists.
126
127	:raise TypeError:
128	When the given :code:`row` is not a :py:class`dict`.
129	"""
130
131	if not isinstance(row, dict):	×
132	raise TypeError(f"<row> should be {dict}, {type(row)} given.")	×
133
134	if not self.is_expired(row):	×
135	if self.exists(row):	×
136	if not ignore_if_exist and self[row["subject"]] != row:	×
137	self.remove(row)	×
138	self.add(row)	×
139	else:
140	self.add(row)	×
141
142	return self	×
143
144	@CSVDatasetBase.execute_if_authorized(None)	15✔
145	def remove(self, row: dict) -> "CSVDatasetBase":	15✔
146	"""
147	Removes the given dataset from the CSV file.
148
149	:param row:
150	The row or dataset to add.
151
152	:raise TypeError:
153	When the given :code:`row` is not a :py:class`dict`.
154	"""
155
156	# If we don't do this, the comparison will fail. #
157	# We don't want to overwrite the whole (remove) method just for what
158	# we need.
159	previous_remove_uneeded_fields = self.remove_unneeded_fields	×
160
161	self.set_remove_unneeded_fields(False)	×
162
163	super().remove(row)	×
164
165	self.set_remove_unneeded_fields(previous_remove_uneeded_fields)	×
166
167	return self	×
168
169	def cleanup(self) -> "CSVWhoisDataset":	15✔
170	"""
171	Cleanups the dataset. Meaning that we delete every entries which are
172	in the past.
173	"""
174
175	for row in self.get_content():	×
176	if self.is_expired(row):	×
177	self.remove(row)	×
178
179	return self	×

funilrys / PyFunceble / 18024309883

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