Попался на глаза отличный перевод статьи "Writing Python like it's Rust", который скорее говорит о возможностях аннотаций типов и структур которые их используют (и еще немного про контекстные менеджеры).

Очень много примеров и очень много полезных советов :)

#статья #хабр

Если обычного itertools вам мало, то можно использовать more-itertools.

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

Например, вот так можно разделить список на 3 части:

data = ["first", "second", "third", "fourth", "fifth", "sixth", "seventh"]
[list(l) for l in divide(3, data)]
#  [['first', 'second', 'third'], ['fourth', 'fifth'], ['sixth', 'seventh']]

Вот еще задача. Надо разделить список с элементами по определенному условию. В этом нам поможет bucket:

class Cat:
    pass
class Dog:
    pass
shapes = [Cat(), Dog(), Cat(), Dog(), Cat(), Cat()]
result = more_itertools.bucket(shapes, key=lambda x: type(x))
len(list(result[Cat])) # 4
len(list(result[Dog])) # 2

По итогу кошки и собаки будут разделены по своему типу на 2 генератора.

Решим самую частую проблему - перевести список с несколькими уровнями вложенностями в "плоский" список:

iterable = [(1, 2), ([3, 4], [[5], [6]])]
list(more_itertools.collapse(iterable)) #[1, 2, 3, 4, 5, 6]

А если в плоский список нам нужно вытащить только элементы с первым уровнем вложенности?

list(more_itertools.collapse(iterable, levels=1)) # [1, 2, [3, 4], [[5], [6]]]

А вот так мы можем посмотреть, все ли элементы в коллекции уникальные:

more_itertools.all_unique([1, 2, 3, 4]) # True
more_itertools.all_unique([1, 2, 1, 4]) # False

Библиотека решает очень много типовых проблем, поэтому если научиться ей пользоваться - она сэкономит очень много времени. Возможно, я еще буду писать какие-то рецепты с ней, но это не точно 🌚...

PyPI | Документация

#more_itertools #itertools #библиотека #рецепт

Как раз сегодня искал фреимворк для организации работы консьюмера RabbitMQ и на глаза попался Propan - декларативный фреимворк для работы с очередями сообщений.

Для чего это нужно? На базе очередей можно построить асинхронную коммуникацию сервисов, а это привет микросервисной архитектуре!

Для сравнения, вот столько кода нам нужно написать, чтобы сделать консьюмер при помощи aio_pika:

import asyncio
import aio_pika
async def main():
    connection = await aio_pika.connect_robust(
        "amqp://guest:[email protected]/"
    )
    queue_name = "test_queue"
    async with connection:
        channel = await connection.channel()
        queue = await channel.declare_queue(queue_name)
        async with queue.iterator() as queue_iter:
            async for message in queue_iter:
                async with message.process():
                    print(message.body)
asyncio.run(main())

А вот столько с Propan:

from propan import PropanApp, RabbitBroker
broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
app = PropanApp(broker)
@broker.handle("test_queue")
async def base_handler(body):
    print(body)

Чистый кайф, не правда ли? Выглядит просто и понятно.

Что ещё умеет?
1) Кастить типы сообщений в модельки при помощи Pydantic.
2) Умеет работать с зависимостями (привет DI)
3) Имеет CLI утилитку, которая поможет сгенерировать проект, запустить несколько процессов воркеров, запустить хот-релоад для разработки.
4) А ещё есть огромное количество примеров, как им пользоваться.
5) Бонус - похоже у автора в планах прикрутить AsyncAPI (это как OpenAPI, только для очередей).

На данный момент стабильно работает с RabbitMQ, Redis и Nats. Kafka и SQS в бете, а NatsJs, MQTT, Redis Streams и Pulsar в планах.
Ну и накиньте звёзд автору, выглядит как то, что в будущем выстрелит :)

Github | Документация

#библиотека #propan

Я знаю на опыте, как иногда не удобно работать с датами и временем в Питоне.

В стандартной библиотеке есть как и большое количество типов данных (date, time, tzinfo, timedelta, relativedelta, ...) так и куча модулей (datetime, time, calendar, dateutil, ...) которые помогают работать с ними. Но содержать в голове, да еще и уметь работать с ними иногда проблематично.

