Коробка с питоном
541 subscribers
49 photos
128 links
Заметки от Python-разработчика: сниппеты, обзоры пакетов, новости и другая полезная информация.

Автор: @kiriharu
Download Telegram
С наступившим 2024 вас! Сегодня расскажу про dirty-equals.

Этот пакет используется для реализации более читаемых и декларативных проверок на равенство/соответствие, но в полной мере он себя раскрывает при написании тестов.
Как обычно, приведу несколько примеров:

from dirty_equals import *

>>> assert 1 == IsPositive # Всё ок, число положительное
>>> assert -2 == IsPositive # AssertionError
>>> # Проверки можно комбинировать при помощи булевой логики
>>> assert ['a', 'b', 'c'] == HasLen(3) & Contains('a') # Ок, список на 3 элемента, содержит 'a'
>>> # Есть проверка словарей
>>> assert {'a': 1, 'b': 2} == IsDict(a=1, b=2)
>>> # Словари можно проверять частично
>>> assert {'a': 1, 'b': 2, 'c': 3} == IsDict(a=1, b=2).settings(partial=True)
>>> # Или вот так
>>> assert {'a': 1, 'b': 2, 'c': 3} == IsPartialDict(a=1, b=2)
>>> Можно проверять соответствие регуляркам
>>> assert 'test@kiriha.ru' == IsStr(regex=r'^[^@]+@[^@]+\.[^@]+$')
>>> # И так далее...


Рекомендую заглянуть в документацию, там таких удобных штук ещё очень много! Ну и изучите исходники - написано не сложно, под капотом используется чуток метаклассов, для изучения темы - самое то!

#библиотека
Тренажер по тайпингу в Python для самых маленьких (и не только)

Python Type Challenges, очень прикольный сайт-тренажер для изучения типизации. Выбираешь тему, дописываешь необходимые куски кода и запускаешь проверку линтером. Почти как олимпиадные задачки, только полезные😁

Для некоторых задач доступны подсказки со ссылками на документацию.

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

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

Некоторые функции, модули, константы переместили или удалили, чтобы сделать неимспейсы более удобными и чистыми. Про это можно прочитать в NEP 52.
Релиз так же затронет юзеров с Виндой - теперь int в numpy 64х-разрядный (для 32х систем - 32х-разрядный), хотя ранее, по историческим причинам, связанных Python 2, был эквивалентен типу long в C.
Подписчиков набежало, а постов всё так же не было. Буду исправляться!

Сегодня расскажу про deptry - тулзу, которая позволяет находить неиспользованные, транзитивные или отсутствующие зависимости в проекте на Python.
Работает это следующим образом - запускаем утилиту в каталоге с проектом, она сканит все импортированные модули и сравнивает их с тем, что описано в зависимостях. Умеет работать с любым пакетным менеджером который строго следует PEP 621, а так же pyproject.toml от Poetry и PDM. Ну и обычный requirements.txt тоже поддерживается!

Где это может быть полезно? Для себя выявил несколько сценариев:
1. Получили старый проект без зависимостей и пытаемся собрать либы, которые он использует (много раз такое было на фрилансе).
2. Проводим глубокий рефакторинг, трекаем ненужные зависимости и таким образом избавляемся от них.
3. Если кто-то напихал нормальные зависимости в качестве dev зависимостей, утилита тоже это покажет.
4. Отлично детектит использование транзитивных зависимостей - это такие пакеты, которые мы не устанавливаем напрямую, но которые устанавливаются при установке другого пакета как его зависимость. Самый простой пример - установка sqlparse при установке Django. Такие зависимости должны быть указаны явно.
5. Добавляем в CI и время от времени смотрим отчеты, таким образом поддерживаем зависимости в нормальном виде.

Настраивается достаточно гибко, документация лежит здесь.

#утилита
Dishka - IoC-контейнер для Python

