libs.data.mysql_checker() - Code Metrics - Inspection of "To use asyncio correctly" - star-inc/pbp-analytics - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 4b6ab6...72d98f )

by Randy

created 2020-05-27 12:38 UTC

libs.data.mysql_checker() A

↳ Parent: libs.data

Complexity

Conditions

Size

Total Lines	19
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	14
dl	0
loc	19
rs	9.7
c	0
b	0
f	0
cc	3
nop	1

from functools import wraps
from hashlib import sha256

import mysql.connector as sql_client

"""
    Copyright (c) 2020 Star Inc.(https://starinc.xyz)

    This Source Code Form is subject to the terms of the Mozilla Public
    License, v. 2.0. If a copy of the MPL was not distributed with this
    file, You can obtain one at http://mozilla.org/MPL/2.0/.
"""


def mysql_checker(function):
    """
    To wrap a function for preventing to disconnect from the database
    """

    @wraps(function)
    def wrapper(*args, **kwargs):
        try:
            result = function(*args, **kwargs)
            args[0].db_error_checkpoint = 0
            return result
        except sql_client.errors.OperationalError:
            if args[0].db_error_checkpoint:
                assert "sql_error"
            args[0].db_error_checkpoint = 1
            args[0].__init__(args[0].handle)
            return function(*args)

    return wrapper