Одна из библиотек, которая облегчает работу с сабжем - arrow. Давайте покажу на практике, насколько с ней удобно работать:

1) Получение времени
Начнём с легкого - получим время по MSK и UTC:

>>> arrow.now() # получаем текущее время по MSK
2023-06-02T18:46:55.188882+03:00
>>> arrow.utcnow() # а теперь текущее, но в UTC
2023-06-02T15:46:55.188882+00:00

А если нам надо текущее время с другой таимзоной?

>>> arrow.get().to("US/Pacific")
2023-06-02T08:49:39.794355-07:00

Это было просто. А что если нам нужно спарсить время из шаблона?

>>> arrow.get('2023/06-02|19:00:{00}', 'YYYY/MM-DD|HH:mm:{ss}') #обратите внимание на шаблон!
2023-06-02T19:00:00+00:00

Ещё arrow позволяет искать даты в строке по шаблону:

>>>arrow.get('Он родился 16 Мая 2020 года', 'DD MMMM YYYY', locale="ru")
2020-05-16T00:00:00+00:00

2) Работа со временем
Например, вот так можно заменить час в исходной дате:

>>>arrow.now().replace(hour=12)
2023-06-02T12:00:29.180755+03:00

Или добавить к текущей дате +5 недель:

>>>arrow.now().shift(weeks=+5)
2023-07-07T19:02:27.174819+03:00

Ну и в конце, давайте отформатируем текущее время в строку по шаблону:

>>>arrow.now().format('YYYY HH:mm:ss MM/DD')
2023 19:05:04 06/02

3) Очеловечивание времени

Возьмем задачу. Нам нужно сказать, через сколько мы прибудем домой. Делается это так:

>>> start_time = arrow.now()
>>> arrive_time = start_time.shift(hours=+3, minutes=10, seconds=12)
>>> humanized = arrive_time.humanize(locale="ru", granularity=["hour", "minute", "second"])
>>> print(f"Вы прибудете домой {humanized}")
Вы прибудете домой через 3 часа 10 минут 12 секунд

А если мы хотим из человеческого формата перевести в машинночитаемый?

>>> start_time.dehumanize(humanized, locale="ru")
2023-06-02T22:28:08.141525+03:00

4) Работа с диапазонами
Как раз недавно решал следующую задачку: дается временной промежуток, для него нужно сделать почасовые интервалы. С arrow всё это делается вот так:

start = datetime(2023, 1, 1, 00, 00)
end = datetime(2023, 1, 1, 23, 00)
for r in arrow.Arrow.span_range('hour', start, end):
    print(r)

(<Arrow [2023-01-01T00:00:00+00:00]>, <Arrow [2023-01-01T00:59:59.999999+00:00]>)
....
(<Arrow [2023-01-01T23:00:00+00:00]>, <Arrow [2023-01-01T23:59:59.999999+00:00]>)

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

GitHub | Документация

#библиотека

Если у вас есть желание понять как работает asyncio, threading или multiprocessing, либо же появились вопросы - рекомендую обратить внимание на superfastpython.com

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

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

Практически каждая статья сопровождается реальными примерами, которые можно самому запустить.
Все материалы на английском, но некоторые из них переводит на русский комьюнити (например цикл статей про asyncio).

#asyncio #threading #multiprocessing #статья

Создание временных файлов

В процессе написания скрипта может потребоваться создание временных файлов, которые будут удалены автоматически после завершения работы скрипта или обработки файла.
Это может быть полезно по разным причинам - при обработке больших данных (которые не вместятся в буфер) или при проведении сложных операций (например, можно создать временный файл и натравить на него ffmpeg).

Для решения этих проблем в Python есть модуль tempfile. Нас интересует 2 функции - это TemporaryFile и NamedTemporaryFile.

TemporaryFile позволяет создать безымянный временный файл. Вот так можно создать временный текстовой файл, открыть его на запись и чтение (за это отвечает первый аргумент "w+t", подробнее можно прочитать здесь):