Когда мы следуем подходу Dependency Injection, а особенно - слоистой архитектуре, у нас образуется отдельная группа функций и классов, выполняющих только одну задачу - создание других объектов. Такой код лучше держать поближе к main, так как он связывает воедино разные части приложения и связан с конфигурацией запуска.

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

В Python меня долго не устраивали существующие контейнеры и я решил сделать свой:

Хочу представить вам Dishka 1.0

Цель этого проекта - предоставить простой и удобный IoC-контейнер, который сможет забрать всю работу с зависимостями. Мне кажется, на текущий момент это самый функциональный вариант контейнера, имеющий при этом самое простое API.

• Вы можете использовать его с любым фреймворком, но для некоторых мы уже подготовили хелперы
• Для создания зависимости можно указать отдельную функцию или использовать __init__ класса
• Зависимости имеют ограниченное время жизни (скоуп) и вы сами управляете им
• Зависимости кэшируются, поэтому один и тот же объект может быть переиспользован пока он жив. Так можно передать одно соединение с БД в несколько гейтвеев
• Фабрики зависимостей можно группировать в классы и компоненты, что позволяет делать контейнер модульным
• Можно декорировать объекты, использовать один объект для нескольких типов
• При старте проверяется корректность конфигурации контейнера, что позволяет исключить многие ошибки

Что значит версия 1.0?

У библиотеки было 9 промежуточных релизов, мы ради объявить, что закончена вся работа по стабилизации её интерфейса и исправлению ошибок. И у нас есть планы по развитию, уникальные фичи сами себя не напишут.

Будем рады новым пользователям, багрепортам, запросам фич и звездам на гитхабе

Github
Pypi
Документация
Там в Python 3.12 добавили нашумевший PEP 659, а у меня пет-проект один давно не обновлялся, и так уж звёзды сошлись, что я сижу второй день обновляю его на 3.12

Задача - есть функционал, который под капотом имеет некоторый класс следующего вида:

class BaseFunction:
serialize_to: None

def serialize(self, data: dict) -> serialize_to:
pass # тут мы используем наш serialize_to

@dataclass
class ModelA:
x: str

class FunctionA(BaseFunction):
serialize_to: ModelA


Мы определяем новые классы наследуясь от BaseFunction, переопределяем в них serialize_to и вызываем serialize который делает нам инстанс serialize_to.

Ну прямо дженерик напрашивается! Тем более в 3.12 их завезли, красивые:

class BaseFunction[T]:
def serialize(self, data: dict) -> T:
pass # тут мы используем наш serialize_to

class FunctionA(BaseFunction[ModelA]):
pass


Встаёт вопрос, а как нам получить наш тип из дженерика?

Для начала, получим __orig_bases__[0] - он вернёт нам классы, от которых мы наследовались. Так как нам нужен только наш первый класс, мы указываем [0]:

>>> FunctionA.__orig_bases__
__main__.BaseFunction[__main__.ModelA]


Ещё можно это сделать с помощью get_original_bases из types, но его добавили только в 3.12 (почему я об этом сказал - узнаете ниже).

Теперь надо получить получить сам тип в дженерике. В этом нам поможет typing.get_args, который получает все аргументы типа. Дополнительно укажем, что нам нужен первый тип:

>>> get_args(FunctionA.__orig_bases__[0])[0]
<class '__main__.ModelA'>


Теперь в методе serialize класса BaseFunction[T] можно написать штуку, которая автоматически сериализует наши данные:

def serialize(self, data: dict) -> T:
type_from_generic = get_args(self.__class__.__orig_bases__[0])[0]
return type_from_generic(**data)


Проверяем:

>>> f = FunctionA()
>>> f.serialize(data={"x": 1})
ModelA(x=1)

Вы восхитительны!

Кстати, эта же штука должна работать ещё вроде как аж с 3.8, так как в нём именно был добавлен __orig_bases__ (PEP 560), ну и под капотом у новых дженериков используется...

>>> FunctionA.__mro__
(<class '__main__.FunctionA'>, <class '__main__.BaseFunction'>, <class 'typing.Generic'>, <class 'object'>)


