Python Заметки

Релиз Python 3.10 случился! Все быстро побежали использовать новые type hints, pattern matching и всё такое😁 А между тем, на днях вышел Qt6.2. Наконец-то портировали такие модули как QtBluetooth, QtMultimedia, QtWebEngine, QtWebView и другие полезняхи. Если вы этого ждали, то пора действовать! PySide6 тоже подтянулся по версии. #qt#libs

В Python есть удобный режим, определяющий код с неверно закрытыми ресурсами. Этот режим называется Development Mode и включается двумя способами: Переменная окружения export PYTHONDEVMODE=1 python3 app.py Аргументы python3 -X dev app.py Если не закрыть файл должным образом, то вы получите в консоль ResourceWarning. Вот пример файла # app.py import psutil, os open('testfile', 'w') 123 print('process handlers:', psutil.Process(os.getpid()).open_files()) В этом примере я использую пакет psutil, чтобы убедиться, что перед выходом открытых файлов в моём процессе не осталось. При этом ResourceWarning всё равно будет выброшен, причём еще до использования psutil. app.py:3: ResourceWarning: unclosed file <_io.TextIOWrapper name='testfile' mode='w' encoding='UTF-8'> open('testfile', 'w') Object allocated at (most recent call last): File "app.py", lineno 3 open('testfile', 'w') process handlers: [] Для отображения строки с ошибкой требуется включить tracemalloc, тоже с помощью переменой или аргументов запуска. Смотрите примеры в доке. ➡️ Отсюда вывод: Всегда явно закрывайте файлы! Пишите чистый и предсказуемый код. #tricks#libs

Наверняка вы знаете что такое перегруженная функция в С++. Если нет, то всё просто. В С++ можно создать несколько функций с одинаковым названием но разными типами аргументов. И это будут разные функции. Во время вызова функции будет выбрана та её версия которая подходит по типам аргументов. Такая конструкция называется параметрический полиморфизм. Это удобно, когда мы точно не знаем какого типа прилетит аргумент и хотим обработать разные ситуации. Как мы это поведение можем повторить в Python? Обычно через проверку типов. def func(value): if isinstance(value, int): return func_int(value) elif isinstance(value, str): return func_str(value) else: raise NotImplementedError Не очень красиво 🧐 Начиная с версии 3.4 в Python добавили способ делать "перегруженные" функции более элегантно. Это декоратор singledispatch. Создадим нашу исходную функцию которая по умолчанию выбрасывает ошибку. @singledispatch def func(value) raise NotImplementedError Декоратор добавил для объекта func новую функцию register() с помощью которой можем регистрировать перегруженные функции. В качестве аргумента указывайте тип который данная функция обрабатывает @func.register(int) def func_int(x): print("INT:", x) Вместо указания типа в аргументах можно использовать аннотации аргумента функции @func.register def func_str(x: str): print("STR:", x) Если не указать тип одним из этих способов то получите ошибку TypeError. Имя новой функции не имеет значения, часто её называют просто "_" @func.register(list) def _(x): print("LIST", x) Если одна функция должна обработать несколько типов, то просто наслаиваем декоратор @func.register(float) @func.register(Decimal) def _(x): print(f'{type(x).__name__.upper()}:', x) Теперь у нас есть 5 отдельных функций которые вызываются в зависимости от типа передаваемого аргумента. При этом всё выглядит логично и компактно. Обработка каждого случая находится в своей отдельной функции! >>> func(1) INT: 1 >>> func('Python') STR: Python >>> func(1.2) FLOAT: 1.2 >>> func({}) NotImplementedError ------ ◽️ Данный способ работает только с первым аргументом. Все остальные аргументы будут переданы как есть и не участвуют в выборе нужной функции. ◽️ В версии 3.8 доступен декоратор singledispatchmethod с таким же функционалом но для методов класса. #tricks#libs