from tempfile import TemporaryFile
with TemporaryFile("w+t") as t:
  t.write("Hello, boxwithpython!")
  t.seek(0)
  data = t.read()

NamedTemporaryFile используется для более продвинутых сценариев, так как он создает файл с именем, поэтому мы можем получить путь к нему и использовать его для дальнейших целей:

from tempfile import 

NamedTemporaryFile


with NamedTemporaryFile("w+t") as t:
  t.write("Hello, boxwithpython!")
  print(t.name) # /tmp/tmpljhsktjt

#std

Shielded execution в asyncio

Допустим, есть следующий обработчик, который производит оплату:

async def handler(request):
    await service.pay(request.user)
    return web.Response(text="payed")

Если соединение отвалится то обработчик упадет с ошибкой, так как серверу будет некуда отправлять ответ. Задача должна отмениться, но что если мы хотим, чтобы она выполнилась наверняка?

Поможет нам в этом asycio.shield(). Он защищает задачу от отмены, даже в случае возникновения ошибки. Выглядит это следующим образом:

async def handler(request):
    await asyncio.shield(service.pay(request.user))
    return web.Response(text="payed")

#asyncio #std

Про __slots__

Python, аналогично другим динамическим языкам, таким как JavaScript, предоставляет возможность манипулирования объектами в рантайме, в том числе позволяет добавлять, изменять и удалять атрибуты. Цена этого – понижение скорости доступа к атрибутам и дополнительные расходы памяти.

Такое поведение нужно не всегда. Бывают случаи, когда мы точно знаем, какие атрибуты будут у наших экземпляров классов. Или же мы хотим ограничить добавление новых атрибутов. Именно для этого и существует __slots__.

Слоты задаются через атрибут __slots__ в классе:

class SlotsClass:
    slots = ('foo', 'bar')

>>> obj = SlotsClass()
>>> obj.foo = 5
>>> obj.foo
# 5
>>> obj.another_attribute = 'test'
Traceback (most recent call last):
  File "python", line 5, in <module>
AttributeError: 'SlotsClass' object has no attribute 'another_attribute'

Теперь мы не можем добавлять новые атрибуты к нашим объектам. Скорость доступа к атрибутам повышается на 25-30%, потому что при доступе к ним их больше не надо вычислять.
В свою очередь, память экономится из-за того, что у класса не создается __dict__, который как раз хранил атрибуты.

#std #slots

__slots__ и наследование

Важно помнить, что при попытке унаследовать класс с __slots__ подкласс их унаследует, но так же и создаст __dict__ для новых атрибутов:

class SlotsClass:
    __slots__ = ('foo', 'bar')

class ChildSlotsClass(SlotsClass):
    pass

>>> obj = ChildSlotsClass()
>>> obj.__slots__
# ('foo', 'bar')
>>> obj.foo = 5
>>> obj.test = 3
>>> obj.__dict__
# {'test': 3}

Это стандартное и понятное поведение. Чтобы избежать создания __dict__, можно снова переопределить __slots__ в подклассе:

class SlotsClass:
    __slots__ = ('foo', 'bar')

class ChildSlotsClass(SlotsClass):
    __slots__ = ('baz',)

>>> obj = ChildSlotsClass()
>>> obj.foo = 5
>>> obj.baz = 6
>>> obj.something_new = 3

AttributeError: 'ChildSlotsClass' object has no attribute 'something_new'

А что с множественным наследованием?

class ClassA:
    __slots__ = ('foo', 'bar',)

class ClassB:
    __slots__ = ('baz',)

class C(ClassA, ClassB):
  pass

TypeError: multiple bases have instance lay-out conflict

Оно не работает. Потому-что каждый класс может иметь свои собственные __slots__, которые могут пересекаться с другими классами, а это может привести к тому, что объекты могут быть созданы неправильно или будут иметь непредсказуемое поведение.
Из-за этого возникает неоднозначность, какой именно слот использовать в результирующем классе.

#std #slots

Интересный доклад, в котором разработчик из Uchi.ru рассказывает, как они строили realtime аналитику. Внутри доклада Kafka, Redis, Postgres и, внимание, Django.