Ага, typing.Generic :)

#рецепт #std
Да, оно существует.

Под капотом низкоуровневая магия, из самого понятного - оно позволяет создавать поддельные объекты на куче. Можно создать байтовый массив с базовым адресом 0 и длинной SSIZE_MAX, что даст доступ к сырой памяти, на чтение и запись.

Такая техника, кстати, используется в эксплоитах движков javascript. Парочкой похожих эксплоитов уже давно хакают PS4.
Итак, ко мне пришли ребята из Podlodka Python Crew и предложили разыграть один бесплатный билет на новый сезон их конференции, где участников с 3 по 7 июня будет ждать целая неделя онлаин-воркшопов и много докладов по теме инфраструктуры.

Из интересного:
— К ним придёт разработчик FastStream, про который я неоднократно рассказывал здесь, покажет насколько удобно можно работать с Kafka, RabbitMQ, NATS и другими брокерами.
— Для енжоеров микросервисной архитектуры расскажут и покажут, как гарантировать консистентную работу всей системы при помощи паттерна "transactional outbox". Будет демо-стенд и реализация на практике.
— Приложение без метрик - плохое приложение. Покажут как собирать и анализировать данные, чтобы всегда быть в курсе состояния системы.
— Так же будут истории от экспертов про фейлы с инфраструктурой и как их можно избежать.

А так же нетворкинг, много практических примеров и дискуссии!

Чтобы попытать счастье, постите под этим постом любой мем про наш любимый язык до 02.06.24. Самый залайканный станет победителем - ему и отправится билет.
А если пытать счастье не хочется - вот вам скидка в 500 рублей по промокоду `podlodka_boxwithpython`

Ждем всех, тыкай чтобы попасть 😎
Сегодня расскажу про модуль, который поможет остановить время в ваших тестах и не только - FreezeGun

Сначала история из жизни.
Недавно мне принесли скрипт, который скрапил некоторую информацию с онлаин-каталога на текущий момент. Задачей было заставить его скрапить данные до определенного момента.
В нём было очень много вызовов datetime.now() (пагинация по датам, ага), перепиливать все это мне не очень хотелось, а результат хотели вчера.

Поэтому я такой хоп:
from freezegun import freeze_time

with freeze_time("01-08-24"):
result = call_api()

И всё работает!

Но вы так не делайте, конечно, это самый настоящий подпор костылём, мне вообще повезло, что оно заработало.
Или делайте если надо, разрешаю.

В основном этот пакет используют для тестов и как было понятно с примера выше - он позволяет "заморозить" время на определенном моменте.
Проверить истечение подписки, корректность фильтров, расчёт на определенное время - этот пакет позволит сделать всё это всего в пару строк кода!

Есть возможность использовать декоратор вместо контекстного менеджера:

@freeze_time("2012-01-14")
def test():
assert datetime.datetime.now() == datetime.datetime(2012, 1, 14)


А если оба варианта вам не нравятся, и хочется более гибкого управления - можно руками задавать когда время остановится и возобновится:

freezer = freeze_time("2012-01-14 12:00:01")
freezer.start()
assert datetime.datetime.now() == datetime.datetime(2012, 1, 14, 12, 0, 1)
freezer.stop()


С асинком, кстати, тоже работает. Библиотека довольно фичастая, так что за примерами использования можно проследовать в тесты.

#библиотека
В asyncio добавили возможность смотреть граф вызова корутин

Ждем в python3.14: https://github.com/python/cpython/commit/188598851d5cf475fa57b4ec21c0e88ce9316ff0

Пример:


import asyncio

async def test():
asyncio.print_call_graph()

async def main():
async with asyncio.TaskGroup() as g:
g.create_task(test(), name=test.__name__)

asyncio.run(main())


Выведет:


* Task(name='test', id=0x10304eee0)
+ Call stack:
| File '/Users/sobolev/Desktop/cpython2/Lib/asyncio/graph.py', line 278, in print_call_graph()
| File '/Users/sobolev/Desktop/cpython2/ex.py', line 4, in async test()
+ Awaited by:
* Task(name='Task-1', id=0x1034a1e60)
+ Call stack:
| File '/Users/sobolev/Desktop/cpython2/Lib/asyncio/taskgroups.py', line 121, in async TaskGroup._aexit()
| File '/Users/sobolev/Desktop/cpython2/Lib/asyncio/taskgroups.py', line 72, in async TaskGroup.__aexit__()
| File '/Users/sobolev/Desktop/cpython2/ex.py', line 7, in async main()


Как оно работает?

Появилось два новых важных изменений:

- Поле Frame.f_generator – оно хранит генератор или корутину, которая владеет данным фреймом. Нужно чтобы отрисовывать + Call stack:
- Новое свойство у Future


@property
def _asyncio_awaited_by(self):
if self.__asyncio_awaited_by is None:
return None
return frozenset(self.__asyncio_awaited_by)


Нужно, чтобы отрисовывать + Awaited by:.

Конечно же есть две иплементации. На питоне уже показал, вот так оно на C:


/*[clinic input]
@critical_section
@getter
_asyncio.Future._asyncio_awaited_by
[clinic start generated code]*/

static PyObject *
_asyncio_Future__asyncio_awaited_by_get_impl(FutureObj *self)
/*[clinic end generated code: output=... input=...]*/
{
/* Implementation of a Python getter. */
if (self->fut_awaited_by == NULL) {
Py_RETURN_NONE;
}
if (self->fut_awaited_by_is_set) {
/* Already a set, just wrap it into a frozen set and return. */
assert(PySet_CheckExact(self->fut_awaited_by));
return PyFrozenSet_New(self->fut_awaited_by);
}

PyObject *set = PyFrozenSet_New(NULL);
if (set == NULL) {
return NULL;
}
if (PySet_Add(set, self->fut_awaited_by)) {
Py_DECREF(set);
return NULL;
}
return set;
}


Как использовать?

Конечно же данная фича умеет не только печатать объекты в stdout. Прежде всего – она предоставляет удобное АПИ для различных IDE и дебагеров, которые смогут использовать данную информацию для визуализации: чего вообще у вас там происходит.

Ну и мониторинги, и sentry, и много кто еще получит дополнительную мета-информацию о процессе выполнения кода.

Документация: https://docs.python.org/3.14/library/asyncio-graph.html

Круто?

| Поддержать | YouTube | GitHub | Чат |
Коробка с питоном
Breaking: PEP 750 - Template Strings был принят: https://github.com/python/steering-council/issues/275#issuecomment-2794619570
Вроде крутой PEP, но напрашиваются вопросы:

1. Умрут ли всякие шаблонизаторы, по типу Jinja/Mako?
В самом PEP пишут что они необходимы в качестве использования средства кастомизации, но выглядит все это так, как будто надстройки над Template смогут их заменить.

2. В motivation написано, что якобы те, кто не осилили доку юзали f-строки для SQLек, что может привести к SQL-инъекциям, и мы щас темплейтами всё решим.
А может не надо ничего решать? Это ведь проблемы этих разработчиков.

3. Пикрилейтед.
Ничего проклятее я сегодня не увижу. Вот это реально не стоило добавлять.

Я конечно побухчу посижу, но всё равно буду пользоваться.
#pep
Rapidy - новый web-фреимворк на базе aiohttp

Недавно заметил, что в чатиках разгоняют про ещё один новый web-фреимворк, только уже построенный на aiohttp. Зачем, конечно, в 2025 году использовать aiohttp - вопрос сложный, но автор отвечает на него так - это для любителей aiohttp. Ну раз так, то ладно.
Вообще, когда я писал сервисы на aiohttp, мне всегда хотелось иметь работу с pydantic как у FastAPI. Можно сказать, автор воплотил одну из моих идей.

