2021-12-20 / Bartłomiej Kurek
Crawler - asynchronicznie (#2 - argparse, opcje programu)

W tej części materiału zajmujemy się opcjami programu. Zaimplementujemy parser argumentów przy użyciu argparse oraz podstawowy test jednostkowy (unittest) do tej części programu. Dla niecierpliwych - gotowy kod parsera zobaczyć można tutaj. Poniżej omawiam od podstaw implementację i niuanse techniczne.

Jeśli komuś argparse nie jest jeszcze znany - można potraktować to jako mini tutorial.
Osobom, które już argparse znają, artykuł może wydać się "banalny". Zajmujemy się jednak implementacją crawlera, a opcje są jego ważną częścią. Przegląd implementacji opcji programu może nam dać zatem pełniejsze zrozumienie jego działania oraz wyzwań implementacyjnych.

argparse

Moduł argparse to moduł biblioteki standardowej Python. Pojawił się już w wersji Python 3.2 zastępując wcześniej używany OptParse. Moduł ten pozwala nam na wygodne budowanie opcji programu, dziedziczenie parserów, czy użycie typów plikowych zgodnie ze standardową konwencją systemów operacyjnych.

Opcje programu możemy grupować. Możemy również budować opcje używając subkomend - podobnie jak np. w narzędziu git, gdzie mamy git status, git commit, czy git push, a każda z subkomend dostarcza swoje własne opcje i przełączniki. W omawianej implementacji zajmiemy się jedynie podstawowym zakresem wykorzystania modułu argparse, gdyż program jest jednoplikowy, a opcji niewiele.

Implementacja opcji

Zaczynamy od stworzenia obiektu parsera:

parser = argparse.ArgumentParser()

Dla ciekawskich - można to zrobić w interaktywnym trybie Python, a po wciśnięciu tabulatora interpreter pokaże nam nam wszystkie metody stworzonego obiektu:

$ python -q
>>> import argparse
>>> parser = argparse.ArgumentParser()
>>> parser.
parser.add_argument(                  parser.format_usage(
parser.add_argument_group(            parser.formatter_class(
parser.add_help                       parser.fromfile_prefix_chars
parser.add_mutually_exclusive_group(  parser.get_default(
parser.add_subparsers(                parser.parse_args(
parser.allow_abbrev                   parser.parse_intermixed_args(
parser.argument_default               parser.parse_known_args(
parser.conflict_handler               parser.parse_known_intermixed_args(
parser.convert_arg_line_to_args(      parser.prefix_chars
parser.description                    parser.print_help(
parser.epilog                         parser.print_usage(
parser.error(                         parser.prog
parser.exit(                          parser.register(
parser.exit_on_error                  parser.set_defaults(
parser.format_help(                   parser.usage
>>> parser.

Stworzyliśmy parser, zaczynamy dodawanie argumentów programu.
Program ma przyjąć opcje (krótkie, jak "-s", oraz długie, jak "--size-limit-kb").
Dla wygody podzielę opcje parsera na grupy odpowiadające komponentom crawlera: Flags, Cache, Scheduler, Downloader oraz argumenty pozycyjne.

parser = argparse.ArgumentParser()
flags = parser.add_argument_group("Flags")
cache = parser.add_argument_group("Cache")
scheduler = parser.add_argument_group("Scheduler")
downloader = parser.add_argument_group("Downloader")

Flagi

Flagi to takie opcje programu, które są włączone, albo nie. Nie przyjmują od użytkownika wartości, a są jedynie przełącznikami. Są więc z założenia wartościami typu boolean ("pstryczek" on/off).
Nasz program będzie obsługiwał dwie flagi:

  • kolorowanie logów: (opcja krótka: "-C", opcja długa: "--color")
  • tryb debug: (opcja krótka: "-D", opcja długa: "--debug")
    # Flags
    flags = parser.add_argument_group("Flags")
    flags.add_argument(
        "-C", "--color", action="store_true",
        help="Color logs",
    )
    flags.add_argument(
        "-D", "--debug", action="store_true",
        help="Display debug information",
    )

Przy implementacji flag w argparse podajemy action, które może być store_true, lub store_false. Moduł sam rozpozna, że jeśli podajemy store_true, to domyślnie flaga jest wyłączona, a jej pojawienie się oznacza, że opcję włączamy. Oczywiście store_false działałoby odwrotnie.
Do opcji dodaję również parametr help, które moduł nam ładnie wyświetli i sformatuje w przypadku wywołania crawler.py --help. Opcja --help oraz jej krótki odpowiednik -h są automatycznie dodawane przez ArgumentParser i zgodne z konwencją unix.

Zachowuje się to w ten sposób:

Dotychczasowy Kod

    parser = argparse.ArgumentParser()

    # Flags
    flags = parser.add_argument_group("Flags")
    flags.add_argument(
        "-C", "--color", action="store_true",
        help="Color logs",
    )
    flags.add_argument(
        "-D", "--debug", action="store_true",
        help="Display debug information",
    )

    args = parser.parse_args()
    print(args)

Uruchomienie z opcją --help:

usage: crawler.py [-h] [-C] [-D]

optional arguments:
  -h, --help   show this help message and exit

Flags:
  -C, --color  Color logs
  -D, --debug  Display debug information

Uruchomienie z podaniem jednej flagi:

$ python crawler.py -D
Namespace(color=False, debug=True)

Podałem flagę -D, zatem opcja debug jest włączona.

Jeśli chodzi o konwencję nazewnictwa opcji - każdy program ma swoją. Ja osobiście staram się stosować wielkie litery do wyrażenia krótkich przełączników flag, a pozostałe opcje umieszczać w kolejności alfabetycznej. Nie ma tutaj jednak żadnej pisanej zasady.

Cache

Wiemy już jak dodawać opcje, jak argparse "mniej więcej" działa i jak wyświetlić pomoc programu.
Kontynuujemy zatem implementację i zajmujemy się opcjami dla Cache. Pamiętamy, że podzieliliśmy opcje na grupy argumentów - wczesniej używaliśmy flags.add_argument(), a teraz użyjemy cache.add_argument(). Cache w programie potrzebuje jedynie "wiedzieć" gdzie jest baza danych. Możemy wskazać ścieżkę pliku bazy danych sqlite3, a domyślnie chcemy, by baza danych była jedynie w pamięci (po zakończeniu programu baza "zniknie"). Dodajemy jedną opcję i ustawiamy wartość domyślną (default):

    # Cache
    cache = parser.add_argument_group("Cache")
    cache.add_argument(
        "-d", "--db", type=str,
        help="Database path (default: %s)" % DEFAULT_DB,
        default=DEFAULT_DB,
    )

Korzystam tutaj z faktu, że w programie stworzyliśmy już zmienne zawierające domyślne ustawienia. To te zmienne DEFAULT_*. Tutaj zatem używam wartości zmiennej DEFAULT_DB. Jeśli nie podamy opcji -d lub --debug, to domyślnie ścieżką bazy będzie :memory:, czyli wartość zmiennej DEFAULT_DB.
Używam też tej zmiennej dla wygenerowania podpowiedzi tej opcji w programie.

Scheduler

Scheduler przyjmuje 3 opcje:

  • max_idle_sec - maksymalny czas bezczynności (wartość typu float wyrażająca czas w sekundach)
  • queue_size - typu int, określa rozmiar kolejki, a tym samym limit jednorazowej "paczki" kolejkowanych linków (pobieranych z bazy)
  • schedule_interval_sec - typ float, interwał czasowy dla cyklicznego pobierania linków z bazy ("co tyle sekund")

Domyślne wartości już widzieliśmy (parametr default). A zatem kod wszystkich opcji schedulera:

    # Scheduler
    scheduler = parser.add_argument_group("Scheduler")
    scheduler.add_argument(
        "-i", "--max-idle-sec", type=float,
        help="Exit if idle for N seconds (default %f)" % DEFAULT_MAX_IDLE_SEC,
        default=DEFAULT_MAX_IDLE_SEC,
    )
    scheduler.add_argument(
        "-q", "--queue-size", type=int,
        help="URL Queue maxsize (default: %d)" % DEFAULT_QUEUE_SIZE,
        default=DEFAULT_QUEUE_SIZE,
    )
    scheduler.add_argument(
        "-t", "--schedule-interval-sec", type=float,
        help="Scheduling interval (default: %f)" % (
            DEFAULT_SCHEDULING_INTERVAL_SEC,
        ),
        default=DEFAULT_SCHEDULING_INTERVAL_SEC,
    )

Spostrzegawczy na pewno zauważą, że odnoszę się do zmiennych w notacji snake_case, a do parsera podaję je z myślnikami. Standardową konwencją dla opcji programów są myślniki, a argparse tworząc zmienne sam zamieni je na snake_case. Jest to konieczne ze względu na fakt, iż nie możemy mieć zmiennych z myślnikami. Myślnik to operator odejmowania.

Krótkie uruchomienie obrazujące to zachowanie na dotyczasowym kodzie:

$ ./crawler.py --queue-size 64 -t 3.14
Namespace(color=False, debug=False, db=':memory:', max_idle_sec=3, queue_size=64, schedule_interval_sec=3.14)

Downloader

Podstawową składnię mamy już opanowaną, podstawowe niuanse są już dla nas jasne, a zatem implementujemy opcje downloadera.

    # Downloader
    downloader = parser.add_argument_group("Downloader")
    downloader.add_argument(
        "-c", "--concurrency", type=int,
        help="Number of downloaders (default: %d)" % DEFAULT_CONCURRENCY,
        default=DEFAULT_CONCURRENCY,
    )
    downloader.add_argument(
        "-s", "--size-limit-kb", type=float,
        help="Download size limit (default: %f)" % DEFAULT_SIZE_LIMIT_KB,
        default=DEFAULT_SIZE_LIMIT_KB,
    )
    downloader.add_argument(
        "-u", "--user-agent", type=str,
        help="User-Agent header (default: %s)" % DEFAULT_USER_AGENT,
        default=DEFAULT_USER_AGENT,
    )
    downloader.add_argument(
        "-w", "--whitelist", action="append",
        metavar="DOMAIN",
        help="Allow http GET from this domain",
    )

Groupa opcji downloadera przyjmuje:

  • concurrency - liczba obiektów typu Downloader, określa tym samym maksymalną liczbę jednocześnie pobieranych adresów
  • size_limit_kb - ograniczenie rozmiaru pobieranych zasobów (jeśli nie chcemy pobierać wielkich plików)
  • user_agent - nazwa naszej "przeglądarki", która pojawi się w nagłówkach żądań. Jeśli chcemy przedstawiać się jako "Firefox Mobile", to nic nie stoi na przeszkodzie. Co wpiszemy, tak będzie. Serwery www logują zwyczajowo m.in. adres ip klienta (nasz), nazwę przeglądarki, żądany przez nas zasób, datę.

Opcja --whitelist ma akcję "append", co oznacza dodawanie do listy (list.append). Możemy podać zatem zero, jedną lub wiele domen przy uruchomieniu. W skrócie:

$ ./crawler.py
Namespace(user_agent='CodeASAP.pl: Crawler', whitelist=None)

$ ./crawler.py -w wikipedia.org -w 3blue1brown.com --whitelist dgmlive.com
./crawler.py -w wikipedia.org -w 3blue1brown.com --whitelist dgmlive.com
Namespace(user_agent='CodeASAP.pl: Crawler', whitelist=['wikipedia.org', '3blue1brown.com', 'dgmlive.com'])

Dla czytelności skróciłem nieco output.

Użyte w whitelist metavar powoduje wyświetlenie podanej wartości ("DOMAIN") w pomocy programu.

Uwaga: na tym etapie nie implementuję obsługi wildcard, czyli np. *.wikipedia.org. Jest to proste (urlparse, wyciągamy netloc, wycinamy ewentualny port, dzielimy po kropkach i dopasowujemy), niemniej w tej implementacji to pomijam. Jeśli chcemy whitelistować wiele subdomen, podajemy je osobno:
"-w pl.wikipedia.org -w en.wikipedia.org".

Argumenty pozycyjne

Zostały nam już tylko argumenty pozycyjne. Program ma opcjonalnie przyjmować startowe linki.
Dodaję zatem argument bez wskazywania opcji krótkiej i długiej, używam jedynie nazwy dla argumentu (tutaj url):

    # Positional
    parser.add_argument(
        "url", nargs="*", action="extend",
        help="Start url(s)",
    )

Akcją jest "extend", które rozszerza listę (list.extend), a parametr nargs o wartości gwiazdki wyraża "0 lub więcej" wystąpień. Parametr nargs może przyjmować konkretną liczbę oczekiwanych argumentów (np. nargs=3), lub je wymuszać (nargs="+" - co najmniej jeden argument). Nasz crawler może otrzymać linki, ale nie musi - mamy przecież bazę danych pełną linków oczekujących na zakolejkowanie, czym zajmuje się scheduler.

Funkcja: create_parser()

Całość tworzenia parsera zawieram w funkcji create_parser(), której treść niecierpliwi mogli już sprawdzić wcześniej tutaj.
Robię to w celu łatwiejszego testowania. Program nie ma zmiennych globalnych (oprócz wartość domyślnych DEFAULT_*), a zatem w testach po prostu zaimportuję funkcję create_parser() i będę tworzył obiekty parsera.

Testy parsera

Test parsera jest najłatwiejszą częścią testów. Parser nie ma żadnych zależności, importujemy funkcję create_parser(), tworzymy string jaki w rzeczywistości do programu trafia z shella, parsujemy argumenty i sprawdzamy czy wartości są takie, jakich oczekujemy.

Tworzę zatem klasę TestParser, która dziedziczy z unittest.TestCase, a następnie implementuję dwie metody testowe. Kod testu w całości będzie w pliku: tests/test_parser.py.

Z modułu crawlera importuję na wstępie funkcję create_parser oraz zmienne DEFAULT_*, których użyję do sprawdzenia wartości początkowych. Szkielet całości wygląda następująco:

from unittest import TestCase

from crawler import (
    create_parser,
    DEFAULT_USER_AGENT,
    DEFAULT_DB,
    DEFAULT_SIZE_LIMIT_KB,
    DEFAULT_CONCURRENCY,
    DEFAULT_QUEUE_SIZE,
    DEFAULT_MAX_IDLE_SEC,
    DEFAULT_SCHEDULING_INTERVAL_SEC,
)


class TestParser(TestCase):
    def test_defaults(self):
        ...

    def test_with_options(self):
        ...

Testy będę uruchamiał przy użyciu python -m unittest, a wybrane metody będę uruchamiał podając dodatkowy argument -k z nazwą funkcji. Komendy zamieszczam poniżej.

Uruchomienie testu w całości:

$ python -m unittest tests/test_parser.py

Uruchomienie testów pasujących do wzorca (tutaj funkcje zawierające w nazwie ciąg "defaults").

$ python -m unittest tests/test_parser.py -v -k defaults

Jesteśmy już gotowi do implementacji kodu metod samych testów.

Test: wartości domyślne

W pierwszej metodzie testuję wartości domyślne. Parser w tym przypadku nie otrzymuje żadnych argumentów.

...

class TestParser(TestCase):
    def test_defaults(self):
        parser = create_parser()
        args = parser.parse_args("")

        # Flags
        self.assertFalse(args.color, "Color off")
        self.assertFalse(args.debug, "Debug off")

        # Cache
        self.assertEqual(args.db, DEFAULT_DB)

        # SCheduler
        self.assertEqual(args.max_idle_sec, DEFAULT_MAX_IDLE_SEC, "Idle sec")
        self.assertEqual(args.queue_size, DEFAULT_QUEUE_SIZE, "Queue size")
        self.assertEqual(args.schedule_interval_sec,
                         DEFAULT_SCHEDULING_INTERVAL_SEC, "Interval")

        # Downloader
        self.assertEqual(args.concurrency, DEFAULT_CONCURRENCY)
        self.assertEqual(args.size_limit_kb, DEFAULT_SIZE_LIMIT_KB, "Limit")
        self.assertEqual(args.user_agent, DEFAULT_USER_AGENT, "User agent")
        self.assertIsNone(args.whitelist, "Whitelist is empty")

        # Positional
        self.assertEqual(args.url, [], "No default urls")

    def test_with_options(self):
        ...

Uruchamiam najpierw test_defaults przekazując wzorzec nazw funkcji, które mają być wykonane.
Podaję do unittest po prostu argument -k z wartością defaults. Ciąg ten występuje tylko w nazwie tej metody, wykonany zostanie zatem jedynie kod TestParser.test_defaults().

python -m unittest tests/test_parser.py -v -k defaults
test_defaults (tests.test_parser.TestParser) ... ok

----------------------------------------------------------------------
Ran 1 test in 0.001s

OK

Test: parsowanie argumentów

W drugiej metodzie przygotuję najpierw string cmdline odpowiadający temu, co zostałoby podane w konsoli przy uruchomieniu programu. Ten string przekażę do metody parsera ArgumentParser.parse_args() i w rezultacie parsowania tych opcji w args spodziewał będę się właściwych, przekazanych wartości.

Kod:

class TestParser(TestCase):
    ...

    def test_with_options(self):
        cmdline = " ".join([
            "-C -D -d test.db -i 3.14 -s 7.3 -c 13 -q 17 -t 21.7",
            "-w localhost -w dev.lan",
            "-u my-browser",
            "http://localhost https://dev.lan"
        ])

        parser = create_parser()
        args = parser.parse_args(cmdline.split())

        # Flags
        self.assertTrue(args.color, "Color on")
        self.assertTrue(args.debug, "Debug on")

        # Cache
        self.assertEqual(args.db, "test.db", "Database")

        # SCheduler
        self.assertEqual(args.max_idle_sec, 3.14, "Idle")
        self.assertEqual(args.queue_size, 17, "Queue size")
        self.assertEqual(args.schedule_interval_sec, 21.7, "Interval")

        # Downloader
        self.assertEqual(args.concurrency, 13, "Concurency")
        self.assertEqual(args.size_limit_kb, 7.3, "Size limit")
        self.assertEqual(args.user_agent, "my-browser", "User agent")
        self.assertEqual(args.whitelist, ["localhost", "dev.lan"], "Whitelist")

        # Positional
        self.assertEqual(args.url,
                         ["http://localhost", "https://dev.lan"], "Urls")

Uruchomienie testu

Testy uruchamiają się w mgnieniu oka.
Unittest:

python -m unittest tests/test_parser.py -v
test_defaults (tests.test_parser.TestParser) ... ok
test_with_options (tests.test_parser.TestParser) ... ok

----------------------------------------------------------------------
Ran 2 tests in 0.002s

OK

Podałem opcję "-v" dla modułu unittest, co powoduje czytelne wyświetlenie naz uruchamianych nazw metody i klas testów.

Testy pisane przy użyciu unittest możemy również uruchomić przy pomocy pytest.

pytest tests/test_parser.py --no-header -v
=========================== test session starts ========================================
collected 2 items

tests/test_parser.py::TestParser::test_defaults PASSED                           [ 50%]
tests/test_parser.py::TestParser::test_with_options PASSED                       [100%]

============================ 2 passed in 0.19s =========================================

Podsumowanie

Nie testuję tutaj wszystkich możliwości. Sprawdzam jedynie "czy działa", gdyż taki jest mój zakres wykorzystania argparse. Odniosę się do tego jeszcze przy okazji omawiania coverage (pokrycia kodu testami). Błędów można popełnić wiele. Sam poprawiłem kilka drobnych szczegółów pisząc ten artykuł. O ile domyślne wartości liczbowe (czas, rozmiar limitu w kilobajtach) początkowo w programie były wyrażone jako liczby całkowite, to po moich drobnych zmianach w kodzie nieco "rozjechały" się typy w parserze i podpowiedziach, co wyłapałem pisząc kod testów.

Istnieją moduły automatyzujące sam argparse, ale nie jest to aż na tyle kluczowa sprawa dla mnie, by dodawać duże ilości kodu lub zależności od innych modułów. Argparse spełnia wszelkie moje wymagania, jest też częścią biblioteki standardowej Pythona. Osobiście wolę napisać kilka linii więcej i wyłapać w trakcie drobnostki. Gdyby to był program umożliwiający składanie parserów klientom kodu (np. pluginom) - być może warto byłoby zadbać tutaj o strictness. Wykracza to jednak poza zakres crawlera w obecnej postaci.

Opcje programu mamy za sobą, wiemy też jak będziemy tworzyć i uruchamiać testy, a zatem w następnej części zajmiemy się kodem i testem klas Base i Worker.