class Data:
    """
    To control MySQL for PBP
    """

    db_error_checkpoint = 0

    def __init__(self, pbp_handle):
        """
        Configure and initialize database details

        :param pbp_handle: Analytics object
        """
        self.handle = pbp_handle
        self.db_client = sql_client.connect(**pbp_handle.cfg["MySQL"])

    @mysql_checker
    def check_trustlist(self, url: str):
        """
        To check URL whether exists in trustlist

        :param url: URL
        :return: string of UUID or NoneType
        """
        cursor = self.db_client.cursor(dictionary=True)
        cursor.execute(
            "SELECT `uuid` FROM `trustlist` WHERE `url` = %s",
            (url,)
        )
        result = cursor.fetchall()
        self.db_client.commit()
        cursor.close()
        if result:
            return result[0]

    @mysql_checker
    def check_trust_domain(self, domain: str):
        """
        To check URL whether exists in trust_domain list

        :param domain: domain
        :return: string of UUID or NoneType
        """
        cursor = self.db_client.cursor(dictionary=True)
        cursor.execute(
            "SELECT `uuid` FROM `trust_domain` WHERE `domain` = %s",
            (domain,)
        )
        result = cursor.fetchall()
        self.db_client.commit()
        cursor.close()
        if result:
            return result[0]

    @mysql_checker
    def check_blacklist(self, url: str):
        """
        To check URL whether exists in blacklist

        :param url: URL
        :return: dict of URL and Mark-Date or NoneType
        """
        cursor = self.db_client.cursor(dictionary=True)
        cursor.execute(
            "SELECT `url`, `date` FROM `blacklist` WHERE `url` = %s",
            (url,)
        )
        result = cursor.fetchall()
        self.db_client.commit()
        cursor.close()
        if result:
            return result[0]

    @mysql_checker
    def check_warnlist(self, url: str):
        """
        To check URL whether exists in warnlist

        :param url: URL
        :return: dict of URL, similar URL and Mark-Date or NoneType
        """
        cursor = self.db_client.cursor(dictionary=True)
        cursor.execute(
            "SELECT `url`, `origin`, `date` FROM `warnlist` WHERE `url` = %s",
            (url,)
        )
        result = cursor.fetchall()
        self.db_client.commit()
        cursor.close()
        if result:
            return result[0]

    @mysql_checker
    def get_urls_from_trustlist(self):
        """
        Fetch all URL in trustlist

        :return: list of URL
        """
        cursor = self.db_client.cursor()
        cursor.execute(
            "SELECT `url` FROM `trustlist`"
        )
        result = cursor.fetchall()
        self.db_client.commit()
        cursor.close()
        if result:
            return [record[0] for record in result]
        return []

    @mysql_checker
    def get_view_narray_from_trustlist(self):
        """
        Fetch all target_view_narray in trustlist

        :return: dict of URL and NumPy Array
        """
        cursor = self.db_client.cursor(dictionary=True)
        cursor.execute(
            "SELECT `url`, `target_view_narray` FROM `trustlist`"
        )
        result = cursor.fetchall()
        self.db_client.commit()
        cursor.close()
        return result

    @mysql_checker
    def find_page_by_view_signature(self, signature: str):
        """
        Search URL by view_signature in trustlist

        :param signature: string hashed
        :return: URL or NoneType
        """
        cursor = self.db_client.cursor()
        cursor.execute(
            "SELECT `url` FROM `trustlist` WHERE `target_view_signature` = %s",
            (signature,)
        )
        result = cursor.fetchall()
        self.db_client.commit()
        cursor.close()
        if result:
            return result[0][0]
        return None

    @mysql_checker
    def find_result_cache_by_url_hash(self, url_hash: str):
        """
        Search cache by url_hash in result_cache

        :param url_hash: URL hashed
        :return: float of the-trust-score or NoneType
        """
        cursor = self.db_client.cursor()
        cursor.execute(
            "SELECT `score` FROM `result_cache` WHERE `url_hash` = %s",
            (url_hash,)
        )
        result = cursor.fetchall()
        self.db_client.commit()
        cursor.close()
        if result:
            return result[0][0]
        return None

    @mysql_checker
    def upload_result_cache(self, url_hash: str, score: float):
        """
        Upload the-trust-score to cache

        :param url_hash: URL hashed
        :param score: float of the-trust-score
        :return:
        """
        cursor = self.db_client.cursor()
        cursor.execute(
            "INSERT INTO `result_cache`(`url_hash`, `score`) VALUES (%s, %s)",
            (url_hash, score)
        )
        self.db_client.commit()
        cursor.close()
        return True

    @mysql_checker
    def clean_result_cache(self):
        """
        Clean result caches

        :return: True
        """
        cursor = self.db_client.cursor()
        cursor.execute("TRUNCATE TABLE `result_cache`")
        self.db_client.commit()
        cursor.close()
        return True

    @mysql_checker
    def upload_view_sample(self, url: str, view_signature: str, view_data: str):
        """
        Upload ViewSample for PageView

        :param url: URL of Sample
        :param view_signature: string hashed with view_data
        :param view_data: string of num array base64 encoded
        :return: True
        """
        cursor = self.db_client.cursor()
        if self.check_trustlist(url):
            cursor.execute(
                "UPDATE `trustlist` SET `target_view_signature` = %s, `target_view_narray` = %s WHERE `url` = %s",
                (view_signature, view_data, url)
            )
        else:
            cursor.execute(
                "INSERT INTO "
                "`trustlist`(`uuid`, `url`, `target_view_signature`, `target_view_narray`) "
                "VALUES (UUID(), %s, %s, %s)",
                (url, view_signature, view_data)
            )
        self.db_client.commit()
        cursor.close()
        return True

    @mysql_checker
    def mark_as_blacklist(self, url: str):
        """
        Mark URL to blacklist by Database

        :param url: URL to mark
        :return: True
        """
        cursor = self.db_client.cursor()
        cursor.execute(
            "INSERT INTO `blacklist`(`uuid`, `url`, `date`, `url_hash`) VALUES (UUID(), %s, NOW(), %s)",
            (url, sha256(url.encode("utf-8")).hexdigest(),)
        )
        self.db_client.commit()
        cursor.close()
        return True

    @mysql_checker
    def mark_as_blacklist_mass(self, urls: list):
        """
        Mark URLs to blacklist by Database

        :param url: URLs to mark
        :return: True
        """
        cursor = self.db_client.cursor()
        datas = [(url, sha256(url.encode("utf-8")).hexdigest())
                 for url in urls]
        cursor.executemany(
            "INSERT INTO `blacklist`(`uuid`, `url`, `date`, `url_hash`) VALUES (UUID(), %s, NOW(), %s)",
            datas
        )
        self.db_client.commit()
        cursor.close()
        return True

    @mysql_checker
    def mark_as_warnlist(self, url: str, origin_url: str):
        """
        Mark URL to warnlist by PageView

        :param url: URL to mark
        :param origin_url: the URL similar to
        :return: True
        """
        cursor = self.db_client.cursor()
        cursor.execute(
            "INSERT INTO `warnlist`(`uuid`, `url`, `origin`, `date`) VALUES (UUID(), %s, %s, NOW())",
            (url, origin_url)
        )
        self.db_client.commit()
        cursor.close()
        return True


