Разматывать укороченные URL-адреса с помощью Python, SQLite, и многопоточность


Я провел последние пару дней чтения/исследования и обучение SQLite и язык Python многопоточность.

Я разматывать укороченные URL-адреса, хранящиеся в SQL таблицу с тремя полями...
идентификатор | url_starting | url_unwound

Код чанков в базе данных, отправляет кусок в какой-нить бассейн такого же размера кусок, разматывает URL и записывает результаты обратно в базу данных, а затем переходит к следующему блоку.

Я доволен результатами, но я ищу обратную связь, так как это моя первая попытка.

Общие мысли высоко ценятся.
Кроме того, любые идеи о том, что делает future_to_url = раздел более читабельным было бы здорово.

import time
import os
import sqlite3
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.request import Request, urlopen


class MyDatabase(object):
    def __init__(self, db):
        self.__conn = sqlite3.connect(db)
        self.__conn.row_factory = sqlite3.Row  # https://stackoverflow.com/a/41920171/25197
        self.__conn.execute('pragma foreign_keys = on')
        self.__conn.commit()
        self.__cursor = self.__conn.cursor()

    def __enter__(self):
        return self

    def __del__(self):
        self.__conn.close()

    def __exit__(self, exc_type, exc_value, traceback):
        self.__conn.close()

    def query(self, arg):
        self.__cursor.execute(arg)
        self.__conn.commit()
        return self.__cursor

    def updateUnwoundUrl(self, url_starting, url_unwound):
        self.__cursor.execute(
            'UPDATE url SET url_unwound = ? WHERE url_starting = ?', (url_unwound, url_starting))
        self.__conn.commit()
        return self.__cursor


def unwindUrl(starting_url, timeout):
    try:
        hdr = {'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64)'}
        response = urlopen(Request(starting_url, headers=hdr), timeout=timeout)
        unwound_url = response.geturl()

    # For any http error just return error. Deal with specifics later.
    except Exception:
        unwound_url = 'error'

    return unwound_url


def chunk_iter(iterable, chunk_size):
    it = iter(iterable)
    while True:
        chunk = tuple(next(it) for _ in range(chunk_size))
        if not chunk:
            break
        yield chunk


def unwindUrls(workers=100, request_timeout=10):
    relative_path = os.path.dirname(__file__)
    my_db_path = os.path.join(relative_path, 'database', 'my.db')

    with MyDatabase(my_db_path) as my_db:
        wound_urls = my_db.query(
            'SELECT url_starting, id FROM url WHERE url_unwound IS NULL ORDER BY id ASC').fetchall()

        chunk_count = len(wound_urls) / workers
        for i, wound_urls_chunk in enumerate(chunk_iter(wound_urls, workers), start=1):

            print('starting chunk {0} of {1} at {2}'.format(
                i, chunk_count, datetime.now().strftime("%I:%M:%S %p")))

            with ThreadPoolExecutor(max_workers=workers) as executor:
                future_to_url = {executor.submit(
                    unwindUrl, row['url_starting'], request_timeout): row['url_starting'] for row in wound_urls_chunk}

                for future in as_completed(future_to_url):
                    try:
                        url_starting = future_to_url[future]
                        url_unwound = future.result()
                        my_db.updateUnwoundUrl(url_starting, url_unwound)
                        # print('starting: {0} unwound: {1}'.format(url_starting, url_unwound))
                    except Exception as exc:
                        print('!! {0}'.format(exc))


if __name__ == '__main__':
    startTime = time.time()
    unwindUrls()
    print('==--- unwind_urls took {0} seconds. ---=='.format(round(time.time() - startTime)))


Комментарии
2 ответа

__

Есть маленькая необходимость __. Один _ должно быть достаточно, чтобы заставить людей ясно, что это отдельный атрибут. Если они действительно хотите использовать его, __ не прекратить доступ

Linelength

Особенно строки с запросами слишком долго. Разве вы не можете сделать эти запросы многострочный?