Что под капотом из того, что мне приглянулось?
- Валидация и сериализация как в хендлерах, так и в middleware при помощи полюбившегося всем Pydantic.
- Нативная интеграция с dishka.
- Декларативные контроллеры в виде классов с наследованием параметров - вкусовщина ещё та, но мне нравится:

@controller('/')
class UserController:
@get('/{user_id}')
async def get_by_id(self, user_id: str = PathParam()) -> dict[str, str]:
return {'user_id': user_id}

@get()
async def get_all_users(self) -> list[dict[str, str]]:
return [{'name': 'John'}, {'name': 'Felix'}]

- Ну и по стандарту, всё что уже и так есть в aiohttp.

А обещают нам:
- Генерацию OpenAPI - очень хотелось бы видеть.
- HTTP клиент обмазанный Pydantic - видимо чтобы быстренько клепать интеграции с http-сервисами.
- И, самое интересное - поддержку GRPC. Вот тут очень стало интересно, как это будет выглядеть.

Всё это, конечно, звучит круто, но у фреимворка есть самый большой недостаток, который вытекает из самого aiohttp - нету asgi. Мне кажется, что если его исправить, то ему можно будет даже подвинуться между легким FastAPI и уже ставшим монструозным Litestar.
Поэтому, пока что для меня - это aiohttp на стероидах. Удобный, легкий, слегка фичастый, подведенный к стандарту разработки 2025 года.

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

#библиотека
Помните про такую старинную питонячью забаву - гуй/фронтенд через Python рисовать?
Я даже пару постов делал об этом (тык, тык) - но все эти инструменты были "ну такие". Где-то фатальный недостаток в архитектуре, где-то кнопки мне не понравились.

В основном это нужно для людей, которые от фронтенда достаточно далеки - ML-щики, аналитики и датасотонисты, например. Я их понимаю, не осуждаю. Вот и ребята из Snowflake поняли, не осудили и придумали... Streamlit!

Что есть внутри:
- Простая генерация веб-интерфейса - на скриншоте можете увидеть, как просто делается интерфейс в принципе.
- Элементы - интерактивные! Чекбоксы, слайдеры, поля для ввода, таблички - здесь есть всё, что может понадобиться!
- Умеет отображать не только обычные списочки/словари, а еще датафреймы pandas, графики altair, bokeh, numpy и кучу еще всего!
- Очень простое API для создания собственных компонентов.
- Умеет в Markdown и Latex.

Поэтому, если хотите MVP прототип наклепать или написать очередной внутренний инструмент, продемонстрировать ML-модель - можно пользоваться!

А еще плейграунд есть - можно прям в браузере попробовать!
P.S. В комментах подсказали, что Streamlit сначала был отдельным проектом, а уже Snowflake его потом купили.

Github | Сайт | PyPi (39 тыщ звёзд!)

#библиотека
niquests - молодой преемник requests

Долгое-долгое время всеми нами любимая библиотека request для http запросов была де-факто стандартом.
Но время шло, в питоне появилось нормальное асинхронное программирование, технологии развивались, а requests никак не адаптировался к изменениям - у них там гордый feature freeze.
Когда у меинтейнеров спрашивали, как там насчёт асинка, видите ли, requests - это "feature-complete piece of software at this time", да и вообще есть весомые причины.

Вот так и жили. Сначала появился aiohttp. Потом httpx, который стал очень популярным и совместил несколько подходов, сразу в одном клиенте.
А потом появился наш парень niquests (первый релиз с 29 августа 2023 года), форк requests, который начал решать проблемы своего отца:

- Наконец появился асинк!
- Появилась поддержка HTTP/2 и HTTP/3 over QUIC (которой НЕТ даже в httpx/aiohttp!)
- Тредсейф, для любителей потоков
- Мультиплексирование соединения
- Добавили поддержку DNSSEC (этого нет ни в requests, ни в httpx ни в aiohttp).
- Есть DNS over HTTPS/QUIC/TLS - этого тоже нет в конкурентах
- Работа с вебсокетами (есть только в aiohttp и то только HTTP/1), и SSE (этого ни у кого нет).
- Ну и аннотации типов (с 3.0.0), хранение сертификатов в памяти и так далее - мы же не в прошлом десятилетии живём.