1			from functools import wraps
2			from hashlib import sha256
3
4			import mysql.connector as sql_client
5
6			"""
7			Copyright (c) 2020 Star Inc.(https://starinc.xyz)
8
9			This Source Code Form is subject to the terms of the Mozilla Public
10			License, v. 2.0. If a copy of the MPL was not distributed with this
11			file, You can obtain one at http://mozilla.org/MPL/2.0/.
12			"""
13
14
15			def mysql_checker(function):
16			"""
17			To wrap a function for preventing to disconnect from the database
18			"""
19
20			@wraps(function)
21			def wrapper(args, *kwargs):
22			try:
23			result = function(args, *kwargs)
24			args[0].db_error_checkpoint = 0
25			return result
26			except sql_client.errors.OperationalError:
27			if args[0].db_error_checkpoint:
28			assert "sql_error"
29			args[0].db_error_checkpoint = 1
30			args[0].__init__(args[0].handle)
31			return function(*args)
32
33			return wrapper
34
35
36			class Data:
37			"""
38			To control MySQL for PBP
39			"""
40
41			db_error_checkpoint = 0
42
43			def __init__(self, pbp_handle):
44			"""
45			Configure and initialize database details
46
47			:param pbp_handle: Analytics object
48			"""
49			self.handle = pbp_handle
50			self.db_client = sql_client.connect(**pbp_handle.cfg["MySQL"])
51
52			@mysql_checker
53			def check_trustlist(self, url: str):
54			"""
55			To check URL whether exists in trustlist
56
57			:param url: URL
58			:return: string of UUID or NoneType
59			"""
60			cursor = self.db_client.cursor(dictionary=True)
61			cursor.execute(
62			"SELECT `uuid` FROM `trustlist` WHERE `url` = %s",
63			(url,)
64			)
65			result = cursor.fetchall()
66			self.db_client.commit()
67			cursor.close()
68			if result:
69			return result[0]
70
71			@mysql_checker
72			def check_trust_domain(self, domain: str):
73			"""
74			To check URL whether exists in trust_domain list
75
76			:param domain: domain
77			:return: string of UUID or NoneType
78			"""
79			cursor = self.db_client.cursor(dictionary=True)
80			cursor.execute(
81			"SELECT `uuid` FROM `trust_domain` WHERE `domain` = %s",
82			(domain,)
83			)
84			result = cursor.fetchall()
85			self.db_client.commit()
86			cursor.close()
87			if result:
88			return result[0]
89
90			@mysql_checker
91			def check_blacklist(self, url: str):
92			"""
93			To check URL whether exists in blacklist
94
95			:param url: URL
96			:return: dict of URL and Mark-Date or NoneType
97			"""
98			cursor = self.db_client.cursor(dictionary=True)
99			cursor.execute(
100			"SELECT `url`, `date` FROM `blacklist` WHERE `url` = %s",
101			(url,)
102			)
103			result = cursor.fetchall()
104			self.db_client.commit()
105			cursor.close()
106			if result:
107			return result[0]
108
109			@mysql_checker
110			def check_warnlist(self, url: str):
111			"""
112			To check URL whether exists in warnlist
113
114			:param url: URL
115			:return: dict of URL, similar URL and Mark-Date or NoneType
116			"""
117			cursor = self.db_client.cursor(dictionary=True)
118			cursor.execute(
119			"SELECT `url`, `origin`, `date` FROM `warnlist` WHERE `url` = %s",
120			(url,)
121			)
122			result = cursor.fetchall()
123			self.db_client.commit()
124			cursor.close()
125			if result:
126			return result[0]
127
128			@mysql_checker
129			def get_urls_from_trustlist(self):
130			"""
131			Fetch all URL in trustlist
132
133			:return: list of URL
134			"""
135			cursor = self.db_client.cursor()
136			cursor.execute(
137			"SELECT `url` FROM `trustlist`"
138			)
139			result = cursor.fetchall()
140			self.db_client.commit()
141			cursor.close()
142			if result:
143			return [record[0] for record in result]
144			return []
145
146			@mysql_checker
147			def get_view_narray_from_trustlist(self):
148			"""
149			Fetch all target_view_narray in trustlist
150
151			:return: dict of URL and NumPy Array
152			"""
153			cursor = self.db_client.cursor(dictionary=True)
154			cursor.execute(
155			"SELECT `url`, `target_view_narray` FROM `trustlist`"
156			)
157			result = cursor.fetchall()
158			self.db_client.commit()
159			cursor.close()
160			return result
161
162			@mysql_checker
163			def find_page_by_view_signature(self, signature: str):
164			"""
165			Search URL by view_signature in trustlist
166
167			:param signature: string hashed
168			:return: URL or NoneType
169			"""
170			cursor = self.db_client.cursor()
171			cursor.execute(
172			"SELECT `url` FROM `trustlist` WHERE `target_view_signature` = %s",
173			(signature,)
174			)
175			result = cursor.fetchall()
176			self.db_client.commit()
177			cursor.close()
178			if result:
179			return result[0][0]
180			return None
181
182			@mysql_checker
183			def find_result_cache_by_url_hash(self, url_hash: str):
184			"""
185			Search cache by url_hash in result_cache
186
187			:param url_hash: URL hashed
188			:return: float of the-trust-score or NoneType
189			"""
190			cursor = self.db_client.cursor()
191			cursor.execute(
192			"SELECT `score` FROM `result_cache` WHERE `url_hash` = %s",
193			(url_hash,)
194			)
195			result = cursor.fetchall()
196			self.db_client.commit()
197			cursor.close()
198			if result:
199			return result[0][0]
200			return None
201
202			@mysql_checker
203			def upload_result_cache(self, url_hash: str, score: float):
204			"""
205			Upload the-trust-score to cache
206
207			:param url_hash: URL hashed
208			:param score: float of the-trust-score
209			:return:
210			"""
211			cursor = self.db_client.cursor()
212			cursor.execute(
213			"INSERT INTO `result_cache`(`url_hash`, `score`) VALUES (%s, %s)",
214			(url_hash, score)
215			)
216			self.db_client.commit()
217			cursor.close()
218			return True
219
220			@mysql_checker
221			def clean_result_cache(self):
222			"""
223			Clean result caches
224
225			:return: True
226			"""
227			cursor = self.db_client.cursor()
228			cursor.execute("TRUNCATE TABLE `result_cache`")
229			self.db_client.commit()
230			cursor.close()
231			return True
232
233			@mysql_checker
234			def upload_view_sample(self, url: str, view_signature: str, view_data: str):
235			"""
236			Upload ViewSample for PageView
237
238			:param url: URL of Sample
239			:param view_signature: string hashed with view_data
240			:param view_data: string of num array base64 encoded
241			:return: True
242			"""
243			cursor = self.db_client.cursor()
244			if self.check_trustlist(url):
245			cursor.execute(
246			"UPDATE `trustlist` SET `target_view_signature` = %s, `target_view_narray` = %s WHERE `url` = %s",
247			(view_signature, view_data, url)
248			)
249			else:
250			cursor.execute(
251			"INSERT INTO "
252			"`trustlist`(`uuid`, `url`, `target_view_signature`, `target_view_narray`) "
253			"VALUES (UUID(), %s, %s, %s)",
254			(url, view_signature, view_data)
255			)
256			self.db_client.commit()
257			cursor.close()
258			return True
259
260			@mysql_checker
261			def mark_as_blacklist(self, url: str):
262			"""
263			Mark URL to blacklist by Database
264
265			:param url: URL to mark
266			:return: True
267			"""
268			cursor = self.db_client.cursor()
269			cursor.execute(
270			"INSERT INTO `blacklist`(`uuid`, `url`, `date`, `url_hash`) VALUES (UUID(), %s, NOW(), %s)",
271			(url, sha256(url.encode("utf-8")).hexdigest(),)
272			)
273			self.db_client.commit()
274			cursor.close()
275			return True
276
277			@mysql_checker
278			def mark_as_blacklist_mass(self, urls: list):
279			"""
280			Mark URLs to blacklist by Database
281
282			:param url: URLs to mark
283			:return: True
284			"""
285			cursor = self.db_client.cursor()
286			datas = [(url, sha256(url.encode("utf-8")).hexdigest())
287			for url in urls]
288			cursor.executemany(
289			"INSERT INTO `blacklist`(`uuid`, `url`, `date`, `url_hash`) VALUES (UUID(), %s, NOW(), %s)",
290			datas
291			)
292			self.db_client.commit()
293			cursor.close()
294			return True
295
296			@mysql_checker
297			def mark_as_warnlist(self, url: str, origin_url: str):
298			"""
299			Mark URL to warnlist by PageView
300
301			:param url: URL to mark
302			:param origin_url: the URL similar to
303			:return: True
304			"""
305			cursor = self.db_client.cursor()
306			cursor.execute(
307			"INSERT INTO `warnlist`(`uuid`, `url`, `origin`, `date`) VALUES (UUID(), %s, %s, NOW())",
308			(url, origin_url)
309			)
310			self.db_client.commit()
311			cursor.close()
312			return True
313

star-inc / pbp-analytics

Push — master ( 4b6ab6...72d98f )

libs.data.mysql_checker() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like