From f70e908364448e6595aaa9fee3791fa0f12fc563 Mon Sep 17 00:00:00 2001 From: SergeiNaum Date: Wed, 11 Oct 2023 10:46:53 +0300 Subject: [PATCH] after 4th code review --- page_analyzer/db.py | 156 +++++++++++++++++++++----------------------- 1 file changed, 76 insertions(+), 80 deletions(-) diff --git a/page_analyzer/db.py b/page_analyzer/db.py index 89d48fc..37140e4 100644 --- a/page_analyzer/db.py +++ b/page_analyzer/db.py @@ -22,113 +22,109 @@ def close_connection(conn): @log def add_url(conn: connection, url_name: str) -> NamedTuple: - with conn.cursor(cursor_factory=NamedTupleCursor) as cur: - try: - cur.execute( - 'INSERT INTO urls (name, created_at) VALUES (%s, %s) RETURNING id;', - (url_name, datetime.now()) - ) - id = cur.fetchone() - conn.commit() + cur = conn.cursor(cursor_factory=NamedTupleCursor) + try: + cur.execute( + 'INSERT INTO urls (name, created_at) VALUES (%s, %s) RETURNING id;', + (url_name, datetime.now()) + ) + id = cur.fetchone() + conn.commit() - return id.id - except psycopg2.DatabaseError as e: - raise Exception('An error occurred while adding the URL') from e + return id.id + except psycopg2.DatabaseError as e: + raise Exception('An error occurred while adding the URL') from e @log def create_url_check(conn: connection, url, status_code: int, tags_data: dict): - with conn.cursor(cursor_factory=NamedTupleCursor) as cur: - try: - cur.execute( - '''INSERT INTO url_checks - (url_id, status_code, h1, title, description, created_at) - VALUES (%s, %s, %s, %s, %s, %s);''', - ( - url.id, - status_code, - tags_data['h1'], - tags_data['title'], - tags_data['description'], - datetime.now(), - ), - ) - conn.commit() - except psycopg2.DatabaseError as e: - raise Exception('An error occurred while creating URL check !') from e + cur = conn.cursor(cursor_factory=NamedTupleCursor) + try: + cur.execute( + '''INSERT INTO url_checks + (url_id, status_code, h1, title, description, created_at) + VALUES (%s, %s, %s, %s, %s, %s);''', + ( + url.id, + status_code, + tags_data['h1'], + tags_data['title'], + tags_data['description'], + datetime.now(), + ), + ) + conn.commit() + except psycopg2.DatabaseError as e: + raise Exception('An error occurred while creating URL check !') from e @log def get_url_by_url_name(conn: connection, url_name: str) -> NamedTuple: - with conn: - with conn.cursor(cursor_factory=NamedTupleCursor) as cur: - try: - cur.execute('SELECT * FROM urls WHERE name = %s LIMIT 1;', (url_name,), ) - url = cur.fetchone() + cur = conn.cursor(cursor_factory=NamedTupleCursor) + try: + cur.execute('SELECT * FROM urls WHERE name = %s LIMIT 1;', (url_name,), ) + url = cur.fetchone() - return url + return url - except psycopg2.DatabaseError as e: - raise Exception('An error occurred while getting URL by URL name !') from e + except psycopg2.DatabaseError as e: + raise Exception('An error occurred while getting URL by URL name !') from e @log def get_urls_and_last_checks_data(conn: connection) -> List[NamedTuple]: - with conn: - with conn.cursor(cursor_factory=NamedTupleCursor) as cur: - try: + cur = conn.cursor(cursor_factory=NamedTupleCursor) + try: - cur.execute( - 'SELECT DISTINCT ON (urls.id) id, name FROM urls;' - ) - data_urls = cur.fetchall() - cur.execute( - '''SELECT DISTINCT ON (url_checks.url_id) url_id, status_code, created_at - FROM url_checks;''' - ) - url_checks = cur.fetchall() - urls_dict = {url.id: url for url in data_urls} - record = namedtuple('Record', ['id', 'name', 'status_code', 'created_at']) + cur.execute( + 'SELECT DISTINCT ON (urls.id) id, name FROM urls;' + ) + data_urls = cur.fetchall() + cur.execute( + '''SELECT DISTINCT ON (url_checks.url_id) url_id, status_code, created_at + FROM url_checks;''' + ) + url_checks = cur.fetchall() + urls_dict = {url.id: url for url in data_urls} + record = namedtuple('Record', ['id', 'name', 'status_code', 'created_at']) - data = [] + data = [] - for check in url_checks: - url = urls_dict.get(check.url_id) - if url: - data.append(record(url.id, url.name, check.status_code, check.created_at)) - data = sorted(data, key=lambda rec: (rec.id, rec.created_at), reverse=True) + for check in url_checks: + url = urls_dict.get(check.url_id) + if url: + data.append(record(url.id, url.name, check.status_code, check.created_at)) + data = sorted(data, key=lambda rec: (rec.id, rec.created_at), reverse=True) - return data + return data - except psycopg2.DatabaseError as e: - raise Exception('An error occurred while getting URLs & last checks !') from e + except psycopg2.DatabaseError as e: + raise Exception('An error occurred while getting URLs & last checks !') from e @log def get_url_by_id(conn: connection, url_id) -> NamedTuple: - with conn: - with conn.cursor(cursor_factory=NamedTupleCursor) as cur: - try: - cur.execute('SELECT * FROM urls WHERE id = %s LIMIT 1;', (url_id,), ) - url = cur.fetchone() + cur = conn.cursor(cursor_factory=NamedTupleCursor) + try: + cur.execute('SELECT * FROM urls WHERE id = %s LIMIT 1;', (url_id,), ) + url = cur.fetchone() - return url + return url - except psycopg2.DatabaseError as e: - raise Exception('An error occurred while getting URL by ID !') from e + except psycopg2.DatabaseError as e: + raise Exception('An error occurred while getting URL by ID !') from e @log def get_url_checks_by_url_id(conn: connection, url_id) -> NamedTuple: - with conn: - with conn.cursor(cursor_factory=NamedTupleCursor) as cur: - try: - cur.execute('''SELECT * - FROM url_checks - WHERE url_id = %s - ORDER BY id DESC;''', (url_id,), ) - url_checks = cur.fetchall() - return url_checks - - except psycopg2.DatabaseError as e: - raise Exception('An error occurred while getting URL checks by URL ID!') from e + cur = conn.cursor(cursor_factory=NamedTupleCursor) + try: + cur.execute('''SELECT * + FROM url_checks + WHERE url_id = %s + ORDER BY id DESC;''', (url_id,), ) + url_checks = cur.fetchall() + return url_checks + + except psycopg2.DatabaseError as e: + raise Exception('An error occurred while getting URL checks by URL ID!') from e