Какой тип данных выбрать для оптимального хранения информации? Имеется в виду объем занимаемой памяти. Можем создать 4 разных варианта объектов и сравнить сколько они занимают оперативки. Будем создавать простой класс, класс со слотами, именованный кортеж и словарь. from collections import namedtuple from sys import getsizeof NT = namedtuple('NT', 'v1 v2 v3') class CLASS: def init(self, x1, x2, x3): self.v1 = x1 self.v2 = x2 self.v3 = x3 class SLOTS: slots = ['x1', 'x2', 'x3'] def init(self, x1, x2, x3): self.x1 = x1 self.x2 = x2 self.x3 = x3 d = dict(x1=1, x2=2, x3=3) c = CLASS(1, 2, 3) s = SLOTS(1, 2, 3) t = NT(1, 2, 3) Теперь распечатаем что там по памяти print(' CLS\t\tSLT\t\tDCT\t\tTPL') print(f'System: {getsizeof(c)}\t\t' f'{getsizeof(s)}\t\t' f'{getsizeof(d)}\t\t' f'{getsizeof(t)}' ) CLS SLT DCT TPL System: 48 56 232 64 Хм, в этой статистике обычный класс самый экономный! Но что-то здесь не так. Неужели он экономичней класса со слотами, который рассчитан на скорость и оптимизацию? Дело в том, что функция sys.getsizeof() показывает не совсем то что мы ожидаем. Она берет результат метода __sizeof__ у объекта и добавляет кое-чего от gc. __sizeof__ возвращает размер, занимаемый данными. Но не учитывает размер обвязки этих данных. А еще он не гарантирует точность размера типов для third-party расширений. Для точного измерения размера лучше использовать модуль pympler. Помимо данных он считает сколько места занимает вся структура классов и другая обвязка объекта. from pympler import asizeof print(f'Pympler: {asizeof.asizeof(c)}\t\t' f'{asizeof.asizeof(s)}\t\t' f'{asizeof.asizeof(d)}\t\t' f'{asizeof.asizeof(t)}' ) CLS SLT DCT TPL Pympler: 416 152 496 160 И вот тут класс со слотами оказывается самым оптимальным решением! И это правильно. А словарь в этом тесте оказался самый расточительный. #tricks#libs

У Python есть очень удобная штука - стек вызовов. Благодаря нему модуль traceback может показать где и что сломалось в момент выброса исключений. Но иногда Python просто падает без какой-либо информации. Лично меня это регулярно настигает на Windows и очень бесит. Совершенно не понятно что где произошло и как искать причину. Ставить сишные дебаггеры? Чтобы быстро понять в какой строке ошибка иногда достаточно использовать встроенный модуль faulthandler Базовое использование очень простое: python.exe -q -X faulthandler main.py Если интерпретатор упадёт, то вы хотя бы узнаете какая строчка привела к такому событию. Попробуйте уронить Python без faulthandler и с ним. #libs#tricks

Модуль objgraph позволяет нарисовать граф объектов со связями между ними. По такому визуальному представлению иногда удобно дебажить архитектуру вашего кода. Особенно когда построение структуры происходит динамически. Не ставьте слишком большую глубину прорисовки, я как-то поставил 50 и ждал пол часа пока движок отрисует мне 3к нод. Вряд ли в таком графе можно что-то разобрать. К тому же рендер падает с ошибкой dot: graph is too large for cairo-renderer bitmaps. И просит уменьшить скейл до 0.2, а значит даже текста не разобрать. Так что глубина 3-5 вполне достаточно. #libs

Наверняка у многих возникло желание потестить отправку сообщений, но Redis не установлен и вообще непонятно как с ним быть. На самом деле запустить его очень просто. 🔸 Для Windows, качаем архив, запускаем redis-server.exe 🔸 Для Linux несколько команд 🔸 Если есть Docker, то еще проще docker run --rm -p 6379:6379 redis Всё, можно экспериментировать! ______________________ А еще с сервером можно поиграть в пинг-понг > redis-cli ping PONG #tricks#libs#linux