От:

def updateUnwoundUrl(self, url_starting, url_unwound):
self.__cursor.execute(
'UPDATE url SET url_unwound = ? WHERE url_starting = ?', (url_unwound, url_starting))
self.__conn.commit()
return self.__cursor

к:

def updateUnwoundUrl(self, url_starting, url_unwound):
self.__cursor.execute(
'''UPDATE url
SET url_unwound = ?
WHERE url_starting = ?
'''
, (url_unwound, url_starting)
)
self.__conn.commit()
return self.__cursor

Подъем НЛ

Если когда-либо вы хотите изменить дБ формат на что-то другое, нужно изменить внутреннее устройство кода и как my_db_path сделан.

Чтобы изменить это, я бы следуйте советам из этого разговора.

Путь к базе данных

def unwindUrls(workers=100, request_timeout=10):
relative_path = os.path.dirname(__file__)
my_db_path = os.path.join(relative_path, 'database', 'my.db')

Во-первых, я хотел бы использовать my_db_path для обработки пути к каталогам.

Во-вторых, я бы добавил relative_path в качестве аргумента, с Path('.') по умолчанию аргумент, таким способом можно указать путь.

def unwindUrls(workers=100, request_timeout=10):
relative_path = os.path.dirname(__file__)
my_db_path = os.path.join(relative_path, 'database', 'my.db')
with MyDatabase(my_db_path) as my_db:

...
if __name__ == '__main__':
unwindUrls()

К:

def unwindUrls(my_db, workers=100, request_timeout=10):

...
if __name__ == '__main__':
my_db_path = pathlib.Path('.') / 'database' / 'my.db'
with MyDatabase(my_db_path) as my_db:
unwindUrls(my_db)

chunk_iter

def chunk_iter(iterable, chunk_size):
it = iter(iterable)
while True:
chunk = tuple(next(it) for _ in range(chunk_size))
if not chunk:
break
yield chunk

поднимает Warning в Python 3.6


C:\miniconda3\envs\test_env\lib\site-packages\ipykernel_launcher.py:4:
DeprecationWarning: генератор 'chunk_iter..' поднял
StopIteration после удаления ухо из sys.путь.

Лучше использовать itertools.islice

def chunk_iter(iterable, chunk_size):
it = iter(iterable)
while True:
chunk = tuple(islice(it, chunk_size))
if not chunk:
return
yield chunk

Исключение

несколько раз, когда вы используете except Exception. Это слишком широкий. Вы должны указать, какие исключения вы хотите поймать

4
ответ дан 5 марта 2018 в 10:03 Источник Поделиться

future_to_url =

Ответить на свой собственный вопрос по этой части...
Я в конечном итоге переименование некоторых переменных и используя цикл вместо понимания, чтобы заполнить словарь.
Я чувствую, что это более читабельным в этом случае.

Вместо этого...

with ThreadPoolExecutor(max_workers=workers) as executor:
future_to_url = {executor.submit(
unwindUrl, row['url_starting'], request_timeout): row['url_starting'] for row in wound_urls_chunk}

for future in as_completed(future_to_url):
try:
url_starting = future_to_url[future]
url_unwound = future.result()
my_db.updateUnwoundUrl(url_starting, url_unwound)
except Exception as exc:
print('!! {0}'.format(exc))

Это...

with ThreadPoolExecutor(max_workers=workers) as executor:
thread_pool = {}
for row in wound_urls_chunk:
unwindUrl_thread = executor.submit(unwindUrl, row['url_starting'], timeout)
thread_pool[unwindUrl_thread] = row['url_starting']

for completed_thread in as_completed(thread_pool):
try:
url_starting = thread_pool[completed_thread]
url_unwound = completed_thread.result()
my_db.updateUnwoundUrl(url_starting, url_unwound)
except Exception as exc:
print('!! {0}'.format(exc))

1
ответ дан 5 марта 2018 в 10:03 Источник Поделиться