Да куча, блин, фич! Можете тут взглянуть на табличку, я не всё перечислил.

Вы думаете это всё? Он рвёт своих конкурентов по скорости - примерно в 3 раза быстрее в тестах - смотрите скриншот.

Самое, блин, грустное - я практически не вижу его в опенсурсных проектах. Но на бумаге выглядит очень круто.

Что думаете? Пользовались ли? Сможет ли он отнять часть рыночка? Пишите в комментарии, обсудим!

Github | Документация | PyPi
#библиотека
😎ovld - перегрузка функций по-человечески

👩‍💻 Наверняка любой из нас когда-то писал код по типу этого:
def process(x: str | int) -> None:
if isinstance(x, int):
do_for_int(x)
elif isinstance(x, str):
do_for_str(x)


🙈 А пишем мы так потому-что в Python из под коробки нет красивой и нормальной поддержки ad-hoc полиморфизма, аля перегрузки методов.
Ad-hoc полиморфизм - это возможность вызывать одну и ту же функцию с разными типами данных, при этом для каждого типа вызывается своя реализация. Например в Java мы можем написать что-то такое и это будет работать:
public int add(int a, int b) { return a + b; }
public double add(double a, double b) { return a + b; }
public String add(String a) { return a; }


🆘 А что делать, если хочется такое в Python? Есть несколько вариантов:
1. В стандартной библиотеке есть есть functools.singledispatch и functools.singledispatchmethod, но корректно они работают только с 1 аргументом 🤬:
from functools import singledispatch

@singledispatch
def process(x: Any):
print(f"Общий случай: {x}")

@process.register
def _(x: str):
print(f"Строка: {x}")

@process.register
def _(x: str, y: int): # сработает, но и сломает предыдущий обработчик
print(f"{x=}, {y=}")


2. Одно время часто советовали multipledispatch, но у него ряд недостатков - его не любят IDE, тайпинги не поддерживает, да и вообще всё еще стремно выглядит ⚰️:
from multipledispatch import dispatch

@dispatch(int, int) # никаких тайп хинтов, только вот так
def add(x, y):
return x + y

@dispatch(object, object)
def add(x, y):
return f"{x} + {y}"

add(1, 2) # 3
add(1, 'hello') #'1 + hello'