Самый большой минус синхронизации из прошлого поста - нестабильность. Не знаю как у вас а у меня эта штука падала несколько раз) По моему не очень production-ready. Что же делать? Отправлять disckcache на сетевой диск? Не, я бы не стал. Ведь есть отличная альтернатива! Это Redis. Redis это жутко быстрая in-memory база данных. Запись в неё похожа на документоориентированные NoSQL базы данных. То есть без схемы, без таблиц. Просто ключ=значение. Redis используется для кэширования и как брокер для передачи сообщений. Имеется подписка на изменения и время жизни записей. Вот пример кода: import redis R = redis.Redis() R.set('key', 'value') R.get('key') # b'value' Имея такой функционал, давайте реализуем что-то очень удобное для обмена данными по сети... Хотя подождите ка, всё уже придумано до нас! И это проект PyRSMQ Что он делает? Создаёт очередь сообщений которые может забирать другой клиент. Как это организовать? Для начала поднимаем cервер Redis. Потом на одном хосте создаём очередь для отправки. queue = RedisSMQ( host=host, port=port, qname='example' ) queue.exceptions(False).\ createQueue(delay=0).\ vt(message_live_time).\ execute() На другом хосте содаём клиента для прослушивания очереди on_receive = lambda msg: print(msg) consumer = RedisSMQConsumer( 'example', on_receive host=host, port=port) Начинаем прослушивание очереди consumer.run() Теперь можем отправлять сообщения queue.sendMessage(delay=0).message(msg).execute() Теперь все сообщения, отправленные в очередь, будут попадать в наш колбек on_receive . ▫️Обязательное условие — наличие поднятого Redis-сервера. ▫️Единственное ограничение, обусловленное спецификой инструментов, данные должны быть JSON serializable. #libs

От многопоточных вычислений переходим к распределённым. То есть вычисления, происходящие на нескольких компьютерах. Конечно, в зависимости от задачи, вы можете взять готовые решения вроде CGRU или Deadline для рендеринга, charm4py или Dask для ML, или замутить что-то на AWS С2. Но хотелось бы чего-то попроще, попитоничней что ли) А ведь в Python есть средства "из коробки" для синхронизации нескольких процессов на разных хостах. Вот простой пример кода, который синхронизирует работу двух процессов на разных компьютерах. В этом случае используется процесс-посредник, который является синхронизирующим сервером. В примере создаётся некий Manager, который шарит общую для клиентов очередь. Все подключившиеся могут что-то в неё писать или забирать. В моём коде один процесс что-то "считает" и складывает в очередь, другой забирает и продолжает какие-то свои "расчёты". Если у вас есть несколько машин, то можете попробовать это запустить по сети (нужно заменить 'localhost' на IP-адрес сервера). Но и на локальной машине сработает. Gist 🌎 #libs#source#tricks

Если вы писали когда-либо мультипоточные или мультипроцессорные приложения то вы знаете какая самая большая проблема у таких программ. Это конечно же синхронизация (мы сейчас не про GIL). Что такое синхронизация? Это когда параллельно работающий код может делать расчёты абсолютно независимо, но вот к общим данным они должны обращаться последовательно в определённом порядке. То есть нужна синхронность вместо асинхронности. Пока один процесс открыл файл, другие просто ждут своей очереди. Пока один поток обновляет переменную, другие просто спят, опять же ждут своей очереди. В Python есть стандартные средства для синхронизации. Это так называемые мьютексы с различной логикой. Например Lock, Semaphore или очередь Queue. Все они работают в контексте одного интерпретатора. То есть все потоки или процессы должны быть запущены из одной программы. Тогда можно использовать один Lock между потоками и общую память между процессами. Но что же делать, если процессы независимы? То есть я просто запускаю два разных интерпретатора с разным кодом. Возможно даже разных версий Python. В моём случае мне потребовалось писать некоторый кеш из одного модуля, который работает совершенно в разных приложениях как вспомогательный инструмент. Сначала я пытался что-то сделать на основе shelve, но потом нашел отличную библиотеку diskcache. Этот проект покрыл все мои потребности: ▫️ thread-safe и process-safe То есть можно выполнять команды из разных процессов и потоков и они всегда будут синхронизированы. Можно писать из разных процессов не опасаясь получить ошибку о том что файл занят другим процессом (как это бывает с shelve) ▫️всегда атомарные операции Это значит что любое действие выполняется в один запрос. ▫️безсерверный Не требуется отдельный процесс для синхронизации. Всё решается через базу данных. Полный список возможностей Из удобных фичей можно еще отметить встроенный Lock, позволяющий синхронизировать независимые процессы. Это как раз то что я искал! На что стоит обратить внимание: 🔸 несовместимы версии python 2 и 3 из-за разницы протоколов pickle 🔸 надёжность работы с сетевыми дисками под вопросом, я бы не стал. Но тут скорей вопрос к сети чем к софту. 🔸 при создании инстанса diskcache.Cache() все данные пишутся в рандомную директорию в temp. Чтобы синхронизировать разные процессы следует указывать одинаковый путь diskcache.Cache(some_path). #libs