#посмотреть #django

А я к вам с новостями.

FastAPI в версии 0.100.0-beta1 поддерживает Pydantic v2 в бета-режиме. Да-да, это тот самый Pydantic, внутренности которого написаны на Rust. Гайд по миграции можно почитать здесь, а релиз тут.

#fastapi #pydantic

Зачем нужно делать кастомную базовую Pydantic модель?

Сейчас некоторые со мной не согласятся, но я часто рекомендую делать базовую pydantic модель и наследовать все модели от неё.

Наличие такой глобальной модели позволяет настраивать поведение всех моделей в приложении. Рассмотрю несколько кейсов, когда это может понадобится.

1) Контроль над входными данными.
Например, мы хотим округлять все поля которые называются price до трех знаков после запятой. Сделать это можно так:

class CustomBaseModel(BaseModel):
    @root_validator()
    def round_price(cls, data: dict) -> dict:
        prices = {k: round(v, 3) for k, v in data.items() if k == "price"}
        return {**data, **prices}

Валидаторы дают возможность изменять входящие данные, но это стоит использовать с осторожностью.

P.S.: В коментах подметили, что такой подход неявный, и я с этим согласен. Ничего не мешает для таких целей сделать ещё одну базовую модель (PriceRoundBaseModel), наследуясь от нашей базовой СustomBaseModel и использовать её там, где такое поведение необходимо.

2) Кастомый энкодер/декодер json.
Пакет json из стандартной библиотеки очень медленный. При необходимости ускорить сервис этот пакет в первую очередь пытаются заменить на что-то побыстрее.
Это происходит из-за того, что операций по (дe)сериализации json в приложении может быть много, и смена библиотеки, в этом случае, дает ощутимый прирост к общей скорости.

В pydantic есть 2 опции в конфиге, которые позволяют изменять поведение энкодера и декодера. Выглядит это так:

def orjson_dumps(v, *, default):
    # orjson.dumps возвращает байты, поэтому нам надо их декодить, чтобы соответствовать сигнатуре json.dumps
    return orjson.dumps(v, default=default).decode()

class CustomBaseModel(BaseModel):
    class Config:
        json_loads = orjson.loads
        json_dumps = orjson_dumps

#pydantic

Автор Propan на Habr'е написал статью "Messaging для чайников. Утилизируем все возможности RabbitMQ на Python", где объясняет основы работы с этим брокером сообщений при помощи своего фреймворка.

Для начинающих - самое то.

#habr #propan #статья

Как запускать синхронные функции в асинхронном роуте FastAPI?

Иногда так случается, что приходится использовать синхронный код в асинхронном роуте.
Если мы попытаемся вызвать синхронную функцию в асинхронном коде - наш event loop заблокируется и всё "зависнет" до тех пор, пока синхронный код не отработает.

Решений, как это сделать, на самом деле много. Самый простой вариант, который предоставляет FastAPI (а если быть точнее - Starlette, который использует anyio) - функция run_in_threadpool, которая запустит синхронный код в потоке:

@app.get("/")
async def my_router():
    result = await service.execute()
    client = SyncClient()
    return await run_in_threadpool(client.execute, data=result)

Кстати, BackgroundTask использует тот же самый способ, только он не возвращает результат выполнения.

А как бы вы решали/решаете такую проблему? Пишите в комментариях 😎 !

#fastapi #anyio

Релиз FastAPI 0.100.0🔥

Себастьян закончил обновление pydantic-а. Теперь можно насладиться звуком ржавчины и приростом скорости при переходе на новую версию FastAPI.

Чейнжлог совсем небольшой получился
https://fastapi.tiangolo.com/release-notes/#01000

p.s. в ближайшее время доберусь до чейнжлога pydantic v2 и сделаю обзорный пост

Yoyo

В текущих реалиях практически любой сервис использующий реляционную БД не может обойтись без миграций. Так уж сложилось, что в нашем питонячьем мире, в проектах почти всегда можно увидеть ORM (например SQLAlchemy) и пакет, который позволяет управлять миграциями, работающий с конкретной ORM (в случае с Алхимией это alembic).

