😎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
#библиотека

💡 asyncssh - асинхронный ssh клиент и не только

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

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

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

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

Документация | Github | PyPi
#библиотека

🔩 Если вам понадобится в рантайме поменить тип принимаемого объекта в роуте 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 #рецепт