🔖 Подводя итоги по прошлым постам плюс пара заметок: 🔸 Если требуется проверять идентичность содержимого архивов, лучше использовать ZIP и проверять только CRC. 🔸 Указанный способ может помочь проверить отдельные файлы в архиве с возможностью перекачать только новые а не весь архив 🔸 Для проверки хеш-суммы файла можно использовать утилиту md5sum (она не умеет проверять хеш внутри ахрхивов) # Linux: md5sum filename # Windows: python md5sum.py filename (находится в директории скриптов /Tools/scripts/md5sum.py) #libs#tricks

На самом деле архивы TAR оказались менее удобными в нашей теме проверки идентичности. Давайте сделаем всё тоже самое для ZIP. Допустим, тестовые файлы мы уже создали используя код из прошлого поста. Теперь создадим архивы. import zipfile def create_zip(archive_path, files): with zipfile.ZipFile(archive_path, "w") as zf: for file in files: zf.write(file) create_zip('archive1.zip', files_to_archive) create_zip('archive2.zip', files_to_archive) Проверим хеш >>> hashlib.md5(open("archive1.zip", "rb").read()).hexdigest() 'd54670be5e01e483797ee4ae30089423' >>> hashlib.md5(open("archive2.zip", "rb").read()).hexdigest() 'd54670be5e01e483797ee4ae30089423' Отлично! ZIP создаёт одинаковые архивы и сразу выдаёт одинаковую хеш-сумму! Ну всё, на этом расходимся... Хотя подождите ка, часто ли вы проверяете один и тот же файл на идентичность? Давайте имитируем ситуацию когда файл был перезаписан или "модифицирован" но при этом фактически не изменился. То есть изменились только его атрибуты. Для этого можно использовать Linux-команду touch, которая обновляет время последнего доступа к файлу. touch example_file0.txt touch example_file1.txt ... Либо альтернативу на Python from pathlib import Path for f in files_to_archive: Path(f).touch() Содержимое файлов не изменилось! Но изменились атрибуты. Пересоздаём второй архив. create_zip('archive2.zip', files_to_archive) Проверяем >>> hashlib.md5(open("archive1.zip", "rb").read()).hexdigest() 'd54670be5e01e483797ee4ae30089423' >>> hashlib.md5(open("archive2.zip", "rb").read()).hexdigest() 'aa508dbba4e223abe45e16dba4ad6e1f' Вот это более правдивая ситуация. Давайте теперь сделаем функцию для проверки файлов внутри архивов, которая считывает непосредственно данные файлов в разжатом виде. def get_hash_zip(path): hash_md5 = hashlib.md5() with zipfile.ZipFile(path, "r") as z: for f_name in z.namelist(): with z.open(f_name) as f: hash_md5.update(f.read()) return hash_md5.hexdigest() Сравним теперь хеш-суммы архивов >>> get_hash_zip('archive1.zip') '0b27c443737b0a84381b827e1d9a913b' >>> get_hash_zip('archive2.zip') '0b27c443737b0a84381b827e1d9a913b' Всё чётко сработало! А что по времени? >>> timeit.timeit("get_hash_zip('archive1.zip')", number=100, globals=globals()) 10.8 Ну тоже неплохо. ⭐️ А теперь самая главная фишка ZIP - при создании архива он СРАЗУ записывает контрольную сумму файла в заголовки! А это значит что мы можем просто считать готовые хеш-суммы и сравнить их! Это называется CRC (cyclic redundancy check) def get_hash_zip2(path): h = hashlib.md5() for info in zipfile.ZipFile(path).infolist(): h.update(info.CRC.to_bytes(8, byteorder='big')) return h.hexdigest() >>> timeit.timeit("get_hash_zip2('archive1.zip')", number=100, globals=globals()) 0.008 То есть даже буфер никакой не считывается, только несколько байт из заголовков каждого файла. Выполняется моментально. В моем случае 100 итераций за 8 мс! 🌐 Полный листинг тестов в Jupyter (для экспериментов жмём Open in Colab) 📌 И просто в Gists #libs#tricks