3. А еще можно использовать ovld 😎. Что он умеет?
👩‍💻 Перегружает по тайп-хинтам
👩‍💻 Полноценно работает с IDE и автодополнением (правда с pyright у меня иногда случаются нюансы)
😎 Умеет нормально работать с вложенными аннотациями (аля list[tuple[str, int])
😈 Не требует регистрации базовой функции (как в singledispatch)
💨 да и работает достаточно быстро (хотя куда без нюансов)?

А теперь небольшой примерчик:

from ovld import ovld

@ovld
def f(x: str):
return f"Строка {x!r}"

@ovld
def f(x: int):
return f"Число {x}"

@ovld
def f(x: int, y: int):
return "Джва числа!"


Ну красиво же? Я уже молчу при всякие фичи в виде кодгена, миксинов, метакласса который проставляет декоратор как бы за вас, типов зависимых по значению (есть проверка текста по регуляркам) и еще много всего!

GitHub | Документация | PyPi
#библиотека
Please open Telegram to view this post
VIEW IN TELEGRAM
💡 asyncssh - асинхронный ssh клиент и не только

👀 Если вы будете искать асинхронную альтернативу paramiko (это реализация SSHv2 на чистом питоне), то скорее всего наткнетесь на asyncssh - paramiko из под коробки не умеет в асинк.

Вообще, зачем он нужен, тем более для python? А давайте посмотрим, какие примеры есть у asyncssh:

➡️ Обычный клиент, который заходит на сервер и отправляет туда команду. Полезно, когда хочется сделать какой-то инструмент автоматизации, который заходит на кучу серверов и что-то на них выполняет (давайте представим что Ansible не существует, ладно?).
Еще у меня встречался кейс, когда есть какая-то штука, которая ставится на сервер, у которой нет API но есть CLI-утилита. Вот такие библиотеки здорово выручают.
➡️ Порт-форвардинг. Например, когда я писал диплом, у меня был скрипт который форвардил нужные для меня сервисы (ума не хватало впн поднять, да) на выданном мне хосте.
➡️ SFTP-клиент. Даже готовый класс есть, бери и накидывай поверх него какой-то гуй.
В последний раз, кстати, я такую штуку использовал для того, чтобы выкачивать бекапы. Был только доступен SFTP, поэтому мы выживали как могли...
➡️ А можно еще свой ssh сервер бахнуть. Например чтобы сделать простой honeypot для ботов, посмотреть что они будут делать на серваке.
➡️ Ну и закончу, пожалуй, sftp сервером. Можно написать небольшой сервер, c кастомной логикой.
Однажды я участвовал в одной CTF, в одной из тасок надо было за 5 команд получить доступ к флагу.
Количество попыток было ограничено. Угадаете, на чем было реализовано?

✏️ В общем, asyncssh это не просто "асинхроннй ssh-клиент, ну там чтобы командики отправить". Тут столько возможностей и идей для использования - бери и пили!

Документация | Github | PyPi
#библиотека
Please open Telegram to view this post
VIEW IN TELEGRAM
🔩 Если вам понадобится в рантайме поменить тип принимаемого объекта в роуте FastAPI, вот вам небольшой рецепт.

😆 Для одного проекта я решил сделать плагин-систему, которая автоматически регистрирует python-файлы и добавляет эндпоинты в API написанном на FastAPI. Так как я использую OpenAPI, очень хочется, чтобы принимаемые и возвращаемые типы отображались в нём корректно.

➡️Допустим, есть у меня вот такой код, создающий эндпоинты в зависимости от переданного Action:
class Action[ReqT, RespT](Protocol):
name: str
request_schema: Type[ReqT] # тип который мы будем принимать
response_schema: Type[RespT] # тип который мы будем возвращать
interact: Callable[[ReqT], Coroutine[Any, Any, RespT]]

def create_action_handler(action: Action):
# какой-то код
async def action_endpoint(entity_id: int, data: dict[str, Any]) -> Any:
# какой-то код
return await action.interact(server, data)
return action_endpoint


➡️ А вот так я добавляю нашу ручку в роутер:
def add_action_route(router: APIRouter, action: Action):
handler = create_action_handler(action)
router.post(
f"/plugins/{action.name}",
response_model=action.response_schema, # установим схему для ответа
)(handler)


Схему для ответа я установил при регистрации нашего эндпоинта в response_model. А что делать с data, тип которого должен быть не dict[str, Any] а action.request_schema

ℹ️ Для начала, я подумал, что раз FastAPI строит доку из аннотаций, наверное я смогу поменять аннотации и на этом дело сделано:
handler = create_action_handler(action)
handler.__annotations__["data"] = action.request_schema

Но это не помогло - аннотации поменялись, а вот схема - нет 🍺 !

😱 Оказывается, FastAPI не использует аннотации, а использует запеченные Signature, которые собираются из Dependency. При чем, в понимании FastAPI, dependency не только источник каких-то функций, а еще и наших схем. А дальше он из них и строит OpenAPI.

Зная это нам остается только одно - изменить signature перед регистрацией роута 😱:

sig = inspect.signature(handler) # достаем сигнатуру
new = []
for p in sig.parameters.values():
if p.name == "data": # находим нашу data
p = p.replace(annotation=action.request_schema) # меняем
new.append(p)
handler.__signature__ = sig.replace(parameters=new) # записываем новую сигнатуру


Вуаля! Теперь в OpenAPI будет указанная в action.request_schema схема 👏

#fastapi #рецепт
Please open Telegram to view this post
VIEW IN TELEGRAM