Но иногда попадаются такие проекты, в которых нам не нужна ORM, по разным причинам. В этом случае, для миграций можно использовать yoyo.

Yoyo - это простой мигратор схем для нашей БД. Миграции здесь записываются в виде скриптов на SQL или на Python.

Сгенерировать папку для миграций и конфиг можно вот так:

yoyo init --database sqlite:///mydb.sqlite3 migrations

А теперь создадим миграцию:

yoyo new

Опишем SQL для применения миграции (apply) и для её отката (rollback):

from yoyo import step

__depends__ = {}

steps = [
    step(
        "CREATE TABLE TEST (a int);",
        "DROP TABLE TEST"
    )
]

Теперь можно применить миграцию командой yoyo apply и после откатить её командой yoyo rollback.

Внутри базы yoyo создает таблицу с логом всех операций (_yoyo_log), таблицу с применёнными миграциями (_yoyo_migration), а так же таблицу yoyo_lock, которая позволяет заблокировать миграции, которые будут идти с нескольких инстансов приложений.

Вот такая вот простая и удобная библиотека :)

Сайт | Sourcehut | PyPI

#библиотека

GIL скорее всего уберут из Python!

Руководящий совет объяснил статус PEP 703 (сделать GIL опциональным). Если кратко, то понятно следующее:

— Они намерены принять PEP 703, но всё ещё работают над деталями, предстоит много работы.
— Вероятно, лет через пять сборка без GIL будет ЕДИНСТВЕННОЙ сборкой, так как они не хотят разделять комьюнити на no-GIL и GIL.
— Совет не хочет повторения ситуации с Python 3, поэтому большое внимание будет уделено обратной совместимости.
— Прежде чем отказаться от GIL, они внимательно будут изучать работу комьюнити в этом направлении. Ребята хотят убедиться, что переход будет плавным, а поддержка режима no-GIL будет достаточна.
— Не смотря на все это, совет хочет иметь возможность "дать заднюю", если для комьюнити и языка это принесёт намного больше проблем, чем профита.

Новость просто невероятная. Остается надеяться на комьюнити, будет ли оно готово адаптировать свои библиотеки под этот режим?

#gil

Появился PEP 723, который предлагает "встраивать" pyproject.toml в однофайловые скрипты.
Предлагается добавить переменную __pyproject__, которая будет содержать в себе валидный TOML, описывающий метадату скрипта, в том числе как скрипт запускать и какие зависимости необходимы для запуска.

К примеру, вот так будет выглядеть скрипт, которому для работы нужна библиотека requests и питон 3.11 или выше:

__pyproject__ = """
[project]
requires-python = ">=3.11"
dependencies = [
  "requests<3",
]
"""
import requests
resp = requests.get("https://peps.python.org/api/peps.json")
print(resp.json())

PEP прикольный, что-то такое есть в качестве экспериментального RFC в Rust. Из минусов хотел бы отметить то, что автоматическая установка зависимостей может привести к запуску нежелательного кода. Но решение банальное - перед тем как что-то запускать, проверяйте, что вы запускаете.

#pep

Занимаюсь в рамках петпроекта обработкой текста, появилась потребность проверять на каком языке написан текст.

Сначала я попытался использовать langdetect, но он часто выдавал неправильные результаты. Как правило, плохие результаты выдавались по нескольким причинам:
1. Нет возможности ограничить языки, которые я хотел бы детектить. Мне надо определять всего четыре языка - украинский, русский, английский и немецкий.
2. Часто исследуемый мною текст слишком мал, из-за чего анализ ломался.

По итогу я пошел искать другую библиотеку и нашел lingua, которая успешно справляется с проблемами langdetect.
Важное отличие этой библиотеки от всех остальных в том, что она использует не только статистическую модель для определения языка, а ещё и механизмы, основанные на правилах - сначала определяется алфавит текста, ищутся символы которые уникальны для языка и после этого выбираются языки, на которых возможно написан текст.
Но есть возможность улучшить этот процесс. Можно самому ограничить языки, на которых возможно будет написан текст, а это как раз то, что мне и нужно:

from lingua import Language, LanguageDetectorBuilder
languages = [Language.ENGLISH, Language.RUSSIAN, Language.GERMAN, Language.UKRAINIAN]
detector = LanguageDetectorBuilder.from_languages(*languages).build()
print(detector.detect_language_of("Hello from box with python!")) # Language.ENGLISH
print(detector.detect_language_of("Привет от коробки с питоном!")) # Language.RUSSIAN

Из-за ограничения языков вероятность совершить ошибку на небольших предложениях сокращается многократно. В добавок ко всему, ребята используют не только триграммы, которые очень часто используют для таких задач, а n-граммы от 1 до 5, из-за чего вероятность предсказания повышается.
Причина такого решения проста - чем короче входной текст, тем меньше n-грамм доступно, а если мы будем проверять триграммами короткие заголовки - будут случаться ошибки.

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

#nlp #библиотека

Немного девопсовский пост, так как на предыдущем месте работы я много админил линукса.

Автоматизация - один из столпов работы любого админа, так как подготавливать сервера руками - занятие тяжелое и трудозатратное.
Для этого я всегда выбирал Ansible - это довольно простой инструмент, который позволяет настраивать сервера из специальных yaml конфигов, которые называются плейбуками.
Но какой бы вы инструмент не выбрали, рано или поздно конфигурация будет усложняться, а плейбуков/скриптов станет много. В какой-то момент всё это может сломаться, поэтому любую автоматизацию надо тестировать.

Когда я админил, команда не хотела полностью завязываться на Ansible (соответственно Molecule отпадает), решено было искать инструмент, который позволял бы тестировать состояние сервера после работы любой автоматизации.

И я нашел такой инструмент, имя ему - testinfra. Это плагин для pytest, который позволяет лаконично описывать тесты проверяющие состояние сервера.
Например, мы написали автоматизацию, которая устанавливает nginx. Проверим, что он установлен:

def test_nginx_is_installed(host):
    nginx = host.package("nginx")
    assert nginx.is_installed

А теперь проверим, что сервис с ним запущен:

def test_nginx_running_and_enabled(host):
    nginx = host.service("nginx")
    assert nginx.is_running
    assert nginx.is_enabled

Самое важное - для него существует много connection backend'ов, которые позволяют подключаться к серверам для теста как по SSH (через paramiko), так и используя инвентарь Ansible. Так же есть бекенды для docker/podman/salt/kubectl/openshift/winrm/lxc - обо всём этом можно почитать здесь.

Напоследок, вот вам еще кейс использования: одно время при помощи testinfra я тестировал правильность сборки контейнеров - все ли есть права, запустился ли сервис, в порядке ли конфиги и так далее.

#devops #библиотека

Тианголо (создатель FastAPI) предложил новый PEP 727, который позволяет стандартизировать механизм документирования параметров.

Причина проста - существует много псевдо-стандартов по форматированию параметров в docstring (например свои стандарты есть у numpy, гугла, много их!), но не все редакторы/IDE имеют возможность поддерживать эти "микроязыки". Он предлагает стандартизировать подход, используя существующий мощный инструмент в виде аннотаций типов.

Основное предложение - добавление в typing новой функции doc, которая принимает единственный параметр documentation. Ожидается, что эта функция будет использоваться вместе с Annotated.

А вот как это будет выглядеть:

def create_user(
    lastname: Annotated[str, doc("The **last name** of the newly created user")],
    firstname: Annotated[str | None, doc("The user's **first name**")] = None,
) -> Annotated[User, doc("The created user after saving in the database")]:
    """
    Create a new user in the system, it needs the database connection to be already
    initialized.
    """
    pass

Тианголо заранее позаботился и о старых версиях - он не стал выдумывать новый синтаксис, так же предлагается добавить doc() в пакет typing_extensions, который используют старые версии питона для исключения проблем с совместимостью.

Выглядит интересно, но важно понимать - автор пытается создать ещё один стандарт. Поддержит ли его комьюнити и уйдет ли PEP в работу? Остается только гадать.

#pep