Безусловно одной из самых популярных технологий доставки оповещений на устройства пользователей являются Push уведомления. Технология такова, что для её работы необходим постоянный доступ к интернету, а именно доступ к серверам, на которых регистрируются устройства пользователя для получения уведомлений. В данной статье мы рассмотрим весь спектр механизмов технологии WebPush уведомлений, спрятанных за словами WebSocket, ServiceWorker, vapid, register, broadcast, message encryption и т.д. Основной причиной побудившей меня к реверсу и изучению механизма, являлась необходимость доставки уведомлений мониторинга на рабочие места техподдержки, находящиеся в закрытом сегменте сети без доступа в интернет. И да, это возможно! Подробности под катом.

Disclaimer

В статье рассматривается режим доставки уведомлений пользователям в рамках использования браузера Mozilla Firefox. Это связано с тем, что на данный момент это единственный продукт позволяющий менять настройки push серверов используемых по умолчанию. Настройки браузеров Google Chrome, Chromium и производных в целях безопасности жёстко "зашиты" производителем в коде продукта.

Статья делится на две части

• Теоретическая информация
• Практические заметки для реализации механизма WebPush уведомлений

Используемые технологии и термины

WebSocket

Транспортным ядром системы Push уведомлений является протокол WebSocket, позволяющий в рамках стандартного HTTP/HTTPS подключения к Web серверу установить постоянный двусторонний канал связи между клиентом и сервером. В рамках установленного канала связи могут использоваться любые, в том числе бинарные, протоколы клиент-серверного взаимодействия заложенные разработчиками сервиса.

ServiceWorker

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

VAPID

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

WebPush

Механизм доставки сообщений до получателя.
Набор документов и спецификаций по WebPush

Workflow

Документации по WebPush довольно много (см. спойлер), но она существует только в парадигме
Client <-> Push Service <-> Application

Модель взаимодействия предполагает следующую схему.

Таким образом создаётся впечатление, что для работы механизма как минимум нужны специализированные сервисы регистрации и приёма/доставки уведомлений, а для отправки сообщений обязателен VAPID, спецзаголовки и ключи. Документация не описывает некоторых внутренних механизмов взаимодействия Push сервера и клиента, которые реально влияют на работу.
Попробуем рассмотреть все процессы подробно.

Фаза обработки сообщения

Меня долгое время интересовало, какие типы сообщений можно отправить браузеру и на что он среагирует.
Вопрос отпал когда я включил отладочный режим для push сообщений в браузере и полез в исходный код Firefox.

    try {      reply = JSON.parse(message);    } catch (e) {      console.warn("wsOnMessageAvailable: Invalid JSON", message, e);      return;    }    // If we receive a message, we know the connection succeeded. Reset the    // connection attempt and ping interval counters.    this._retryFailCount = 0;    let doNotHandle = false;    if (      message === "{}" ||      reply.messageType === undefined ||      reply.messageType === "ping" ||      typeof reply.messageType != "string"    ) {      console.debug("wsOnMessageAvailable: Pong received");      doNotHandle = true;    }    // Reset the ping timer.  Note: This path is executed at every step of the    // handshake, so this timer does not need to be set explicitly at startup.    this._startPingTimer();    // If it is a ping, do not handle the message.    if (doNotHandle) {      return;    }    // A whitelist of protocol handlers. Add to these if new messages are added    // in the protocol.    let handlers = [      "Hello",      "Register",      "Unregister",      "Notification",      "Broadcast",    ];    // Build up the handler name to call from messageType.    // e.g. messageType == "register" -> _handleRegisterReply.    let handlerName =      reply.messageType[0].toUpperCase() +      reply.messageType.slice(1).toLowerCase();    if (!handlers.includes(handlerName)) {      console.warn(        "wsOnMessageAvailable: No whitelisted handler",        handlerName,        "for message",        reply.messageType      );      return;    }    let handler = "_handle" + handlerName + "Reply";

Ни одно сообщение отправленное через websocket в сторону браузера не будет обработано, если оно не является системным сообщением проверки доступности конечной стороны "{}" или ответом на запрос от Push сервера. Это означает, что Push сервер не имеет никакого способа воздействия на работу клиентской стороны, кроме проверки её доступности. Аналогично, кроме 5 типов ответных сообщений, ничего обработано не будет.

Фаза инициализации

При запуске браузера Firefox, его внутренний механизм автоматически инициирует соединение с WebSocket(WS) сервером находящимся в системной настройке dom.push.serverURL с сообщением следующего формата.

{  "messageType": "hello",  "broadcasts":    {      "remote-settings/monitor_changes": "v923"    },    "use_webpush": True}

При первичной инициализации соединения(первый запуск браузера после установки/запуск нового профиля), поле "uaid" отсутствует, что является сигналом Push серверу о необходимости регистрации нового идентификатора. Как мы видим в разделе "broadcasts" присутствует некая пара "remote-settings/monitor_changes": "v923". Данная пара используется как буфер для хранения информации, отправляемой в сторону сервера при установлении соединения. В продукте Mozilla autopush, промышленной версии webpush сервера используемого на стороне серверов Mozilla, данная переменная используется как идентификатор последнего полученного пользователем сообщения из глобальной очереди сервера. Об изменении данного идентификатора мы поговорим позже. Итак, после принятия сообщения от клиента, сервер отвечает сообщением следующего вида

{  "messageType": "hello",  "status": 200,  "uaid": "b4ab795089784bbb978e6c894fe753c0",  "use_webpush": True}

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

Фаза регистрации

Под фазой регистрации подразумевается процесс, который предполагает готовность приёма событийной машиной браузера специально сформированных сообщений содержащих данные, транслируемые пользователю.
Фаза регистрации состоит из нескольких шагов:

• Проверка разрешения пользователя на получение информации
• Регистрация ServiceWorker
• Получение параметров подписки
• Формирование ключей шифрования для обслуживания подписки

Проверка разрешений пользователя на получение информации

На данном этапе браузер перед установкой ServiceWorker, запрашивает пользователя и системные настройки: "готов ли пользователь получать сообщения о подписке?"
В случае одного из отказов, установка ServiceWorker прерывается

Регистрация ServiceWorker

Как мы ранее оговаривали, ServiceWorker это автономная страница с обработчиками событий в отдельном пространстве браузера недоступном пользователю.
На работу с этим компонентом накладываются довольно серъёзные ограничения:

• Загрузка компонента ServiceWorker должна производиться через защищённое соединение(HTTPS), либо в целях отладки с localhost. Возможно включение флагов на "небезопасное" использование внешних ресурсов, но это не рекомендуется
• соединение WebSocket должно устанавливаться по защищённому соединению(WSS), либо в целях отладки по обычному WS соединению с localhost
• если в локальной сети имя сервера(ресурса) с которого происходит регистрация ServiceWorker, отличается от полного fqdn ресурса на котором находится ServiceWorker, то будет вызвано исключение о небезопасном вызове
  Жизненный цикл ServiceWorker от Google
  Жизненный цикл ServiceWorker от Mozilla

Процесс подписки

Процесс подписки подразумевает собой запуск механизма формирования ключей шифрования для создания сообщений и последующего расшифровки данных со стороны Push сервера.
Он состоит из следующих этапов:

• получение публичного ключа шифрования с Web сервера, являющегося промежуточным звеном между Push сервером и приложением осуществляющем отправку сообщений
• формирование браузером ключей шифрования для защиты сообщений
• вызов механизма подписки через канал WebSocket и получение точки для отправки сообщений

Получение публичного ключа

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

Запуск процесса генерации ключей шифрования

После получения публичного VAPID ключа вызывается процесс подписки ServiceWorker. Запущенный процесс подписки, используя в качестве идентификатора публичный VAPID ключ, формирует сессионный набор ключей шифрования (приватный ключ, публичный ключ, авторизационный ключ). Сессионный набор публичных ключей является экспортируемым и после окончания подписки может быть получен из пользовательской сессии.

Получение точки для отправки сообщений

После формирования ключей шифрования вызывается процесс внутри браузера называемый register. В сторону Push сервера через WebSocket браузер отправляет запрос вида

{  "channelID": "f9cb8f1c-05e0-403f-a09b-dd7864a03eb7",  "messageType": "register",  "key": "BO_C-Ou.......zKu2U4HZ9XeElUIdRfc6EBbRudAjq4="}

Это запрос на регистрацию соединения. В качестве параметров регистрации передаётся уникальный номер канала, формируемый каждый раз заново при старте процесса регистрации, а также публичный VAPID ключ сервера, через который будет производиться предварительная обработка и шифрование сообщений.
В ответ на запрос регистрации браузер ожидает сообщение вида

{  "messageType": "register",  "channelID": "f9cb8f1c-05e0-403f-a09b-dd7864a03eb7",  "status": 200,  "pushEndpoint": "https://webpush.example.net/wpush/f9cb8f1c-05e0-403f-a09b-dd7864a03eb7/",  "scope": "https://webpush.example.net/"}

В данном сообщении содержится адрес конечной точки сформированный Push(WebSocket) сервером, на которую необходимо отправить зашифрованное сообщение для получения его пользователем. Для передачи сообщения получателю, между WEB сервером, принимающим внешние запросы и WS сервером, отправляющим оповещения, должна быть организована логическая связь.

Итого по окончании процесса регистрации и подписки мы имеем следующий набор данных:

Браузер:

• приватный ключ шифрования сообщений
• публичный ключ шифрования сообщений
• ключ авторизации(DH)
• конечная точка для доставки сообщений получателю
• номер канала зарегистрированный на WebSocket сервере
• идентификатор клиента внутри WS соединения
• публичный ключ WebPush сервера

WebPush сервер:

• публичный ключ WebPush сервера
• приватный ключ WebPush сервера

Push(WebSocket) сервер:

• публичный ключ WebPush сервера
• адрес конечной точки клиента
• номер канала клиента привязанный к конечной точке
• идентификатор клиента внутри WS соединения

Из всего набора данных самым странным выглядит WebPush сервер. Я долго не мог понять каким образом происходит весь процесс доставки сообщения до пользователя, но после реверса всех механизмов, а также дебага autopush получилась следующая схема:

• некто хочет отправить сообщение в браузер пользователя
• для защиты сообщения необходимо извлечь из браузера настройки текущей подписки к Push серверу (конечную точку для отправки сообщения, публичный ключ шифрования сообщения, ключ авторизации)
• полученные настройки передаются на промежуточный WebPush сервер вместе с текстом сообщения
• промежуточный WebPush сервер формирует авторизационный JWT токен, содержащий время создания сообщения, адрес администратора WebPush сервера, время действия сообщения и подписывает его при помощи своего приватного ключа
• промежуточный WebPush сервер производит шифрование сообщения при помощи публичного ключа и ключа авторизации из браузера
• промежуточный WebPush сервер вызывает конечную точку полученную из браузера, передавая в неё связку JWT токен+публичный ключ для их проверки в заголовке Authorization, а также бинарный массив зашифрованного сообщения в теле запроса
• Push сервер по вызываемой конечной точке производит привязку запроса к каналу получателя
• Push сервер проверяет валидность JWT токена
• Push сервер конвертирует бинарный массив принятых данных в base64, формирует сообщение типа "notification" с каналом получателя, ставит сообщение в очередь, после чего механизм контроля очереди отправляет сообщение по WebSocket каналу в сторону клиента

Здесь мы прервём процесс для описания формата сообщения типа "notification".
Дело в том, что формат сообщения типа "notification" имеет два варианта. От того, что получил браузер и передал в ServiceWorker зависит логика работы по получению и отображению сообщения. Первый вариант, это "пустое" сообщение:

{  "messageType": "notification",  "channelID": "f7dfeed8-f868-47ca-a066-fbe629879fbf",  "version": "bf82eea1-69fd-4be0-b943-da96ff0041fb"}

"Пустое" сообщение как бы говорит браузеру "Эй, тебя тут ждут данные, приходи за ними". Браузер по логике работы должен сделать GET запрос на URL конечной точки и получить первую запись из очереди для отображения её пользователю. Схема конечно хорошая, только совсем небезопасная. В большинстве случаев она не применяется.
Вторым вариантом является передача данных совместно с сообщением.

{  "messageType": "notification",  "channelID": "f7dfeed8-f868-47ca-a066-fbe629879fbf",  "version": "bf82eea1-69fd-4be0-b943-da96ff0041fb",  "data": "I_j8p....eMlYK6jxE2-pHv-TRhqQ",  "headers":  {    "encoding": "aes128gcm"  }}

Браузер реагирует на поле headers в структуре сообщения типа "notification". При наличии этого поля автоматически включается механизм обработки зашифрованных данных из поля "data". На основании номера канала, событийная машина браузера выбирает набор ключей шифрования и пытается расшифровать полученные данные. После расшифровки расшифрованные данные передаются в обработчик "push" сообщений ServiceWorker. Как вы успели заметить, сообщение типа "notification" имеет поле "version", которое представляет собой уникальный номер сообщения. Уникальный номер сообщения используется в системе доставки и отображения сообщений для дедупликации данных.
Она работает следующим образом:

• любое принятое сообщение имеющее поле "version" заносится во внутренний реестр исключений
• корректно принятые и обработанные сообщения остаются в реестре исключений
• некорректно принятые и не обработанные сообщения из рееста исключений удаляются
  Информация о причинах такого поведения будет ниже.

Продолжим разбор процесса.

• Если сообщение принято и расшифровано, от браузера в сторону Push сервера формируется новое сообщение с типом "ack", включающее в себя номер канала и номер обработанного сообщения. Это является сигналом удаления сообщения из очереди сообщений для данного канала
• Если сообщение по какой-то причине не может быть обработано, от браузера в сторону Push сервера формируется новое сообщение с типом "noack", включающее в себя номер канала и номер отвергнутого сообщения. Это является сигналом постановки сообщения на повторную доставку через 60 секунд

Вернёмся к сообщениям с типом "broadcast". Продукт "autopush" от Mozilla использует их в качестве хранилища на стороне клиента, для определения последнего отправленного клиенту сообщения. Дело в том, что отправка сообщения типа "broadcast" со сменой значения ключа "remote-settings/monitor_changes", приводит к срабатыванию механизма, сохраняющего полученное значение в хранилище браузера. При потере соединения или каком-то программном сбое, сохранённое значение будет автоматически передано на сторону Push сервера в момент инициализации соединения и будет являться начальной точкой для последующей переотправки пропущенных сообщений из очереди.

Описывать сообщения типа "unregister" смысла не имеет, т.к. оно ни на что, кроме удаления сессии не влияет.

К чему же было приведено подробное описание всех процессов происходящих при Push оповещениях?
Смысл в том, что на основании этих данных можно довольно быстро построить свой Push сервер с необходимым функционалом. Продукт "autopush" от Mozilla является продуктом промышленного масштаба, который рассчитан на многомилионные подключения клиентов. В его составе присутствует TornadoDB, PyPy, CPython. К сожалению движок написан на Python 2.7, который массово выводится из эксплуатации.
Нам же нужен небольшой сервер с простым, желательно асинхронным кодом. А именно, без промежуточного WebPush сервера, VAPID, лишних межсерверных проверок и прочего. Сервер должен уметь привязывать клиентские подключения Push сервера к именам пользователей, а также иметь возможность организации эндпоинтов и webhook'ов для отправки сообщений этим пользователям.

Пишем свой сервер

У нас есть следующие данные:

• Пользователь с браузером Mozilla Firefox;
• Точка регистрации пользователя на сервере уведомлений для получения этих самых уведомлений;
• WebSocket сервер, обслуживающий подключения движка уведомлений, встроенного в браузер;
• Web сервер, формирующий интерфейс для пользователя и обслуживающий точки для отправки уведомлений;

Шаг 1
Первым делом мы должны подготовить WebSocket сервер, обслуживающий описанную ранее логику работы и подключения к нему клиентов.
В качестве фреймворка для реализации логики сервера используется AsyncIO Python.
Изначально стоит сразу разделить понятие "регистрация" для WebSocket движка браузера и понятие "регистрация" на сервере уведомлений. Разница заключается в том, что "регистрация" WebSocket движка браузера происходит автоматически без участия пользователя, в то время как разрешение на "регистрацию" на сервере уведомлений это осознанное действие со стороны пользователя.
Первичной задачей WebSocket сервера является принятие входящего соединения и его контроль на протяжении всего времени подключения браузера к серверу. Поэтому мы должны принять внешнее соединение, сделать его привязку к каналу и сохранить для дальнейшей работы.

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

# внешнее имя сервераSERVERNAME='webpush.example.net'# вебсокетыWS = set()# каналыCHANNELS = dict()async def register(websocket):    try:        WS.add(websocket)        websocket.handler = PushConnectionHandler(websocket)    except Exception as ex:        logger.error('Register exception: %s' % ex)async def unregister(websocket):    try:        CHANNELS.remove(websocket.handler.channel_id)        WS.remove(websocket)        logger.debug('UnregisterWebsocket[websocket]: %s'%websocket)    except Exception as ex:        logger.error('Unregister exception: %s' % ex)async def pushserver(websocket, path):    await register(websocket)    try:        await websocket.send(json.dumps({}))        async for message in websocket:            data = json.loads(message)            logger.info('Incoming message[data]: %s => %s '%(message, data))            if message == '{}':                await websocket.send(json.dumps({}))            elif 'messageType' in data:                logger.info('Processing WebSocket Data')                # Подключение к вебсокету из браузера                if data['messageType'] == 'hello':                    # Если это первичное подключение, то нужно задать идентификатор подключения и вернуть его браузеру                    if 'uaid' not in data:                        data['uaid'] = '%s' % uuid.uuid4()                    # Принудительно включить webpush                    if 'use_webpush' not in data:                        data['use_webpush'] = True                    helloreturn = {                        "messageType": "hello",                        "status": 200,                        "uaid": data['uaid'],                        "use_webpush": data['use_webpush']                        }                    websocket.handler.uaid = data['uaid']                    if 'broadcasts' in data:                        websocket.handler.register_broadcasts(data['broadcasts'])                    logger.debug('Hello websocket: %s' % vars(websocket.handler))                    CHANNELS.update({ data['uaid'] : websocket.handler })                    await websocket.send(json.dumps(helloreturn))                elif data['messageType'] == 'register':                    # Регистрация serviceWorker                    logger.debug('Register[data]: %s'%data)                    registerreturn = {                        "messageType": "register",                        "channelID": data['channelID'],                        "status": 200,                        "pushEndpoint": "https://%s/wpush/%s/" % (SERVERNAME,data['channelID']),                        "scope": "https://%s/" % SERVERNAME                    }                    websocket.handler.channel_id = data['channelID']                    if 'key' in data:                        websocket.handler.server_public_key = data['key']                    logger.debug('Register[registerreturn]: %s'%registerreturn)                    CHANNELS.update({ data['channelID'] : websocket.handler })                    await websocket.send(json.dumps(registerreturn))                elif data['messageType'] == 'unregister':                    unregisterreturn = {                        "messageType": "unregister",                        "channelID": data['channelID'],                        "status": 200                    }                    if data['channelID'] in CHANNELS:                        del CHANNELS[data['channelID']]                    logger.debug('Unregister[unregisterreturn]: %s'%unregisterreturn)                    logger.debug('Unregister[CHANNELS]: %s'%CHANNELS)                    await websocket.send(json.dumps(unregisterreturn))                elif data['messageType'] == 'ack':                    logger.debug('Ack: %s' % data)                    for update in data['updates']:                        if CHANNELS[update['channelID']].mqueue.count(update['version']) > 0:                            CHANNELS[update['channelID']].mqueue.remove(update['version'])                    logger.debug('Mqueue for channel %s is %s' % (websocket.handler.channel_id, websocket.handler.mqueue))                    await websocket.send('{}')                elif data['messageType'] == 'nack':                    await websocket.send('{}')            else:                logger.error("unsupported event: {}", data)    finally:        await unregister(websocket)

Как вы видите, никакой большой магии в WebSocket сервисе не заложено. Обрабатывается только список основных команд внутри WS сессии согласно спецификации.

Шаг 2
Следующим шагом нужно обеспечить регистрацию клиентской сессии на сервере оповещений.
Для этого необходимо задействовать механизм регистрации и установки ServiceWorker. В сети довольно много примеров, поэтому я взял готовый пример из сети и изменил его, добавив расширение логики.

'use strict';let isSubscribed = false;let swRegistration = null;var wait = ms => new Promise((r, j)=>setTimeout(r, ms));function urlB64ToUint8Array(base64String) {    const padding = '='.repeat((4 - base64String.length % 4) % 4);    const base64 = (base64String + padding)        .replace(/\-/g, '+')        .replace(/_/g, '/');    const rawData = window.atob(base64);    const outputArray = new Uint8Array(rawData.length);    for (let i = 0; i < rawData.length; ++i) {        outputArray[i] = rawData.charCodeAt(i);    }    return outputArray;}function subscribeUser() {    const applicationServerPublicKey = localStorage.getItem('applicationServerPublicKey');    const applicationServerKey = urlB64ToUint8Array(applicationServerPublicKey);    swRegistration.pushManager.subscribe({            userVisibleOnly: true,            applicationServerKey: applicationServerKey        })        .then(function(subscription) {            console.log('User is subscribed.', JSON.stringify(subscription));            localStorage.setItem('sub_token',JSON.stringify(subscription));            isSubscribed = true;            fetch(subscription.endpoint, {                method: 'POST',                cache: 'no-cache',                body: JSON.stringify(subscription)            })            .then(function(response) {                console.log('Push keys Update Response: ' + JSON.stringify(response));            })        })        .catch(function(err) {            console.log('Failed to subscribe the user: ', err);        });}function unsubscribeUser() {    swRegistration.pushManager.getSubscription()        .then(function(subscription) {            if (subscription) {                return subscription.unsubscribe();            }        })        .catch(function(error) {            console.log('Error unsubscribing', error);        })        .then(function() {            console.log('User is unsubscribed.');            isSubscribed = false;        });}function initializeUI() {    // Set the initial subscription value    swRegistration.pushManager.getSubscription()        .then(function(subscription) {            isSubscribed = !(subscription === null);            if (isSubscribed) {                console.log('User IS subscribed. Unsubscribing.');                subscription.unsubscribe();            } else {                console.log('User is NOT subscribed. Subscribing.');                subscribeUser();            }        });    (async () => {    await wait(2000);    console.warn('Wait for operation is ok');         swRegistration.pushManager.getSubscription()                .then(function(subscription) {                        isSubscribed = !(subscription === null);                        if (!isSubscribed) {                                console.log('ReSubscribe user');                                subscribeUser();                        }                })    })()}console.log(navigator);console.log(window);if ('serviceWorker' in navigator && 'PushManager' in window) {    console.log('Service Worker and Push is supported');    navigator.serviceWorker.register("/sw.js")        .then(function(swReg) {            console.log('Service Worker is registered', swReg);            swRegistration = swReg;            initializeUI();        })        .catch(function(error) {            console.error('Service Worker Error', error);        });} else {    console.warn('Push messaging application ServerPublicKey is not supported');}$(document).ready(function(){    $.ajax({        type:"GET",        url:'/subscription/',        success:function(response){            console.log("response",response);            localStorage.setItem('applicationServerPublicKey',response.public_key);        }    })});

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

'use strict';/* eslint-disable max-len *//* eslint-enable max-len */function urlB64ToUint8Array(base64String) {  const padding = '='.repeat((4 - base64String.length % 4) % 4);  const base64 = (base64String + padding)    .replace(/\-/g, '+')    .replace(/_/g, '/');  const rawData = window.atob(base64);  const outputArray = new Uint8Array(rawData.length);  for (let i = 0; i < rawData.length; ++i) {    outputArray[i] = rawData.charCodeAt(i);  }  return outputArray;}function getEndpoint() {  return self.registration.pushManager.getSubscription()  .then(function(subscription) {    if (subscription) {      return subscription.endpoint;    }    throw new Error('User not subscribed');  });}self.popNotification = function(title, body, tag, icon, url) {  console.debug('Popup data:', tag, body, title, icon, url);  self.registration.showNotification(title, {      body: body,      tag: tag,      icon: icon    });  self.onnotificationclick = function(event){      console.debug('On notification click: ', event.notification.tag);      event.notification.close();      event.waitUntil(        clients.openWindow(url)      );  };}var wait = ms => new Promise((r, j)=>setTimeout(r, ms));self.addEventListener('push', function(event) {   console.log('[Push]', event);  if (event.data) {    var data = event.data.json();    var evtag = data.tag || 'notag';    self.popNotification(data.title || 'Default title', data.body || 'Body is not present', evtag, data.icon || '/static/images/default.svg', data.url || '/getevent?tag='+evtag);  }  else {    event.waitUntil(      getEndpoint().then(function(endpoint) {        return fetch(endpoint);      }).then(function(response) {          return response.json();      }).then(function(payload) {          console.debug('Payload',JSON.stringify(payload), payload.length);          var evtag = payload.tag || 'notag';          self.popNotification(payload.title || 'Default title', payload.body || 'Body is not present', payload.tag || 'notag', payload.icon || '/static/images/default.svg', payload.url || '/getevent?tag='+evtag);      })    );  }});self.addEventListener('pushsubscriptionchange', function(event) {  console.log('[Service Worker]: \'pushsubscriptionchange\' event fired.');  const applicationServerPublicKey = localStorage.getItem('applicationServerPublicKey');  const applicationServerKey = urlB64ToUint8Array(applicationServerPublicKey);  event.waitUntil(    self.registration.pushManager.subscribe({      userVisibleOnly: true,      applicationServerKey: applicationServerKey    })    .then(function(newSubscription) {      // TODO: Send to application server      console.log('[Service Worker] New subscription: ', newSubscription);    })  );});

Согласно представленного кода, Javascript файл main.js инициирует при своём запуске получение публичного VAPID ключа и принудительно вызывает подписку браузера на оповещения.
Для простоты отладки WebSocket сервер во время регистрации подписки отдаёт URL вида: https://webpush.example.net/wpush/ChannelGuid.
Откуда же берётся имя пользователя в сервере уведомлений. Вся суть в том, что инициирование подписки /subscription/ происходит полуавтоматически. Соответственно в зависимости от того, что вы хотите увидеть в качестве идентификатора пользователя, вы можете передать после оформления подписки в момент передачи ключей.
Это происходит путём вызова метода POST по адресу WebPush endpoint присланного сервером из модуля ServiceWorker.

function subscribeUser() {    const applicationServerPublicKey = localStorage.getItem('applicationServerPublicKey');    const applicationServerKey = urlB64ToUint8Array(applicationServerPublicKey);    swRegistration.pushManager.subscribe({            userVisibleOnly: true,            applicationServerKey: applicationServerKey        })        .then(function(subscription) {            console.log('User is subscribed.', JSON.stringify(subscription));            localStorage.setItem('sub_token',JSON.stringify(subscription));            isSubscribed = true;            fetch(subscription.endpoint, {                method: 'POST',                cache: 'no-cache',                body: JSON.stringify(subscription)            })            .then(function(response) {                console.log('Push keys Update Response: ' + JSON.stringify(response));            })        })        .catch(function(err) {            console.log('Failed to subscribe the user: ', err);        });}

Как было написано ранее, в сервере используется обработчик точек подключения. Это отдельная часть кода в скрипте сервера, но обрабатывающая вместо WebSocket, клиентский WEB трафик от браузера.
В качестве обрабатываемого заголовка, содержащего идентификатор пользователя, в базовом варианте сервиса использовался basiclogin полученный при авторизации пользователя в LDAP.

        location ~ /subscription|/pushdata|/getdata|/wpush|/notify {            proxy_pass http://localhost:8090;            proxy_set_header LDAP-AuthUser $remote_user;            proxy_set_header 'X-Remote-Addr' $remote_addr;            add_header "Access-Control-Allow-Origin" "*";            add_header Last-Modified $date_gmt;        proxy_hide_header "Authorization";            add_header Cache-Control 'no-store, no-cache, must-revalidate, proxy-revalidate, max-age=0';            if_modified_since off;            expires off;            etag off;        }

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

USERIDHEADERNAME='X-Remote-Addr'async def update_channel_keys(request, data):    channel = request.path.replace('wpush','').replace('/','')    logger.debug('update channel keys data: %s'%data)    logger.debug('Update Channel keys Headers: %s' % request.headers)    if USERIDHEADERNAME not in set(request.headers):        return False    basiclogin = request.headers[USERIDHEADERNAME]    logger.debug('Login %s' % basiclogin)    if basiclogin not in LOGINS_IN_CHANNELS:        LOGINS_IN_CHANNELS.update({ '%s'%basiclogin : {} })    LOGINS_IN_CHANNELS['%s'%basiclogin].update({'%s' % channel : {} })    logger.debug('LOGINS_IN_CHANNELS: %s' % LOGINS_IN_CHANNELS)    try:        jdata = json.loads(data)        if 'endpoint' in jdata and 'keys' in jdata:            logger.debug('Saving Keys for Channel: %s => %s' % (channel, jdata))            CHANNELS[channel].register_keys(jdata['keys'])            logger.debug('Registered channel keys %s:' % vars(CHANNELS[channel]))        return True    except Exception as ex:        logger.error('Exception %s'%ex)        return False

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

Шаг 3
Сессия зарегистрирована, ключи на сервер переданы, пора отправлять и получать сообщения.
Как я описывал в самом начале статьи, у сервиса оповещений существует два способа доставки сообщений:

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

При корректном формировании сообщения необходимо передать поле tag содержащее уникальный идентификатор сообщения. Если сервер имеет сессионные ключи для клиента в сторону которого передаётся уведомление, то он может зашифровать это уведомление. Если нет, то сервер отправит пустое уведомление и клиент сам придёт за этим уведомлением по полю tag. Логику получения браузером сообщения реализует следущий блок кода в ServiceWorker:

self.addEventListener('push', function(event) {   console.log('[Push]', event);  if (event.data) {    var data = event.data.json();    var evtag = data.tag || 'notag';    self.popNotification(data.title || 'Default title', data.body || 'Body is not present', evtag, data.icon || '/static/images/default.svg', data.url || '/getevent?tag='+evtag);  }  else {    event.waitUntil(      getEndpoint().then(function(endpoint) {        return fetch(endpoint);      }).then(function(response) {          return response.json();      }).then(function(payload) {          console.debug('Payload',JSON.stringify(payload), payload.length);          var evtag = payload.tag || 'notag';          self.popNotification(payload.title || 'Default title', payload.body || 'Body is not present', payload.tag || 'notag', payload.icon || '/static/images/default.svg', payload.url || '/getevent?tag='+evtag);      })    );  }});

В случае отсутствия ключей шифрования для уведомления, необходимо организовать очередь сообщений, через которую ServiceWorker сможет получить все ожидающие его сообщения. Так как мы строим простую реализацию сервера, всё что нам нужно это сохранить сообщения до определённого момента, а потом их удалить. Т.к. сообщения мониторинга ждущие получения больше 10 минут, в большинстве случаев уже не актуальны. При этом можно не хранить в очереди "пустые" сообщения для индикации наличия уведомлений на сервере.

Блок шифрования сообщений передаваемых был взят из кода сервера "autopush", дабы не нарушать совместимости.

    def encrypt_message(self, data, content_encoding="aes128gcm"):        """Encrypt the data.        :param data: A serialized block of byte data (String, JSON, bit array,            etc.) Make sure that whatever you send, your client knows how            to understand it.        :type data: str        :param content_encoding: The content_encoding type to use to encrypt            the data. Defaults to RFC8188 "aes128gcm". The previous draft-01 is            "aesgcm", however this format is now deprecated.        :type content_encoding: enum("aesgcm", "aes128gcm")        """        # Salt is a random 16 byte array.        if not data:            logger.error("PushEncryptMessage: No data found...")            return        if not self.auth_key or not self.receiver_key:            raise WebPushException("No keys specified in subscription info")        logger.debug("PushEncryptMessage: Encoding data...")        salt = None        if content_encoding not in self.valid_encodings:            raise WebPushException("Invalid content encoding specified. "                                   "Select from " +                                   json.dumps(self.valid_encodings))        if content_encoding == "aesgcm":            logger.debug("PushEncryptMessage: Generating salt for aesgcm...")            salt = os.urandom(16)        # The server key is an ephemeral ECDH key used only for this        # transaction        server_key = ec.generate_private_key(ec.SECP256R1, default_backend())        crypto_key = server_key.public_key().public_bytes(            encoding=serialization.Encoding.X962,            format=serialization.PublicFormat.UncompressedPoint        )        if isinstance(data, str):            data = bytes(data.encode('utf8'))        if content_encoding == "aes128gcm":            logger.debug("Encrypting to aes128gcm...")            encrypted = http_ece.encrypt(                data,                salt=salt,                private_key=server_key,                dh=self.receiver_key,                auth_secret=self.auth_key,                version=content_encoding)            reply = CaseInsensitiveDict({                'data': base64.urlsafe_b64encode(encrypted).decode()            })        else:            logger.debug("Encrypting to aesgcm...")            crypto_key = base64.urlsafe_b64encode(crypto_key).strip(b'=')            encrypted = http_ece.encrypt(                data,                salt=salt,                private_key=server_key,                keyid=crypto_key.decode(),                dh=self.receiver_key,                auth_secret=self.auth_key,                version=content_encoding)            reply = CaseInsensitiveDict({                'crypto_key': crypto_key,                'data': base64.urlsafe_b64encode(encrypted).decode()            })            if salt:                reply['salt'] = base64.urlsafe_b64encode(salt).strip(b'=')        reply['headers'] = { 'encoding': content_encoding }        return reply

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

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

WebPush AsyncIO server

Для развёртывания сервера необходимо:

• Установить необходимые Python модули, а также настроить nginx по примеру приложенного конфигурационного файла.
• Поместить содержимое директории web в корень ранее настроенного виртуального сервера
• Перезапустить/перечитать конфиг nginx
• В браузере через about:config поменять параметр dom.push.serverURL на адрес wss://ваш.сервер/ws
• Перед сменой адреса push сервера можно очистить поле dom.push.userAgentID, которое автоматически заполнится если ваш Push сервер работает корректно и принимает соединения.
• Для тестирования оповещений необходимо зайти на страницу https://ваш.сервер/indexpush.html и открыв окно отладки удостовериться в корректной регистрации ServiceWorker
• Нажать кнопку "Check Push Notify"
• Если всё правильно настроено, появится всплывающее сообщение

Как говорилось в начале статьи, система проектировалась для оперативного оповещения службы техподдержки. В зависимости от требуемой логики поведения, в Zabbix можно использовать следующий обработчик webhook

var req = new CurlHttpRequest();req.AddHeader('Content-Type: application/x-www-form-urlencoded');var jv = JSON.parse(value);if (jv.recovery_nstatus == '{EVENT.RECOVERY.VALUE}') {jv.icon = '/static/images/problem/' + jv.event_severity + '.svg';}else{jv.icon = '/static/images/recovery/' + jv.event_severity + '.svg';}value = JSON.stringify(jv);Zabbix.Log(2, 'webhook request value='+value);req.Post('https://webpush.server.net/pushdata/',  value);Zabbix.Log(2, 'response code: '+req.Status());return JSON.stringify({  'tags': {    'endpoint': 'webpush'  }});

c параметрами

Если добавить красивых картинок из FontAwesome, то получится вот так

WebPush сервер поддерживает следующие вызовы :

• POST https://webpush.example.net/wpush/channelId сохранение ключей шифрования и имени пользователя
• GET https://webpush.example.net/wpush/channelId получение тестового сообщения
• GET https://webpush.example.net/subscription получение публичного VAPID ключа
• POST https://webpush.example.net/pushdata отправка JSON структуры передаваемой в качестве сообщения в браузер
  

  {        "url": "http://personeltest.ru/away/habr.com/", // URL на который необходимо перейти при клике        "recipient": login, // Логин или идентификатор пользователя        "title": "Заголовок сообщения",        "body": "Тело сообщения",         "icon": "/static/images/new-notification.png", // путь к иконке сообщения        "version": uuid, // идентификатор сообщения        "tag": uuid, // тег сообщения для получения        "mtime": parseInt(new Date().getTime()/1000) //Время }
  

• GET https://webpush.example.net/getdata Получение очереди сообщений
• POST https://webpush.example.net/notify/login Отправка пустого оповещения пользователю
• POST https://webpush.example.net/notifychannel/channelId Отправка пустого оповещения в канал

Вот в принципе и всё. Надеюсь у вас снялась часть вопросов с тем, как работает WebPush. Спасибо за потраченное время на чтение материала.

Aborche 2020

Введение

Как Python умудряется делать несколько вещей одновременно?

from multiprocessing import Processdef print_func(continent='Asia'):    print('The name of continent is : ', continent)if __name__ == "__main__":  # confirms that the code is under main function    names = ['America', 'Europe', 'Africa']    procs = []    proc = Process(target=print_func)  # instantiating without any argument    procs.append(proc)    proc.start()    # instantiating process with arguments    for name in names:        # print(name)        proc = Process(target=print_func, args=(name,))        procs.append(proc)        proc.start()    # complete the processes    for proc in procs:        proc.join()

The name of continent is :  AsiaThe name of continent is :  AmericaThe name of continent is :  EuropeThe name of continent is :  Africa

import threading def print_cube(num):    """    function to print cube of given num    """    print("Cube: {}".format(num * num * num)) def print_square(num):    """    function to print square of given num    """    print("Square: {}".format(num * num)) if __name__ == "__main__":    # creating thread    t1 = threading.Thread(target=print_square, args=(10,))    t2 = threading.Thread(target=print_cube, args=(10,))     # starting thread 1    t1.start()    # starting thread 2    t2.start()     # wait until thread 1 is completely executed    t1.join()    # wait until thread 2 is completely executed    t2.join()     # both threads completely executed    print("Done!")

Square: 100Cube: 1000Done!

def print_name(prefix):    print("Searching prefix:{}".format(prefix))    try :         while True:                # yeild used to create coroutine                name = (yield)                if prefix in name:                    print(name)    except GeneratorExit:            print("Closing coroutine!!") corou = print_name("Dear")corou.__next__()corou.send("James")corou.send("Dear James")corou.close()

Searching prefix:DearDear JamesClosing coroutine!!

• event loop управляет и распределяет выполнение различных задач. Он регистрирует их и обрабатывает распределение потока управления между ними.
• Корутины (о которых мы говорили выше) это специальные функции, работа которых схожа с работой генераторов в Python, с помощью await они возвращают поток управления обратно в event loop. Запуск корутины должен быть запланирован в event loop. Запланированные корутины будут обернуты в Tasks, что является типом Future.
• Future отражает результат таска, который может или не может быть выполнен. Результатом может быть exception.

import signal  import sys  import asyncio  import aiohttp  import jsonloop = asyncio.get_event_loop()  client = aiohttp.ClientSession(loop=loop)async def get_json(client, url):      async with client.get(url) as response:        assert response.status == 200        return await response.read()async def get_reddit_top(subreddit, client):      data1 = await get_json(client, 'https://www.reddit.com/r/' + subreddit + '/top.json?sort=top&t=day&limit=5')    j = json.loads(data1.decode('utf-8'))    for i in j['data']['children']:        score = i['data']['score']        title = i['data']['title']        link = i['data']['url']        print(str(score) + ': ' + title + ' (' + link + ')')    print('DONE:', subreddit + '\n')def signal_handler(signal, frame):      loop.stop()    client.close()    sys.exit(0)signal.signal(signal.SIGINT, signal_handler)asyncio.ensure_future(get_reddit_top('python', client))  asyncio.ensure_future(get_reddit_top('programming', client))  asyncio.ensure_future(get_reddit_top('compsci', client))  loop.run_forever()

50: Undershoot: Parsing theory in 1965 (http://personeltest.ru/away/jeffreykegler.github.io/Ocean-of-Awareness-blog/individual/2018/07/knuth_1965_2.html)12: Question about best-prefix/failure function/primal match table in kmp algorithm (http://personeltest.ru/aways/www.reddit.com/r/compsci/comments/8xd3m2/question_about_bestprefixfailure_functionprimal/)1: Question regarding calculating the probability of failure of a RAID system (http://personeltest.ru/aways/www.reddit.com/r/compsci/comments/8xbkk2/question_regarding_calculating_the_probability_of/)DONE: compsci336: /r/thanosdidnothingwrong -- banning people with python (http://personeltest.ru/aways/clips.twitch.tv/AstutePluckyCocoaLitty)175: PythonRobotics: Python sample codes for robotics algorithms (http://personeltest.ru/aways/atsushisakai.github.io/PythonRobotics/)23: Python and Flask Tutorial in VS Code (http://personeltest.ru/aways/code.visualstudio.com/docs/python/tutorial-flask)17: Started a new blog on Celery - what would you like to read about? (http://personeltest.ru/aways/www.python-celery.com)14: A Simple Anomaly Detection Algorithm in Python (http://personeltest.ru/aways/medium.com/@mathmare_/pyng-a-simple-anomaly-detection-algorithm-2f355d7dc054)DONE: python1360: git bundle (http://personeltest.ru/aways/dev.to/gabeguz/git-bundle-2l5o)1191: Which hashing algorithm is best for uniqueness and speed? Ian Boyd's answer (top voted) is one of the best comments I've seen on Stackexchange. (http://personeltest.ru/aways/softwareengineering.stackexchange.com/questions/49550/which-hashing-algorithm-is-best-for-uniqueness-and-speed)430: ARM launches Facts campaign against RISC-V (http://personeltest.ru/aways/riscv-basics.com/)244: Choice of search engine on Android nuked by Anonymous Coward (2009) (http://personeltest.ru/aways/android.googlesource.com/platform/packages/apps/GlobalSearch/+/592150ac00086400415afe936d96f04d3be3ba0c)209: Exploiting freely accessible WhatsApp data or Why does WhatsApp web know my phones battery level? (http://personeltest.ru/aways/medium.com/@juan_cortes/exploiting-freely-accessible-whatsapp-data-or-why-does-whatsapp-know-my-battery-level-ddac224041b4)DONE: programming

Использование Redis и Redis Queue RQ

from mymodule import count_words_at_urlfrom redis import Redisfrom rq import Queueq = Queue(connection=Redis())job = q.enqueue(count_words_at_url, 'http://nvie.com')******mymodule.py******import requestsdef count_words_at_url(url):    """Just an example function that's called async."""    resp = requests.get(url)    print( len(resp.text.split()))    return( len(resp.text.split()))

15:10:45 RQ worker 'rq:worker:EMPID18030.9865' started, version 0.11.015:10:45 *** Listening on default...15:10:45 Cleaning registries for queue: default15:10:50 default: mymodule.count_words_at_url('http://nvie.com') (a2b7451e-731f-4f31-9232-2b7e3549051f)32215:10:51 default: Job OK (a2b7451e-731f-4f31-9232-2b7e3549051f)15:10:51 Result is kept for 500 seconds

Заключение

Цвет функций

Дублирование кода

def fetch_resource_size(url: str) -> int:    response = client_get(url)    return len(response.content)

async def fetch_resource_size(url: str) -> int:    response = await client_get(url)    return len(response.content)

• синхронный код;
• асинхронный код.

Абстракция и композиция

def fetch_resource_size(url: str) -> Abstraction[int]:    return client_get(url).map(        lambda response: len(response.content),    )

def fetch_resource_size(url: str) -> AsyncAbstraction[int]:    return client_get(url).map(        lambda response: len(response.content),    )

def fetch_resource_size(    client_get: Callable[[str], AbstactionType[Response]],    url: str,) -> AbstactionType[int]:    return client_get(url).map(        lambda response: len(response.content),    )

IOResult и FutureResult

Синхронный вариант

pip install returns httpx anyio

from typing import Callable import httpx from returns.io import IOResultE, impure_safe def fetch_resource_size(    client_get: Callable[[str], IOResultE[httpx.Response]],    url: str,) -> IOResultE[int]:    return client_get(url).map(        lambda response: len(response.content),    ) print(fetch_resource_size(    impure_safe(httpx.get),    'https://sobolevn.me',))# => <IOResult: <Success: 27972>>

• Использовать IOResultE функциональный способ обработки ошибок синхронного IO (исключения не всегда подходят). Типы, основанные на Result, позволяют имитировать исключения, но с раздельными значениями Failure(). Успешные выходы при этом оборачиваются в тип Success. Обычно никому нет дела до исключений, а нам есть.
• Использовать httpx, который может работать с синхронными и асинхронными запросами.
• Использовать функцию impure_safe, чтобы преобразовывать тип, который возвращает httpx.get, в абстракцию IOResultE.

Асинхронный вариант

from typing import Callable import anyioimport httpx from returns.future import FutureResultE, future_safe def fetch_resource_size(    client_get: Callable[[str], FutureResultE[httpx.Response]],    url: str,) -> FutureResultE[int]:    return client_get(url).map(        lambda response: len(response.content),    ) page_size = fetch_resource_size(    future_safe(httpx.AsyncClient().get),    'https://sobolevn.me',)print(page_size)print(anyio.run(page_size.awaitable))# => <FutureResult: <coroutine object async_map at 0x10b17c320>># => <IOResult: <Success: 27972>>

• Синхронный IOResultE изменился на асинхронный FutureResultE, impure_safe на future_safe. Работает так же, но возвращает другую абстракцию: FutureResultE.
• Используется AsyncClient из httpx.
• Результирующее значение FutureResult необходимо запустить, потому что красные функции не могут вызывать сами себя.
• Утилита anyio используется, чтобы показать, что этот подход работает с любой асинхронной библиотекой: asyncio, trio, curio.

Два в одном

from typing import Callable, Union, overload import anyioimport httpx from returns.future import FutureResultE, future_safefrom returns.io import IOResultE, impure_safe @overloaddef fetch_resource_size(    client_get: Callable[[str], IOResultE[httpx.Response]],    url: str,) -> IOResultE[int]:    """Sync case.""" @overloaddef fetch_resource_size(    client_get: Callable[[str], FutureResultE[httpx.Response]],    url: str,) -> FutureResultE[int]:    """Async case.""" def fetch_resource_size(    client_get: Union[        Callable[[str], IOResultE[httpx.Response]],        Callable[[str], FutureResultE[httpx.Response]],    ],    url: str,) -> Union[IOResultE[int], FutureResultE[int]]:    return client_get(url).map(        lambda response: len(response.content),    )

# Sync:print(fetch_resource_size(    impure_safe(httpx.get),    'https://sobolevn.me',))# => <IOResult: <Success: 27972>> # Async:page_size = fetch_resource_size(    future_safe(httpx.AsyncClient().get),    'https://sobolevn.me',)print(page_size)print(anyio.run(page_size.awaitable))# => <FutureResult: <coroutine object async_map at 0x10b17c320>># => <IOResult: <Success: 27972>>

 mypy async_and_sync.pySuccess: no issues found in 1 source file

---lambda response: len(response.content),+++lambda response: response.content,

 mypy async_and_sync.pyasync_and_sync.py:33: error: Argument 1 to "map" of "IOResult" has incompatible type "Callable[[Response], bytes]"; expected "Callable[[Response], int]"async_and_sync.py:33: error: Argument 1 to "map" of "FutureResult" has incompatible type "Callable[[Response], bytes]"; expected "Callable[[Response], int]"async_and_sync.py:33: error: Incompatible return value type (got "bytes", expected "int")

from typing import Callable, TypeVarimport anyioimport httpxfrom returns.future import future_safefrom returns.interfaces.specific.ioresult import IOResultLike2from returns.io import impure_safefrom returns.primitives.hkt import Kind2, kinded_IOKind = TypeVar('_IOKind', bound=IOResultLike2)@kindeddef fetch_resource_size(    client_get: Callable[[str], Kind2[_IOKind, httpx.Response, Exception]],    url: str,) -> Kind2[_IOKind, int, Exception]:    return client_get(url).map(        lambda response: len(response.content),    )# Sync:print(fetch_resource_size(    impure_safe(httpx.get),    'https://sobolevn.me',))# => <IOResult: <Success: 27972>># Async:page_size = fetch_resource_size(    future_safe(httpx.AsyncClient().get),    'https://sobolevn.me',)print(page_size)print(anyio.run(page_size.awaitable))# => <FutureResult: <coroutine object async_map at 0x10b17c320>># => <IOResult: <Success: 27972>>

Больше возможностей dry-python

• Типизированные функции partial и @curry.

from returns.curry import curry, partial def example(a: int, b: str) -> float:    ... reveal_type(partial(example, 1))# note: Revealed type is 'def (b: builtins.str) -> builtins.float' reveal_type(curry(example))# note: Revealed type is 'Overload(def (a: builtins.int) -> def (b: builtins.str) -> builtins.float, def (a: builtins.int, b: builtins.str) -> builtins.float)'

@currydef example(a: int, b: str) -> float:    return float(a + len(b)) assert example(1, 'abc') == 4.0assert example(1)('abc') == 4.0

• Функциональные пайплайны с выводом типов.

from returns.pipeline import flowassert flow(    [1, 2, 3],    lambda collection: max(collection),    lambda max_number: -max_number,) == -3

• RequiresContextFutureResult для типизированной инъекции функциональных зависимостей.

Планы на будущее

• Реализовать Higher Kinded Types или их эмуляцию (issue).
• Добавить надлежащие type-классы, чтобы реализовать необходимые абстракции (issue).
• Возможно, попробовать компилятор mypyc, что потенциально позволит компилировать типизированные аннотированные Python-программы в двоичный файл. Тогда код с dry-python/returns будет работать в несколько раз быстрее (issue).
• Исследовать новые способы написания функционального кода на Python, например, такие как do-notation.

Выводы

Схема базы данных и миграции

import sqlalchemyfrom sqlalchemy.dialects.postgresql import UUIDmetadata = sqlalchemy.MetaData()users_table = sqlalchemy.Table(    "users",    metadata,    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),    sqlalchemy.Column("email", sqlalchemy.String(40), unique=True, index=True),    sqlalchemy.Column("name", sqlalchemy.String(100)),    sqlalchemy.Column("hashed_password", sqlalchemy.String()),    sqlalchemy.Column(        "is_active",        sqlalchemy.Boolean(),        server_default=sqlalchemy.sql.expression.true(),        nullable=False,    ),)tokens_table = sqlalchemy.Table(    "tokens",    metadata,    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),    sqlalchemy.Column(        "token",        UUID(as_uuid=False),        server_default=sqlalchemy.text("uuid_generate_v4()"),        unique=True,        nullable=False,        index=True,    ),    sqlalchemy.Column("expires", sqlalchemy.DateTime()),    sqlalchemy.Column("user_id", sqlalchemy.ForeignKey("users.id")),)

import sqlalchemyfrom .users import users_tablemetadata = sqlalchemy.MetaData()posts_table = sqlalchemy.Table(    "posts",    metadata,    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),    sqlalchemy.Column("user_id", sqlalchemy.ForeignKey(users_table.c.id)),    sqlalchemy.Column("created_at", sqlalchemy.DateTime()),    sqlalchemy.Column("title", sqlalchemy.String(100)),    sqlalchemy.Column("content", sqlalchemy.Text()),)

$ pip install alembic

$ alembic init migrations

• каталог versions, в котором будут хранится файлы миграций
• скрипт env.py, запускающийся при вызове alembic
• файл script.py.mako, содержащий шаблон для новых миграций.

sqlalchemy.url = postgresql://%(DB_USER)s:%(DB_PASS)s@%(DB_HOST)s:5432/%(DB_NAME)s

from os import environfrom alembic import contextfrom app.models import posts, users# Alembic Config объект предоставляет доступ# к переменным из файла alembic.iniconfig = context.configsection = config.config_ini_sectionconfig.set_section_option(section, "DB_USER", environ.get("DB_USER"))config.set_section_option(section, "DB_PASS", environ.get("DB_PASS"))config.set_section_option(section, "DB_NAME", environ.get("DB_NAME"))config.set_section_option(section, "DB_HOST", environ.get("DB_HOST"))fileConfig(config.config_file_name)target_metadata = [users.metadata, posts.metadata]

$ alembic revision --autogenerate -m "Added required tables"$ alembic upgrade head

Запускаем приложение и подключаем БД

from fastapi import FastAPIapp = FastAPI()@app.get("/")def read_root():    return {"Hello": "World"}

$ uvicorn main:app --reload

{"Hello": "World"}

from os import environimport databases# берем параметры БД из переменных окруженияDB_USER = environ.get("DB_USER", "user")DB_PASSWORD = environ.get("DB_PASSWORD", "password")DB_HOST = environ.get("DB_HOST", "localhost")DB_NAME = "async-blogs"SQLALCHEMY_DATABASE_URL = (    f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:5432/{DB_NAME}")# создаем объект database, который будет использоваться для выполнения запросовdatabase = databases.Database(SQLALCHEMY_DATABASE_URL)app = FastAPI()@app.on_event("startup")async def startup():    # когда приложение запускается устанавливаем соединение с БД    await database.connect()@app.on_event("shutdown")async def shutdown():    # когда приложение останавливается разрываем соединение с БД    await database.disconnect()@app.get("/")def read_root():    # изменим роут таким образом, чтобы он брал данные из БД    query = (        select(            [                posts_table.c.id,                posts_table.c.created_at,                posts_table.c.title,                posts_table.c.content,                posts_table.c.user_id,                users_table.c.name.label("user_name"),            ]        )        .select_from(posts_table.join(users_table))        .order_by(desc(posts_table.c.created_at))    )    return await database.fetch_all(query)

Валидация запроса и ответа

pip install pydantic

from pydantic import BaseModel, EmailStrclass UserCreate(BaseModel):    """ Проверяет sign-up запрос """    email: EmailStr    name: str    password: str

pip install email-validator

from typing import Optionalfrom pydantic import UUID4, BaseModel, EmailStr, Field, validatorclass UserCreate(BaseModel):    """ Проверяет sign-up запрос """    email: EmailStr    name: str    password: strclass UserBase(BaseModel):    """ Формирует тело ответа с деталями пользователя """    id: int    email: EmailStr    name: strclass TokenBase(BaseModel):    token: UUID4 = Field(..., alias="access_token")    expires: datetime    token_type: Optional[str] = "bearer"    class Config:        allow_population_by_field_name = True    @validator("token")    def hexlify_token(cls, value):        """ Конвертирует UUID в hex строку """        return value.hexclass User(UserBase):    """ Формирует тело ответа с деталями пользователя и токеном """    token: TokenBase = {}

import hashlibimport randomimport stringfrom datetime import datetime, timedeltafrom sqlalchemy import and_from app.models.database import databasefrom app.models.users import tokens_table, users_tablefrom app.schemas import users as user_schemadef get_random_string(length=12):    """ Генерирует случайную строку, использующуюся как соль """    return "".join(random.choice(string.ascii_letters) for _ in range(length))def hash_password(password: str, salt: str = None):    """ Хеширует пароль с солью """    if salt is None:        salt = get_random_string()    enc = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), 100_000)    return enc.hex()def validate_password(password: str, hashed_password: str):    """ Проверяет, что хеш пароля совпадает с хешем из БД """    salt, hashed = hashed_password.split("$")    return hash_password(password, salt) == hashedasync def get_user_by_email(email: str):    """ Возвращает информацию о пользователе """    query = users_table.select().where(users_table.c.email == email)    return await database.fetch_one(query)async def get_user_by_token(token: str):    """ Возвращает информацию о владельце указанного токена """    query = tokens_table.join(users_table).select().where(        and_(            tokens_table.c.token == token,            tokens_table.c.expires > datetime.now()        )    )    return await database.fetch_one(query)async def create_user_token(user_id: int):    """ Создает токен для пользователя с указанным user_id """    query = (        tokens_table.insert()        .values(expires=datetime.now() + timedelta(weeks=2), user_id=user_id)        .returning(tokens_table.c.token, tokens_table.c.expires)    )    return await database.fetch_one(query)async def create_user(user: user_schema.UserCreate):    """ Создает нового пользователя в БД """    salt = get_random_string()    hashed_password = hash_password(user.password, salt)    query = users_table.insert().values(        email=user.email, name=user.name, hashed_password=f"{salt}${hashed_password}"    )    user_id = await database.execute(query)    token = await create_user_token(user_id)    token_dict = {"token": token["token"], "expires": token["expires"]}    return {**user.dict(), "id": user_id, "is_active": True, "token": token_dict}

from fastapi import APIRouterfrom app.schemas import usersfrom app.utils import users as users_utilsrouter = APIRouter()@router.post("/sign-up", response_model=users.User)async def create_user(user: users.UserCreate):    db_user = await users_utils.get_user_by_email(email=user.email)    if db_user:        raise HTTPException(status_code=400, detail="Email already registered")    return await users_utils.create_user(user=user)

from app.routers import usersapp.include_router(users.router)

Аутентификация и контроль доступа

from fastapi import Dependsfrom fastapi.security import OAuth2PasswordRequestForm@router.post("/auth", response_model=users.TokenBase)async def auth(form_data: OAuth2PasswordRequestForm = Depends()):    user = await users_utils.get_user_by_email(email=form_data.username)    if not user:        raise HTTPException(status_code=400, detail="Incorrect email or password")    if not users_utils.validate_password(        password=form_data.password, hashed_password=user["hashed_password"]    ):        raise HTTPException(status_code=400, detail="Incorrect email or password")    return await users_utils.create_user_token(user_id=user["id"])

from app.utils import users as users_utilsfrom fastapi import Depends, HTTPException, statusfrom fastapi.security import OAuth2PasswordBeareroauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth")async def get_current_user(token: str = Depends(oauth2_scheme)):    user = await users_utils.get_user_by_token(token)    if not user:        raise HTTPException(            status_code=status.HTTP_401_UNAUTHORIZED,            detail="Invalid authentication credentials",            headers={"WWW-Authenticate": "Bearer"},        )    if not user["is_active"]:        raise HTTPException(            status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user"        )    return user

from app.utils.dependencies import get_current_user@router.get("/users/me", response_model=users.UserBase)async def read_users_me(current_user: users.User = Depends(get_current_user)):    return current_user

from datetime import datetimefrom app.models.database import databasefrom app.models.posts import posts_tablefrom app.models.users import users_tablefrom app.schemas import posts as post_schemafrom sqlalchemy import desc, func, selectasync def create_post(post: post_schema.PostModel, user):    query = (        posts_table.insert()        .values(            title=post.title,            content=post.content,            created_at=datetime.now(),            user_id=user["id"],        )        .returning(            posts_table.c.id,            posts_table.c.title,            posts_table.c.content,            posts_table.c.created_at,        )    )    post = await database.fetch_one(query)    # Convert to dict and add user_name key to it    post = dict(zip(post, post.values()))    post["user_name"] = user["name"]    return postasync def get_post(post_id: int):    query = (        select(            [                posts_table.c.id,                posts_table.c.created_at,                posts_table.c.title,                posts_table.c.content,                posts_table.c.user_id,                users_table.c.name.label("user_name"),            ]        )        .select_from(posts_table.join(users_table))        .where(posts_table.c.id == post_id)    )    return await database.fetch_one(query)async def get_posts(page: int):    max_per_page = 10    offset1 = (page - 1) * max_per_page    query = (        select(            [                posts_table.c.id,                posts_table.c.created_at,                posts_table.c.title,                posts_table.c.content,                posts_table.c.user_id,                users_table.c.name.label("user_name"),            ]        )        .select_from(posts_table.join(users_table))        .order_by(desc(posts_table.c.created_at))        .limit(max_per_page)        .offset(offset1)    )    return await database.fetch_all(query)async def get_posts_count():    query = select([func.count()]).select_from(posts_table)    return await database.fetch_val(query)async def update_post(post_id: int, post: post_schema.PostModel):    query = (        posts_table.update()        .where(posts_table.c.id == post_id)        .values(title=post.title, content=post.content)    )    return await database.execute(query)

from app.schemas.posts import PostDetailsModel, PostModelfrom app.schemas.users import Userfrom app.utils import posts as post_utilsfrom app.utils.dependencies import get_current_userfrom fastapi import APIRouter, Depends, HTTPException, statusrouter = APIRouter()@router.post("/posts", response_model=PostDetailsModel, status_code=201)async def create_post(post: PostModel, current_user: User = Depends(get_current_user)):    post = await post_utils.create_post(post, current_user)    return post@router.get("/posts")async def get_posts(page: int = 1):    total_cout = await post_utils.get_posts_count()    posts = await post_utils.get_posts(page)    return {"total_count": total_cout, "results": posts}@router.get("/posts/{post_id}", response_model=PostDetailsModel)async def get_post(post_id: int):    return await post_utils.get_post(post_id)@router.put("/posts/{post_id}", response_model=PostDetailsModel)async def update_post(    post_id: int, post_data: PostModel, current_user=Depends(get_current_user)):    post = await post_utils.get_post(post_id)    if post["user_id"] != current_user["id"]:        raise HTTPException(            status_code=status.HTTP_403_FORBIDDEN,            detail="You don't have access to modify this post",        )    await post_utils.update_post(post_id=post_id, post=post_data)    return await post_utils.get_post(post_id)

from app.routers import postsapp.include_router(posts.router)

Тестирование

$ pip install pytest

from app.main import appfrom fastapi.testclient import TestClientclient = TestClient(app)def test_health_check():    response = client.get("/")    assert response.status_code == 200    assert response.json() == {"Hello": "World"}

from os import environimport databasesDB_USER = environ.get("DB_USER", "user")DB_PASSWORD = environ.get("DB_PASSWORD", "password")DB_HOST = environ.get("DB_HOST", "localhost")TESTING = environ.get("TESTING")if TESTING:    # Используем отдельную базу данных для тестов    DB_NAME = "async-blogs-temp-for-test"    TEST_SQLALCHEMY_DATABASE_URL = (        f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:5432/{DB_NAME}"    )    database = databases.Database(TEST_SQLALCHEMY_DATABASE_URL)else:    DB_NAME = "async-blogs"    SQLALCHEMY_DATABASE_URL = (        f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:5432/{DB_NAME}"    )    database = databases.Database(SQLALCHEMY_DATABASE_URL)

import osimport pytest# Устанавливаем `os.environ`, чтобы использовать тестовую БДos.environ['TESTING'] = 'True'from alembic import commandfrom alembic.config import Configfrom app.models import databasefrom sqlalchemy_utils import create_database, drop_database@pytest.fixture(scope="module")def temp_db():    create_database(database.TEST_SQLALCHEMY_DATABASE_URL) # Создаем БД    base_dir = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))    alembic_cfg = Config(os.path.join(base_dir, "alembic.ini")) # Загружаем конфигурацию alembic     command.upgrade(alembic_cfg, "head") # выполняем миграции    try:        yield database.TEST_SQLALCHEMY_DATABASE_URL    finally:        drop_database(database.TEST_SQLALCHEMY_DATABASE_URL) # удаляем БД

def test_sign_up(temp_db):    request_data = {        "email": "vader@deathstar.com",        "name": "Darth Vader",        "password": "rainbow"    }    with TestClient(app) as client:        response = client.post("/sign-up", json=request_data)    assert response.status_code == 200    assert response.json()["id"] == 1    assert response.json()["email"] == "vader@deathstar.com"    assert response.json()["name"] == "Darth"    assert response.json()["token"]["expires"] is not None    assert response.json()["token"]["access_token"] is not None

import asynciofrom app.main import appfrom app.schemas.users import UserCreatefrom app.utils.users import create_user, create_user_tokenfrom fastapi.testclient import TestClientdef test_create_post(temp_db):    user = UserCreate(        email="vader@deathstar.com",        name="Darth",        password="rainbow"    )    request_data = {      "title": "42",      "content": "Don't panic!"    }    with TestClient(app) as client:        # Create user and use his token to add new post        loop = asyncio.get_event_loop()        user_db = loop.run_until_complete(create_user(user))        response = client.post(            "/posts",            json=request_data,            headers={"Authorization": f"Bearer {user_db['token']['token']}"}        )    assert response.status_code == 201    assert response.json()["id"] == 1    assert response.json()["title"] == "42"    assert response.json()["content"] == "Don't panic!"def test_create_post_forbidden_without_token(temp_db):    request_data = {      "title": "42",      "content": "Don't panic!"    }    with TestClient(app) as client:        response = client.post("/posts", json=request_data)    assert response.status_code == 401def test_posts_list(temp_db):    with TestClient(app) as client:        response = client.get("/posts")    assert response.status_code == 200    assert response.json()["total_count"] == 1    assert response.json()["results"][0]["id"] == 1    assert response.json()["results"][0]["title"] == "42"    assert response.json()["results"][0]["content"] == "Don't panic!"def test_post_detail(temp_db):    post_id = 1    with TestClient(app) as client:        response = client.get(f"/posts/{post_id}")    assert response.status_code == 200    assert response.json()["id"] == 1    assert response.json()["title"] == "42"    assert response.json()["content"] == "Don't panic!"def test_update_post(temp_db):    post_id = 1    request_data = {      "title": "42",      "content": "Life? Don't talk to me about life."    }    with TestClient(app) as client:        # Create user token to add new post        loop = asyncio.get_event_loop()        token = loop.run_until_complete(create_user_token(user_id=1))        response = client.put(            f"/posts/{post_id}",            json=request_data,            headers={"Authorization": f"Bearer {token['token']}"}        )    assert response.status_code == 200    assert response.json()["id"] == 1    assert response.json()["title"] == "42"    assert response.json()["content"] == "Life? Don't talk to me about life."def test_update_post_forbidden_without_token(temp_db):    post_id = 1    request_data = {      "title": "42",      "content": "Life? Don't talk to me about life."    }    with TestClient(app) as client:        response = client.put(f"/posts/{post_id}", json=request_data)    assert response.status_code == 401

import asyncioimport pytestfrom app.main import appfrom app.schemas.users import UserCreatefrom app.utils.users import create_user, create_user_tokenfrom fastapi.testclient import TestClientdef test_sign_up(temp_db):    request_data = {        "email": "vader@deathstar.com",        "name": "Darth",        "password": "rainbow"    }    with TestClient(app) as client:        response = client.post("/sign-up", json=request_data)    assert response.status_code == 200    assert response.json()["id"] == 1    assert response.json()["email"] == "vader@deathstar.com"    assert response.json()["name"] == "Darth"    assert response.json()["token"]["expires"] is not None    assert response.json()["token"]["token"] is not Nonedef test_login(temp_db):    request_data = {"username": "vader@deathstar.com", "password": "rainbow"}    with TestClient(app) as client:        response = client.post("/auth", data=request_data)    assert response.status_code == 200    assert response.json()["token_type"] == "bearer"    assert response.json()["expires"] is not None    assert response.json()["access_token"] is not Nonedef test_login_with_invalid_password(temp_db):    request_data = {"username": "vader@deathstar.com", "password": "unicorn"}    with TestClient(app) as client:        response = client.post("/auth", data=request_data)    assert response.status_code == 400    assert response.json()["detail"] == "Incorrect email or password"def test_user_detail(temp_db):    with TestClient(app) as client:        # Create user token to see user info        loop = asyncio.get_event_loop()        token = loop.run_until_complete(create_user_token(user_id=1))        response = client.get(            "/users/me",            headers={"Authorization": f"Bearer {token['token']}"}        )    assert response.status_code == 200    assert response.json()["id"] == 1    assert response.json()["email"] == "vader@deathstar.com"    assert response.json()["name"] == "Darth"def test_user_detail_forbidden_without_token(temp_db):    with TestClient(app) as client:        response = client.get("/users/me")    assert response.status_code == 401@pytest.mark.freeze_time("2015-10-21")def test_user_detail_forbidden_with_expired_token(temp_db, freezer):    user = UserCreate(        email="sidious@deathstar.com",        name="Palpatine",        password="unicorn"    )    with TestClient(app) as client:        # Create user and use expired token        loop = asyncio.get_event_loop()        user_db = loop.run_until_complete(create_user(user))        freezer.move_to("'2015-11-10'")        response = client.get(            "/users/me",            headers={"Authorization": f"Bearer {user_db['token']['token']}"}        )    assert response.status_code == 401

P.S. Исходники

Привет, Хабр!

Сегодня я хочу рассказать о замечательной библиотеке для разработке ботов ВКонтакте с помощью языка программирования Python.

VKWave

VKWave это фреймворк для разработки ботов ВКонтакте, написанный с помощью asyncio. Основные цели проекта дать возможность разработчику конфигурировать фреймворк максимально под себя, в тоже время обеспечивая достойную скорость разработки.

Минимальная требуемая версия Python 3.7

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

Также у нас есть чат в Telegram

Установка

Установка очень проста и содержит в себе всего одну команду:

pip install vkwave

А теперь предлагаю перейти к примерам!

Echo-бот

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

# Импортируем нужные классы.# SimpleLongPollBot: обёртка для более удобной работы с фреймворком# SimpleBotEvent: тип события, который предоставляет SimpleLongPollBotfrom vkwave.bots import SimpleLongPollBot, SimpleBotEvent# инициализируем бота (можно ввести список токенов, тогда vkwave сможет обходить лимиты ВКонтакте)bot = SimpleLongPollBot(tokens=TOKEN, group_id=GROUP_ID)# декоратор для создания обработчиков.# можно передавать свои фильтры, но в данном случае мы хотим принимать все сообщения@bot.message_handler()def echo(event: SimpleBotEvent) -> str:    # мы можем сразу возвращать текст, т.к vkwave понимает, что если вы возвращаете строку, то вы хотите ответить на сообщение этим текстом. пользователь может задать свои типы данных, которые он сможет возвращать из хендлеров (а также написать нужную логику для их преобразования в нужные действия)    return event.object.object.message.text# запускаем бота с игнорированием ошибок (не останавливаться даже при них)bot.run_forever()

Код содержит буквально пару строк и выглядит очень легко.

Предлагаю написать что-нибудь поинтереснее. Давайте напишем такого же бота, но он будет печатать в ответ только тот текст, который идёт как аргументы команды /echo. Например /echo мой текст.

Вторая версия Echo-бота

# мы используем фильтр для команд. он фильтрует все сообщения которые не выглядит как `/<наша команда>`. можно задать свои префиксы, а также передать список команд@bot.message_handler(bot.command_filter("echo"))def echo(event: SimpleBotEvent) -> str:    # получаем все аргуметы команды    args = event.object.object.message.text.split()    # проверяем, что есть хотя бы один аргумент    # в противном случае - пишем, что пользователь должен ввести какой-нибудь текст    if len(args) < 2:        return "Напиши какой-нибудь текст!"    # возвращаем итоговый текст (соединяем все аргументы через пробел)    return " ".join(args[1:])

Уже интереснее. Как мы видим, VKWave предоставляет нам фильтр для команд. Скажу сразу: стандартных фильтров в VKWave много, а также вы можете написать свои.

Итоговая версия Echo-бота

А сейчас мы попробуем написать свои фильтры для того, чтобы весь наш обработчик выглядел максимально просто.

# Импортируем нужные классы для разработки своих фильтровfrom vkwave.bots.core.dispatching.filters.base import BaseFilter, BaseEvent, FilterResult# объявляем свой фильтр, который наследуется от базовогоclass EchoFilter(BaseFilter):    # мы можем определить `__init__` для дополнительной настройки фильтра    # объявляем асинхронный метод `check`, который принимает событие и возвращает результат фильтра    async def check(self, event: BaseEvent) -> FilterResult:        # делаем алиас для текста сообщения        text = event.object.object.message.text        # разбиваем сообщение по пробелам        all_args = text.split()        # если частей меньше двух - фильтр не прошёл        if len(all_args) < 2:            # возвращаем False.            # так же можем, например, что-то написать пользователю            # у нас есть `event.api_ctx`, который предоставляет лёгкий доступ ко всем методам            return FilterResult(False)        # если нулевой аргумент (сама команда) не "/echo" возвращаем False        if all_args[0] != "/echo":            return FilterResult(False)        # передаём обработчику уже готовый ответ на сообщение        event["echo_answer"] = " ".join(all_args[1:])        return FilterResult(True)# используем фильтр@bot.message_handler(EchoFilter())def echo(event: SimpleBotEvent) -> str:    # возвращаем текст, который мы уже "собрали" в фильтре    return event["echo_answer"]

Заключение

Конечно, я показал вам не все, вообще не все, возможности VKWave. Они очень широки, начиная от возможности своих фильтров и middlewares, типов обработчиков, способов получения событий, заканчивая мультиботом, собственным HTTP клиентом и многое другое!

Этой статьёй я хочу мотивировать вас попробовать создать своего бота ВКонтакте, не используя не удобные vk_api и vk.

Репозиторий на GitHub
Наш чат в Telegram

1. Введение

2. Совсем простой пример с asyncio

import asyncioimport timeasync def spider(site_name): for page in range(1, 4):     await asyncio.sleep(1)     print(site_name, page)spiders = [ asyncio.ensure_future(spider("Blog")), asyncio.ensure_future(spider("News")), asyncio.ensure_future(spider("Forum"))]start = time.time()event_loop = asyncio.get_event_loop()event_loop.run_until_complete(asyncio.gather(*spiders))event_loop.close()print("{:.2F}".format(time.time() - start))

import timeclass PSleep:    def __init__(self, t, p_FSM): self.SetTime = t; self.nState = 0; self.bIfLoop = False; self.p_mainFSM = p_FSM    def x1(self): return time.time() - self.t0 <= self.SetTime    def y1(self): self.t0 = time.time()    def loop(self):        if (self.nState == 0): self.y1(); self.nState = 1        elif (self.nState == 1):            if (not self.x1()): self.nState = 4

# "паук" для страницы "Blog"class PBSpider:    def __init__(self, name):        self.nState = 0; self.bIfLoop = True; self.site_name = name; self.page = 1;        self.p_mainFSM = b_sleep;    def x1(self): return self.page < 4    def y1(self):        self.bIfLoop = False; automaton.append(b_sleep);        b_sleep.p_mainFSM = blog        automaton[-1].bIfLoop = True;        automaton[-1].nState = 0    def y2(self): print(self.site_name, self.page)    def y3(self): self.page += 1    def y4(self): self.page = 1    def loop(self):        if (self.x1() and self.nState == 0):  self.y1(); self.nState = 1        elif (not self.x1()  and self.nState == 0): self.y1(); self.y4(); self.nState = 33        elif (self.nState == 1): self.y2(); self.y3(); self.nState = 0# "паук" для страницы "News"class PNSpider:    def __init__(self, name):        self.nState = 0; self.bIfLoop = True; self.site_name = name; self.page = 1;        self.p_mainFSM = n_sleep;    def x1(self): return self.page < 4    def y1(self):        self.bIfLoop = False; automaton.append(n_sleep);        n_sleep.p_mainFSM = news        automaton[-1].bIfLoop = True;        automaton[-1].nState = 0    def y2(self): print(self.site_name, self.page)    def y3(self): self.page += 1    def y4(self): self.page = 1    def loop(self):        if (self.x1() and self.nState == 0):  self.y1(); self.nState = 1        elif (not self.x1()  and self.nState == 0): self.y1(); self.y4(); self.nState = 33        elif (self.nState == 1): self.y2(); self.y3(); self.nState = 0# паук для страницы "Forum"class PFSpider:    def __init__(self, name):        self.nState = 0; self.bIfLoop = True; self.site_name = name; self.page = 1;        self.p_mainFSM = f_sleep;    def x1(self): return self.page < 4    def y1(self):        self.bIfLoop = False; automaton.append(f_sleep);        f_sleep.p_mainFSM = forum        automaton[-1].bIfLoop = True;        automaton[-1].nState = 0    def y2(self): print(self.site_name, self.page)    def y3(self): self.page += 1    def y4(self): self.page = 1    def loop(self):        if (self.x1() and self.nState == 0):  self.y1(); self.nState = 1        elif (not self.x1()  and self.nState == 0): self.y1(); self.y4(); self.nState = 33        elif (self.nState == 1): self.y2(); self.y3(); self.nState = 0# задержкиb_sleep = PSleep(1, 0)n_sleep = PSleep(1, 0)f_sleep = PSleep(1, 0)# "пауки"blog = PBSpider("Blog")news = PNSpider("News")forum = PFSpider("Forum")# формирование исходного списка процессовautomaton = []automaton.append(blog);automaton.append(news);automaton.append(forum);start = time.time()# управление процессами (аналог event_loop)while True:    ind = 0;    while True:        while ind < len(automaton):            if automaton[ind].nState == 4:                automaton[ind].p_mainFSM.bIfLoop = True                automaton.pop(ind)                ind -=1            elif automaton[ind].bIfLoop:                automaton[ind].loop()            elif automaton[ind].nState == 33:                print("{:.2F}".format(time.time() - start))                exit()            ind += 1        ind = 0

Рис. 1. Автоматная и блок-схемная модели самолета-паука

3. О проблемах реализации автоматов в Python

// Задержка (в дискретном времени)#include "lfsaappl.h"#include <QTime>class FAwaitSleep :    public LFsaAppl{public:    FAwaitSleep(int n);protected:    int x1();    QTime time;    int nAwaitSleep;};#include "stdafx.h"#include "FAwaitSleep.h"static LArc TBL_AwaitSleep[] = {    LArc("s1","s1","x1",  "--"),//    LArc("s1","00","^x1","--"),//    LArc()};FAwaitSleep::FAwaitSleep(int n):    LFsaAppl(TBL_AwaitSleep, "FAwaitSleep"){    nAwaitSleep = n; time.start();}int FAwaitSleep::x1() { return time.elapsed() < nAwaitSleep; }

// "Паук". Имитация страниц сайта#include "lfsaappl.h"class FAwaitSleep;class FSpider :    public LFsaAppl{public:    LFsaAppl* Create(CVarFSA *pCVF) { Q_UNUSED(pCVF)return new FSpider(nameFsa); }    bool FCreationOfLinksForVariables() override;    FSpider(string strNam);    virtual ~FSpider(void);    CVar *pVarStrSiteName;// имя сайта    FAwaitSleep *pFAwaitSleep{nullptr};protected:    int x1(); void y1(); void y2(); void y3(); void y4();    int page{1};};#include "stdafx.h"#include "FSpider.h"#include "FSleep.h"#include "FAwaitSleep.h"#include <QDebug>static LArc TBL_Spider[] = {    LArc("st","s1","--","--"),    LArc("s1","s2","x1","y1"),  // x1- номер<макс.числа страниц; y1-задержка;    LArc("s2","s3","--","y2"),  // y2- печатать номера страницы;    LArc("s3","s1","--","y3"),  // y3- увеличить номер страницы    LArc("s1","st","^x1","y4"), // y4- сбросит номера страницы    LArc()};FSpider::FSpider(string strNam):    LFsaAppl(TBL_Spider, strNam){ }FSpider::~FSpider(void) { if (pFAwaitSleep) delete pFAwaitSleep; }bool FSpider::FCreationOfLinksForVariables() {    pVarStrSiteName = CreateLocVar("strSiteName", CLocVar::vtString, "name of site");    return true;}// счетчик страниц меньше заданного числа страниц?int FSpider::x1() { return page < 4; }// create delay - pure sleep (synchronous function) or await sleep (asynchronous function)void FSpider::y1() {    //sleep(1000);    // await sleep (asynchronous function)    if (pFAwaitSleep) delete pFAwaitSleep;    pFAwaitSleep = new FAwaitSleep(1000);    pFAwaitSleep->FCall(this);}void FSpider::y2() {#ifdef QT_DEBUG    string str = pVarStrSiteName->strGetDataSrc();    printf("%s%d", str.c_str(), page);    qDebug()<<str.c_str()<<page;#endif}void FSpider::y3() { page++; }void FSpider::y4() { page = 1; }

4. Выводы

import asyncioimport time# # Easy Python. Asyncio в python 3.7 https://www.youtube.com/watch?v=PaY-hiuE5iE# # 10:10# async def teatime():#     await asyncio.sleep(1)#     print('take a cap of tea')#     await asyncio.sleep(1)## async def read():#     print('Reading for 1 hour...')#     await teatime()#     print('...reading for 1 hour...')## if __name__ == '__main__':#     asyncio.run(read())class PSleep:    def __init__(self, t, p_FSM): self.SetTime = t; self.nState = 0; self.bIfLoop = False; self.p_mainFSM = p_FSM    def x1(self): return time.time() - self.t0 <= self.SetTime    def y1(self): self.t0 = time.time()    def loop(self):        if (self.nState == 0): self.y1(); self.nState = 1        elif (self.nState == 1):            if (not self.x1()): self.nState = 4class PTeaTime:    def __init__(self, p_FSM): self.nState = 0; self.bIfLoop = False; self.p_mainFSM = p_FSM;    def y1(self): self.bIfLoop = False; automaton.append(sl); automaton[-1].bIfLoop = True; automaton[-1].nState = 0    def y2(self): print('take a cap of tea')    def loop(self):        if (self.nState == 0):  self.y1(); self.nState = 1        elif (self.nState == 1): self.y2(); self.nState = 2        elif (self.nState == 2): self.y1(); self.nState = 3        elif (self.nState == 3): self.nState = 4class PRead:    def __init__(self): self.nState = 0; self.bIfLoop = False;    def y1(self): print('Reading for 1 hour...')    def y2(self): self.bIfLoop = False; automaton.append(rt); automaton[-1].bIfLoop = True; automaton[-1].nState = 0    def loop(self):        if (self.nState == 0): self.y1(); self.nState = 1        elif (self.nState == 1): self.y2(); self.nState = 2        elif (self.nState == 2): self.y1(); self.nState = 33; self.bIfLoop = Falseread = PRead()rt = PTeaTime(read)sl = PSleep(5, rt)automaton = []automaton.append(read); automaton[-1].bIfLoop = Truewhile True:    ind = 0;    while True:        while ind < len(automaton):            if automaton[ind].nState == 4:                automaton[ind].p_mainFSM.bIfLoop = True                automaton.pop(ind)                ind -=1            elif automaton[ind].bIfLoop:                automaton[ind].loop()            elif automaton[ind].nState == 33:                exit()            ind += 1        ind = 0

Рис. 2. Автоматная модель чаепития

#include "lfsaappl.h"class FRead :    public LFsaAppl{public:    LFsaAppl* Create(CVarFSA *pCVF) { Q_UNUSED(pCVF)return new FRead(nameFsa); }    FRead(string strNam);    virtual ~FRead(void);protected:    void y1(); void y2();  void y3();    LFsaAppl *pFRealTime{nullptr};};#include "stdafx.h"#include "FRead.h"#include "FTeaTime.h"#include <QDebug>static LArc TBL_Read[] = {    LArc("s1","s2","--","y1"),// Reading for 1 hour...    LArc("s2","s3","--","y2"),// Call(TeaTime)    LArc("s3","s4","--","y1"),// Reading for 1 hour...    LArc("s4","s5","--","y3"),// sleep(5)    LArc("s5","s1","--","--"),//    LArc()};FRead::FRead(string strNam):    LFsaAppl(TBL_Read, strNam){ }FRead::~FRead(void) { if (pFRealTime) delete pFRealTime; }void FRead::y1() {#ifdef QT_DEBUG    qDebug()<<"Reading for 1 hour...";#endif}void FRead::y2() {    if (pFRealTime) delete pFRealTime;    pFRealTime = new FTeaTime("TeaTime");    pFRealTime->FCall(this);}void FRead::y3() { FCreateDelay(5000); }#include "lfsaappl.h"class FTeaTime :    public LFsaAppl{public:    FTeaTime(string strNam);protected:    void y1(); void y2();};#include "stdafx.h"#include "FTeaTime.h"#include <QDebug>#include "./LSYSLIB/FDelay.h"static LArc TBL_TeaTime[] = {    LArc("s1","s2","--","y1"),// sleep(1)    LArc("s2","s3","--","y2"),// take a cap of tea    LArc("s3","s4","--","y1"),// sleep(1)    LArc("s4","00","--","--"),//    LArc()};FTeaTime::FTeaTime(string strNam):    LFsaAppl(TBL_TeaTime, strNam){ }void FTeaTime::y1() { FCreateDelay(2000); }void FTeaTime::y2() {#ifdef QT_DEBUG    qDebug()<<"take a cap of tea";#endif}

Литература

Введение

Сегодня мы будем рассматривать такой достаточно известный музыкальный сервис, как Yandex.Music. Хороший в целом сервис, но с существенным недостатком невозможностью работы оффлайн. Мы попробуем исправить это досадное недоразумение, используя подручные инструменты.

Инструментарий

Итак, нам понадобится:

• Относительно свежий python: 3.7 и выше
• Всякая асинхронщина: aiohttp и aiofile
• Классический инструмент для работы с html-API: BeautifulSoup
• Для развлечения пользователя во время процесса: tqdm
• Для заполнения тэгов: mutagen

Авторизация

Неавторизованным пользователям сервиса доступны только отрезки песен длинной до 30 секунд. Этого явно недостаточно для качественного прослушивания. Авторизоваться мы будем самым что ни на есть естественным способом, через веб-форму с получением печенюшек. В этом нам поможет opener для выполнения запросов и HTMLParser для парсинга форм.

def resolve_cookie(login: str, password: str) -> str:    cookies = CookieJar()    opener = urllib.request.build_opener(        urllib.request.HTTPCookieProcessor(cookies),        urllib.request.HTTPRedirectHandler())    response = opener.open("https://passport.yandex.ru")    doc = response.read()    parser = FormParser()    parser.feed(doc.decode("utf-8"))    parser.close()    parser.params["login"] = login    response = opener.open(parser.url or response.url, urllib.parse.urlencode(parser.params).encode("utf-8"))    doc = response.read()    parser = FormParser()    parser.feed(doc.decode("utf-8"))    parser.close()    parser.params["login"] = login    parser.params["passwd"] = password    response = opener.open(parser.url or response.url, urllib.parse.urlencode(parser.params).encode("utf-8"))    cookie_data = {}    for item in cookies:        if item.domain == ".yandex.ru":            cookie_data[item.name] = item.value    if "yandex_login" not in cookie_data:        keys = ", ".join(cookie_data.keys())        raise Exception(f"Invalid cookie_data {keys}")    return "; ".join(map(lambda v: f"{v[0]}={v[1]}", cookie_data.items()))

По адресу https://passport.yandex.ru нас ожидает первая форма с полем login. Если будет указан актуальный логин, то следующая форма предложит ввести пароль, что мы незамедлительно сделаем. Результат отправки формы нас особо не интересует. Наша цель кукисы. Если среди полученых кук окажется yandex_login, то авторизация прошла успешно. Вариант с двухфакторной аутентификацией мы пока не рассматриваем.

Yandex Music (HTML) API

Следующим нашим шагом будет получение списка избранных пользователем исполнителей. Чтобы показать наш стиль и молодёжность, для запросов применим aiohttp. Полученный html распарсим с помощью BeautifulSoup. Это классический инструмент, который используют ребята, заинтересованные в получении контента с чужих веб-страниц.

class YandexMusicApi:    host = "music.yandex.ru"    base_url = f"https://{host}"    def __init__(self, cookie: str):        self.headers = Headers(self.host, cookie)    async def _request(self, end_point: str):        async with aiohttp.ClientSession() as session:            url = f"{self.base_url}/{end_point}"            async with session.request(method="GET", url=url) as response:                return await response.read()    async def get_favorite_artists(self, login: str) -> List[Artist]:        body = await self._request(f"users/{login}/artists")        soup = BeautifulSoup(body, "lxml")        artists_soup = soup.find("div", class_="page-users__artists")        if artists_soup is None:            caption = soup.find("div", class_="page-users__caption")            if caption:                raise Exception(caption.contents[0])        result = []        for artist_soup in artists_soup.find_all("div", class_="artist"):            title_soup = artist_soup.find("div", class_="artist__name")            title = title_soup.attrs["title"]            title_href_soup = title_soup.find("a")            id_ = int(title_href_soup.attrs["href"].split("/")[-1])            result.append(Artist(id_, title))        return result

Здесь всё довольно просто, на странице https://music.yandex.ru/users/<login>/artists в блоке page-users__artists отображается требуемый нам список. Имя исполнителя можно обнаружить в аттрибуте title блока artist__name. Id исполнителя лёгким движением split извлекается из адреса ссылки.
Аналогичные действия позволят нам получить список альбомов исполнителя, а также список треков в альбоме.

Проникновение в хранилище

В этом месте мы уже потираем наши шаловливые ручки в ожидании прослушивания своей музыкальной коллекции. Но если бы всё было так просто, то я бы не писал эту статью. Нас ожидает новый противник yandex-хранилище. Для проникновения в хранилище потребуется воспроизвести жонглирование ссылками, которое происходит на страницах музыкального ресурса. При внимательном изучении вкладки Network браузерного инструментария выяснилось, что все элементы запроса легко выводятся за исключением одного участка в адресе https://{host}/get-mp3/{sign}/{ts}/{path}, который после непродолжительного гугления получил имя sign. Соль для сигнатуры (XGRlBW9FXlekgbPrRHuSiA) тоже была обнаружена на просторах интернета. Скорее всего не составит большого труда извлечь это значение из исходников страницы, но это уже не требуется.

    async def get_track_url(self, album_id: int, track_id: int) -> str:        async with aiohttp.ClientSession() as session:            url = f"{self.base_url}/api/v2.1/handlers/track/{track_id}:{album_id}/" \                  f"web-album-track-track-main/download/m?hq=0&external-domain={self.host}&overembed=no&__t={timestamp()}"            page = f"album/{album_id}"            headers = self.headers.build(page)            async with session.request(method="GET", url=url, headers=headers) as response:                body = await response.json()                src = body["src"]                src += f"&format=json&external-domain={self.host}&overembed=no&__t={timestamp()}"                result = parse.urlparse(src)                headers = self.headers.build(page, {                    ":authority": "storage.mds.yandex.net",                    ":method": "GET",                    ":path": f"{result.path}/{result.query}",                    ":scheme": "https",                }, True)                async with session.request(method="GET", url=src, headers=headers) as response:                    body = await response.json()                    host = body["host"]                    path = body["path"]                    s = body["s"]                    ts = body["ts"]                    sign = md5(f"XGRlBW9FXlekgbPrRHuSiA{path[1::]}{s}".encode("utf-8")).hexdigest()                    url = f"https://{host}/get-mp3/{sign}/{ts}/{path}"                    return url

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

Скачивание

На этом этапе уже немного надоедает всё это дело, поэтому берём в одну руку ранее используемое нами aiohttp, в другую aiofile и реализуем скачивание в лоб.

    async def download_file(cls, url: str, filename: str):        async with aiohttp.ClientSession() as session:            async with session.request(method="GET", url=url) as response:                data = await response.read()                async with AIOFile(filename, "wb") as afp:                    await afp.write(data)                    await afp.fsync()

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

Раскручиваем шарманку

Далее, нам нужно поставить закачку на поток. Методично перебираем исполнителей, альбомы, треки. Создаём директории, скачиваем альбомный арт, скачиваем сами треки. Прописываем тэги (почему-то у полученных mp3 они отсутствовали).

    async def download_artist(self, artist: Artist, depth: Depth = Depth.NORMAL):        artist_progress = tqdm(total=0, desc=artist.title, position=1, ascii=True)        albums = await self.api.get_artist_albums(artist.id)        artist_progress.total = len(albums)        artist_progress.refresh()        for album in albums:            album_dir = os.path.join(self.target_dir, normalize(artist.title), f"{album.year} - {normalize(album.title)}")            if depth < Depth.ALBUMS and os.path.exists(album_dir):                artist_progress.update()                continue            album_progress = tqdm(total=0, desc=f"> {album.title}", position=0, ascii=True)            tracks = await self.api.get_album_tracks(album.id)            album_progress.total = len(tracks)            album_progress.refresh()            os.makedirs(album_dir, exist_ok=True)            if album.cover:                album_progress.total += 1                cover_filename = os.path.join(album_dir, "cover.jpg")                if not os.path.exists(cover_filename):                    await self.download_file(album.cover, cover_filename)                album_progress.update()            for track in tracks:                target_filename = os.path.join(album_dir, f"{track.num:02d}. {normalize(track.title)}.mp3")                if depth >= Depth.TRACKS or not os.path.exists(target_filename):                    url = await self.api.get_track_url(track.album_id, track.id)                    await self.download_file(url, target_filename)                    self.write_tags(target_filename, {                        "title": track.title,                        "tracknumber": str(track.num),                        "artist": artist.title,                        "album": album.title,                        "date": str(album.year),                    })                album_progress.update()            album_progress.close()            artist_progress.update()        artist_progress.close()

После первой загрузки, я с удивлением обнаружил, что любимая мною группа AC/DC оказалась разделена на две директории. Это послужило поводом к реализации метода normalize:

def normalize(name: str) -> str:    return name.replace("/", "-")

Как несложно заметить, процесс скачивания можно (точнее нужно) распараллелить. Почва для этого подготовлена благодаря использованию асинхронных библиотек для сетевых запросов и записи в файлы. Для этого можно использовать asyncio.Semaphore в комбинации с asyncio.gather.
Но реализация асинхронной очереди не является предметом данной статьи, поэтому пока и так сойдёт.

Запуск

Теперь, когда у нас всё уже практически готово, нужно подумать о том, как это запускать. Явки и пароли мы будем вычитывать из файла .credentials, ненавязчиво бросая эксепшен в случае его отсутствия. Полученную куку бережно сохраним в файл .cookie для дальнейшего использования.

def resolve_cookie() -> str:    base_dir = os.path.normpath(os.path.join(os.path.dirname(__file__), os.path.pardir))    cookie_file = os.path.join(base_dir, ".cookie")    if os.path.exists(cookie_file):        with open(cookie_file, "rt") as file:            return file.read()    credentials_file = os.path.join(base_dir, ".credentials")    if os.path.exists(credentials_file):        config = configparser.ConfigParser()        config.read(credentials_file)        login = config["yandex"]["login"]        password = config["yandex"]["password"]    else:        raise Exception(f"""Create \"{credentials_file}\" with content[yandex]login=<user_login>password=<user_password>""")    cookie = auth.resolve_cookie(login, password)    with open(cookie_file, "wt") as file:        file.write(cookie)    return cookie

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

• -a (--artist), Id исполнителя, если указан, то скачиваются только треки этого исполнителя
• -o (--output), директория для хранения музыкальной коллекции, по умолчанию Music в домашней директории пользователя.
• -d (--depth), параметр родился как костыль, вызванный возможным прерыванием процесса
  
  • При значении по-умолчанию 0 (NORMAL) проверяется наличие директории с альбомом, и, если она существует, то загрузка альбома пропускается
  • Значение 1 (ALBUMS) перебирает все треки в альбоме и скачивает недостающие
  • Значение 2 (TRACKS) скачивает и перезаписывает треки, даже если они уже присутствуют в файловой системе

async def main():    parser = argparse.ArgumentParser()    parser.add_argument("-a", "--artist", help="Artist ID")    parser.add_argument("-o", "--output", default=f"{Path.home()}/Music",                        help=f"Output directory, default {Path.home()}/Music")    parser.add_argument("-d", "--depth", default=0, type=int,                        help=f"Exists files check depth, {enum_print(Depth)}")    args = parser.parse_args()    cookie = resolve_cookie()    api = YandexMusicApi(cookie)    agent = YandexMusicAgent(api, args.output)    if args.artist:        artist = await api.get_artist(args.artist)        await agent.download_artist(artist, args.depth)    else:        email = re.compile(".*?yandex_login=(.*?);.*?", re.M).match(cookie).group(1)        await agent.download_favorites(email, args.depth)

И вот, наконец-то, мы можем всё это запустить:

if __name__ == "__main__":    asyncio.run(main())

Спасибо за внимание. Теперь вы знаете, как реализовать API которого нет, скачать нескачиваемое и стать счастливым обладателем собственной музыкальной коллекции.

Результат можно увидеть в репозитории yandex.music.agent

1. Что мы будем строить?
2. Проверка инструментов
3. Структура проекта
4. Подготовка окружения
5. Логирование и конфигурация
6. Диспетчер
7. Мониторинг example.com
8. Мониторинг httpbin.org
9. Тесты
10. Заключение

• Начальные знания по asyncio
• Общее представление о принципе dependency injection

Что мы будем строить?

• Код ответа
• Количество байт в ответе
• Время, затраченное на выполнение запроса

Проверка инструментов

docker --versiondocker-compose --version

Docker version 19.03.12, build 48a66213fedocker-compose version 1.26.2, build eefe0d31

Если Docker или docker-compose не установлены, их нужно установить перед тем как продолжить. Следуйте этим руководствам:

• Установка Docker
• Установка docker-compose

Структура проекта

mkdir monitoring-daemon-tutorialcd monitoring-daemon-tutorial

./ monitoringdaemon/    __init__.py    __main__.py    containers.py config.yml docker-compose.yml Dockerfile requirements.txt

Подготовка окружения

• dependency-injector dependency injection фреймворк
• aiohttp веб фреймворк (нам нужен только http клиент)
• pyyaml библиотека для парсинга YAML файлов, используется для чтения конфига
• pytest фреймворк для тестирования
• pytest-asyncio библиотека-помогатор для тестирования asyncio приложений
• pytest-cov библиотека-помогатор для измерения покрытия кода тестами

dependency-injectoraiohttppyyamlpytestpytest-asynciopytest-cov

pip install -r requirements.txt

FROM python:3.8-busterENV PYTHONUNBUFFERED=1WORKDIR /codeCOPY . /code/RUN apt-get install openssl \ && pip install --upgrade pip \ && pip install -r requirements.txt \ && rm -rf ~/.cacheCMD ["python", "-m", "monitoringdaemon"]

version: "3.7"services:  monitor:    build: ./    image: monitoring-daemon    volumes:      - "./:/code"

docker-compose build

Successfully built 5b4ee5e76e35Successfully tagged monitoring-daemon:latest

docker-compose up

Creating network "monitoring-daemon-tutorial_default" with the default driverCreating monitoring-daemon-tutorial_monitor_1 ... doneAttaching to monitoring-daemon-tutorial_monitor_1monitoring-daemon-tutorial_monitor_1 exited with code 0

Логирование и конфигурация

"""Application containers module."""import loggingimport sysfrom dependency_injector import containers, providersclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    config = providers.Configuration()    configure_logging = providers.Callable(        logging.basicConfig,        stream=sys.stdout,        level=config.log.level,        format=config.log.format,    )

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

Сначала используем, потом задаем значения.

log:  level: "INFO"  format: "[%(asctime)s] [%(levelname)s] [%(name)s]: %(message)s"

"""Main module."""from .containers import ApplicationContainerdef main() -> None:    """Run the application."""    container = ApplicationContainer()    container.config.from_yaml('config.yml')    container.configure_logging()if __name__ == '__main__':    main()

Контейнер первый объект в приложении. Он используется для получения всех остальных объектов.

Диспетчер

./ monitoringdaemon/    __init__.py    __main__.py    containers.py    dispatcher.py    monitors.py config.yml docker-compose.yml Dockerfile requirements.txt

"""Monitors module."""import loggingclass Monitor:    def __init__(self, check_every: int) -> None:        self.check_every = check_every        self.logger = logging.getLogger(self.__class__.__name__)    async def check(self) -> None:        raise NotImplementedError()

""""Dispatcher module."""import asyncioimport loggingimport signalimport timefrom typing import Listfrom .monitors import Monitorclass Dispatcher:    def __init__(self, monitors: List[Monitor]) -> None:        self._monitors = monitors        self._monitor_tasks: List[asyncio.Task] = []        self._logger = logging.getLogger(self.__class__.__name__)        self._stopping = False    def run(self) -> None:        asyncio.run(self.start())    async def start(self) -> None:        self._logger.info('Starting up')        for monitor in self._monitors:            self._monitor_tasks.append(                asyncio.create_task(self._run_monitor(monitor)),            )        asyncio.get_event_loop().add_signal_handler(signal.SIGTERM, self.stop)        asyncio.get_event_loop().add_signal_handler(signal.SIGINT, self.stop)        await asyncio.gather(*self._monitor_tasks, return_exceptions=True)        self.stop()    def stop(self) -> None:        if self._stopping:            return        self._stopping = True        self._logger.info('Shutting down')        for task, monitor in zip(self._monitor_tasks, self._monitors):            task.cancel()        self._logger.info('Shutdown finished successfully')    @staticmethod    async def _run_monitor(monitor: Monitor) -> None:        def _until_next(last: float) -> float:            time_took = time.time() - last            return monitor.check_every - time_took        while True:            time_start = time.time()            try:                await monitor.check()            except asyncio.CancelledError:                break            except Exception:                monitor.logger.exception('Error executing monitor check')            await asyncio.sleep(_until_next(last=time_start))

"""Application containers module."""import loggingimport sysfrom dependency_injector import containers, providersfrom . import dispatcherclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    config = providers.Configuration()    configure_logging = providers.Callable(        logging.basicConfig,        stream=sys.stdout,        level=config.log.level,        format=config.log.format,    )    dispatcher = providers.Factory(        dispatcher.Dispatcher,        monitors=providers.List(            # TODO: add monitors        ),    )

Каждый компонент добавляется в контейнер.

"""Main module."""from .containers import ApplicationContainerdef main() -> None:    """Run the application."""    container = ApplicationContainer()    container.config.from_yaml('config.yml')    container.configure_logging()    dispatcher = container.dispatcher()    dispatcher.run()if __name__ == '__main__':    main()

docker-compose up

Starting monitoring-daemon-tutorial_monitor_1 ... doneAttaching to monitoring-daemon-tutorial_monitor_1monitor_1  | [2020-08-08 16:12:35,772] [INFO] [Dispatcher]: Starting upmonitor_1  | [2020-08-08 16:12:35,774] [INFO] [Dispatcher]: Shutting downmonitor_1  | [2020-08-08 16:12:35,774] [INFO] [Dispatcher]: Shutdown finished successfullymonitoring-daemon-tutorial_monitor_1 exited with code 0

Мониторинг example.com

./ monitoringdaemon/    __init__.py    __main__.py    containers.py    dispatcher.py    http.py    monitors.py config.yml docker-compose.yml Dockerfile requirements.txt

"""Http client module."""from aiohttp import ClientSession, ClientTimeout, ClientResponseclass HttpClient:    async def request(self, method: str, url: str, timeout: int) -> ClientResponse:        async with ClientSession(timeout=ClientTimeout(timeout)) as session:            async with session.request(method, url) as response:                return response

"""Application containers module."""import loggingimport sysfrom dependency_injector import containers, providersfrom . import http, dispatcherclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    config = providers.Configuration()    configure_logging = providers.Callable(        logging.basicConfig,        stream=sys.stdout,        level=config.log.level,        format=config.log.format,    )    http_client = providers.Factory(http.HttpClient)    dispatcher = providers.Factory(        dispatcher.Dispatcher,        monitors=providers.List(            # TODO: add monitors        ),    )

"""Monitors module."""import loggingimport timefrom typing import Dict, Anyfrom .http import HttpClientclass Monitor:    def __init__(self, check_every: int) -> None:        self.check_every = check_every        self.logger = logging.getLogger(self.__class__.__name__)    async def check(self) -> None:        raise NotImplementedError()class HttpMonitor(Monitor):    def __init__(            self,            http_client: HttpClient,            options: Dict[str, Any],    ) -> None:        self._client = http_client        self._method = options.pop('method')        self._url = options.pop('url')        self._timeout = options.pop('timeout')        super().__init__(check_every=options.pop('check_every'))    @property    def full_name(self) -> str:        return '{0}.{1}(url="{2}")'.format(__name__, self.__class__.__name__, self._url)    async def check(self) -> None:        time_start = time.time()        response = await self._client.request(            method=self._method,            url=self._url,            timeout=self._timeout,        )        time_end = time.time()        time_took = time_end - time_start        self.logger.info(            'Response code: %s, content length: %s, request took: %s seconds',            response.status,            response.content_length,            round(time_took, 3)        )

• Добавить фабрику example_monitor.
• Передать example_monitor в диспетчер.

"""Application containers module."""import loggingimport sysfrom dependency_injector import containers, providersfrom . import http, monitors, dispatcherclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    config = providers.Configuration()    configure_logging = providers.Callable(        logging.basicConfig,        stream=sys.stdout,        level=config.log.level,        format=config.log.format,    )    http_client = providers.Factory(http.HttpClient)    example_monitor = providers.Factory(        monitors.HttpMonitor,        http_client=http_client,        options=config.monitors.example,    )    dispatcher = providers.Factory(        dispatcher.Dispatcher,        monitors=providers.List(            example_monitor,        ),    )

log:  level: "INFO"  format: "[%(asctime)s] [%(levelname)s] [%(name)s]: %(message)s"monitors:  example:    method: "GET"    url: "http://example.com"    timeout: 5    check_every: 5

docker-compose up

Starting monitoring-daemon-tutorial_monitor_1 ... doneAttaching to monitoring-daemon-tutorial_monitor_1monitor_1  | [2020-08-08 17:06:41,965] [INFO] [Dispatcher]: Starting upmonitor_1  | [2020-08-08 17:06:42,033] [INFO] [HttpMonitor]: Checkmonitor_1  |     GET http://example.commonitor_1  |     response code: 200monitor_1  |     content length: 648monitor_1  |     request took: 0.067 secondsmonitor_1  |monitor_1  | [2020-08-08 17:06:47,040] [INFO] [HttpMonitor]: Checkmonitor_1  |     GET http://example.commonitor_1  |     response code: 200monitor_1  |     content length: 648monitor_1  |     request took: 0.073 seconds

Мониторинг httpbin.org

"""Application containers module."""import loggingimport sysfrom dependency_injector import containers, providersfrom . import http, monitors, dispatcherclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    config = providers.Configuration()    configure_logging = providers.Callable(        logging.basicConfig,        stream=sys.stdout,        level=config.log.level,        format=config.log.format,    )    http_client = providers.Factory(http.HttpClient)    example_monitor = providers.Factory(        monitors.HttpMonitor,        http_client=http_client,        options=config.monitors.example,    )    httpbin_monitor = providers.Factory(        monitors.HttpMonitor,        http_client=http_client,        options=config.monitors.httpbin,    )    dispatcher = providers.Factory(        dispatcher.Dispatcher,        monitors=providers.List(            example_monitor,            httpbin_monitor,        ),    )

log:  level: "INFO"  format: "[%(asctime)s] [%(levelname)s] [%(name)s]: %(message)s"monitors:  example:    method: "GET"    url: "http://example.com"    timeout: 5    check_every: 5  httpbin:    method: "GET"    url: "https://httpbin.org/get"    timeout: 5    check_every: 5

docker-compose up

Starting monitoring-daemon-tutorial_monitor_1 ... doneAttaching to monitoring-daemon-tutorial_monitor_1monitor_1  | [2020-08-08 18:09:08,540] [INFO] [Dispatcher]: Starting upmonitor_1  | [2020-08-08 18:09:08,618] [INFO] [HttpMonitor]: Checkmonitor_1  |     GET http://example.commonitor_1  |     response code: 200monitor_1  |     content length: 648monitor_1  |     request took: 0.077 secondsmonitor_1  |monitor_1  | [2020-08-08 18:09:08,722] [INFO] [HttpMonitor]: Checkmonitor_1  |     GET https://httpbin.org/getmonitor_1  |     response code: 200monitor_1  |     content length: 310monitor_1  |     request took: 0.18 secondsmonitor_1  |monitor_1  | [2020-08-08 18:09:13,619] [INFO] [HttpMonitor]: Checkmonitor_1  |     GET http://example.commonitor_1  |     response code: 200monitor_1  |     content length: 648monitor_1  |     request took: 0.066 secondsmonitor_1  |monitor_1  | [2020-08-08 18:09:13,681] [INFO] [HttpMonitor]: Checkmonitor_1  |     GET https://httpbin.org/getmonitor_1  |     response code: 200monitor_1  |     content length: 310monitor_1  |     request took: 0.126 seconds

Тесты

./ monitoringdaemon/    __init__.py    __main__.py    containers.py    dispatcher.py    http.py    monitors.py    tests.py config.yml docker-compose.yml Dockerfile requirements.txt

"""Tests module."""import asyncioimport dataclassesfrom unittest import mockimport pytestfrom .containers import ApplicationContainer@dataclasses.dataclassclass RequestStub:    status: int    content_length: int@pytest.fixturedef container():    container = ApplicationContainer()    container.config.from_dict({        'log': {            'level': 'INFO',            'formant': '[%(asctime)s] [%(levelname)s] [%(name)s]: %(message)s',        },        'monitors': {            'example': {                'method': 'GET',                'url': 'http://fake-example.com',                'timeout': 1,                'check_every': 1,            },            'httpbin': {                'method': 'GET',                'url': 'https://fake-httpbin.org/get',                'timeout': 1,                'check_every': 1,            },        },    })    return container@pytest.mark.asyncioasync def test_example_monitor(container, caplog):    caplog.set_level('INFO')    http_client_mock = mock.AsyncMock()    http_client_mock.request.return_value = RequestStub(        status=200,        content_length=635,    )    with container.http_client.override(http_client_mock):        example_monitor = container.example_monitor()        await example_monitor.check()    assert 'http://fake-example.com' in caplog.text    assert 'response code: 200' in caplog.text    assert 'content length: 635' in caplog.text@pytest.mark.asyncioasync def test_dispatcher(container, caplog, event_loop):    caplog.set_level('INFO')    example_monitor_mock = mock.AsyncMock()    httpbin_monitor_mock = mock.AsyncMock()    with container.example_monitor.override(example_monitor_mock), \            container.httpbin_monitor.override(httpbin_monitor_mock):        dispatcher = container.dispatcher()        event_loop.create_task(dispatcher.start())        await asyncio.sleep(0.1)        dispatcher.stop()    assert example_monitor_mock.check.called    assert httpbin_monitor_mock.check.called

docker-compose run --rm monitor py.test monitoringdaemon/tests.py --cov=monitoringdaemon

platform linux -- Python 3.8.3, pytest-6.0.1, py-1.9.0, pluggy-0.13.1rootdir: /codeplugins: asyncio-0.14.0, cov-2.10.0collected 2 itemsmonitoringdaemon/tests.py ..                                    [100%]----------- coverage: platform linux, python 3.8.3-final-0 -----------Name                             Stmts   Miss  Cover----------------------------------------------------monitoringdaemon/__init__.py         0      0   100%monitoringdaemon/__main__.py         9      9     0%monitoringdaemon/containers.py      11      0   100%monitoringdaemon/dispatcher.py      43      5    88%monitoringdaemon/http.py             6      3    50%monitoringdaemon/monitors.py        23      1    96%monitoringdaemon/tests.py           37      0   100%----------------------------------------------------TOTAL                              129     18    86%

Обратите внимание как в тесте test_example_monitor мы подменяем HttpClient моком с помощью метода .override(). Таким образом можно переопределить возвращаемое значения любого провайдера.

Такие же действия выполняются в тесте test_dispatcher для подмены моками мониторинговых задач.

Заключение

"""Application containers module."""import loggingimport sysfrom dependency_injector import containers, providersfrom . import http, monitors, dispatcherclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    config = providers.Configuration()    configure_logging = providers.Callable(        logging.basicConfig,        stream=sys.stdout,        level=config.log.level,        format=config.log.format,    )    http_client = providers.Factory(http.HttpClient)    example_monitor = providers.Factory(        monitors.HttpMonitor,        http_client=http_client,        options=config.monitors.example,    )    httpbin_monitor = providers.Factory(        monitors.HttpMonitor,        http_client=http_client,        options=config.monitors.httpbin,    )    dispatcher = providers.Factory(        dispatcher.Dispatcher,        monitors=providers.List(            example_monitor,            httpbin_monitor,        ),    )

Что дальше?

• Узнайте больше о Dependency Injector на GitHub
• Ознакомтесь с документацией на Read the Docs
• Есть вопрос или нашли баг? Откройте issue на Github

Как я дошёл до жизни такой?

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

Скажу так, проект очень интересный и вполне успешно работает в других приложениях нашей команды, да и сам автор говорит о том, что смог выкатить в прод, заюзав асинхронный пул. Но, к сожалению, мне это не очень подошло, так как обнаружилась проблема с групповым запуском задач (см. group). На момент написания статьи issue уже закрыта, однако, работа велась на протяжении месяца. В любом случае, автору удачи и всех благ, так как рабочие штуки на либе уже есть в общем, дело во мне и для меня оказался инструмент сыроват. Вдобавок, в некоторых задачах было по 2-3 http-запроса к разным сервисам, таким образом даже при оптимизации задач мы создаём 4 тысячи tcp соединений, примерно каждые 2 часа не очень Хотелось бы создавать сессию на один тип задач при запуске воркеров. Чуть подробнее о большом кол-ве запросов через aiohttp тут.

В связи с этим, я начал искать альтернативы и нашёл! Создателями celery, а конкретно, как я понял Ask Solem, была создана Faust, изначально для проекта robinhood. Faust написана под впечатлением от Kafka Streams и работает с Kafka в качестве брокера, также для хранения результатов от работы агентов используется rocksdb, а самое главное это то, что библиотека асинхронна.

Также, можете глянуть краткое сравнение celery и faust от создателей последней: их различия, различия брокеров, реализацию элементарной задачи. Всё весьма просто, однако, в faust привлекает внимание приятная особенность типизированные данные для передачи в топик.

Что будем делать?

Итак, в небольшой серии статей я покажу, как собирать данные в фоновых задачах с помощью Faust. Источником для нашего пример-проекта будет, как следует из названия, alphavantage.co. Я продемонстрирую, как писать агентов (sink, топики, партиции), как делать регулярное (cron) выполнение, удобнейшие cli-комманды faust (обёртка над click), простой кластеринг, а в конце прикрутим datadog (работающий из коробки) и попытаемся, что-нибудь увидеть. Для хранения собранных данных будем использовать mongodb и motor для подключения.

P.S. Судя по уверенности, с которой написан пункт про мониторинг, думаю, что читатель в конце последней статьи всё-таки будет выглядеть, как-то так:

Требования к проекту

В связи с тем, что я уже успел наобещать, составим небольшой списочек того, что должен уметь сервис:

1. Выгружать ценные бумаги и overview по ним (в т.ч. прибыли и убытки, баланс, cash flow за последний год) регулярно
2. Выгружать исторические данные (для каждого торгового года находить экстремумы цены закрытия торгов) регулярно
3. Выгружать последние торговые данные регулярно
4. Выгружать настроенный список индикаторов для каждой ценной бумаги регулярно

Как полагается, выбираем имя проекту с потолка: horton

Готовим инфраструктуру

Заголовок конечно сильный, однако, всё что нужно сделать это написать небольшой конфиг для docker-compose с kafka (и zookeeper в одном контейнере), kafdrop (если нам захочется посмотреть сообщения в топиках), mongodb. Получаем [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) следующего вида:

version: '3'services:  db:    container_name: horton-mongodb-local    image: mongo:4.2-bionic    command: mongod --port 20017    restart: always    ports:      - 20017:20017    environment:      - MONGO_INITDB_DATABASE=horton      - MONGO_INITDB_ROOT_USERNAME=admin      - MONGO_INITDB_ROOT_PASSWORD=admin_password  kafka-service:    container_name: horton-kafka-local    image: obsidiandynamics/kafka    restart: always    ports:      - "2181:2181"      - "9092:9092"    environment:      KAFKA_LISTENERS: "INTERNAL://:29092,EXTERNAL://:9092"      KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka-service:29092,EXTERNAL://localhost:9092"      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"      KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNAL"      KAFKA_ZOOKEEPER_SESSION_TIMEOUT: "6000"      KAFKA_RESTART_ATTEMPTS: "10"      KAFKA_RESTART_DELAY: "5"      ZOOKEEPER_AUTOPURGE_PURGE_INTERVAL: "0"  kafdrop:    container_name: horton-kafdrop-local    image: 'obsidiandynamics/kafdrop:latest'    restart: always    ports:      - '9000:9000'    environment:      KAFKA_BROKERCONNECT: kafka-service:29092    depends_on:      - kafka-service

Тут вообще ничего сложного. Для kafka объявили два listener'а: одного (internal) для использования внутри композной сети, а второго (external) для запросов из вне, поэтому пробросили его наружу. 2181 порт zookeeper'а. По остальному, я думаю, ясно.

Готовим скелет проекта

В базовом варианте структура нашего проекта должна выглядеть так:

horton docker-compose.yml horton     agents.py *     alphavantage.py *     app.py *     config.py     database      connect.py      cruds       base.py       __init__.py       security.py *      __init__.py     __init__.py     records.py *     tasks.py *

*Всё что я отметил мы пока не трогаем, а просто создаём пустые файлы.**

Создали структуру. Теперь добавим необходимые зависимости, напишем конфиг и подключение к mongodb. Полный текст файлов приводить в статье не буду, чтобы не затягивать, а сделаю ссылки на нужные версии.

Начнём с зависимостей и мета о проекте pyproject.toml

Далее, запускаем установку зависимостей и создание virtualenv (либо, можете сами создать папку venv и активировать окружение):

pip3 install poetry (если ещё не установлено)poetry install

Теперь создадим config.yml креды и куда стучаться. Сразу туда можно разместить и данные для alphavantage. Ну и переходим к config.py извлекаем данные для приложения из нашего конфига. Да, каюсь, заюзал свою либу sitri.

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

Что будет дальше?

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

Итак, а в этой самой следующей части мы:

1. Напишем небольшой клиентик для alphavantage на aiohttp с запросами на нужные нам эндпоинты.
2. Сделаем агента, который будет собирать данные о ценных бумагах и исторические цены по ним.

Код проекта

Код этой части

Как я дошёл до жизни такой?

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

Скажу так, проект очень интересный и вполне успешно работает в других приложениях нашей команды, да и сам автор говорит о том, что смог выкатить в прод, заюзав асинхронный пул. Но, к сожалению, мне это не очень подошло, так как обнаружилась проблема с групповым запуском задач (см. group). На момент написания статьи issue уже закрыта, однако, работа велась на протяжении месяца. В любом случае, автору удачи и всех благ, так как рабочие штуки на либе уже есть в общем, дело во мне и для меня оказался инструмент сыроват. Вдобавок, в некоторых задачах было по 2-3 http-запроса к разным сервисам, таким образом даже при оптимизации задач мы создаём 4 тысячи tcp соединений, примерно каждые 2 часа не очень Хотелось бы создавать сессию на один тип задач при запуске воркеров. Чуть подробнее о большом кол-ве запросов через aiohttp тут.

В связи с этим, я начал искать альтернативы и нашёл! Создателями celery, а конкретно, как я понял Ask Solem, была создана Faust, изначально для проекта robinhood. Faust написана под впечатлением от Kafka Streams и работает с Kafka в качестве брокера, также для хранения результатов от работы агентов используется rocksdb, а самое главное это то, что библиотека асинхронна.

Также, можете глянуть краткое сравнение celery и faust от создателей последней: их различия, различия брокеров, реализацию элементарной задачи. Всё весьма просто, однако, в faust привлекает внимание приятная особенность типизированные данные для передачи в топик.

Что будем делать?

Итак, в небольшой серии статей я покажу, как собирать данные в фоновых задачах с помощью Faust. Источником для нашего пример-проекта будет, как следует из названия, alphavantage.co. Я продемонстрирую, как писать агентов (sink, топики, партиции), как делать регулярное (cron) выполнение, удобнейшие cli-комманды faust (обёртка над click), простой кластеринг, а в конце прикрутим datadog (работающий из коробки) и попытаемся, что-нибудь увидеть. Для хранения собранных данных будем использовать mongodb и motor для подключения.

P.S. Судя по уверенности, с которой написан пункт про мониторинг, думаю, что читатель в конце последней статьи всё-таки будет выглядеть, как-то так:

Требования к проекту

В связи с тем, что я уже успел наобещать, составим небольшой списочек того, что должен уметь сервис:

1. Выгружать ценные бумаги и overview по ним (в т.ч. прибыли и убытки, баланс, cash flow за последний год) регулярно
2. Выгружать исторические данные (для каждого торгового года находить экстремумы цены закрытия торгов) регулярно
3. Выгружать последние торговые данные регулярно
4. Выгружать настроенный список индикаторов для каждой ценной бумаги регулярно

Как полагается, выбираем имя проекту с потолка: horton

Готовим инфраструктуру

Заголовок конечно сильный, однако, всё что нужно сделать это написать небольшой конфиг для docker-compose с kafka (и zookeeper в одном контейнере), kafdrop (если нам захочется посмотреть сообщения в топиках), mongodb. Получаем [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) следующего вида:

version: '3'services:  db:    container_name: horton-mongodb-local    image: mongo:4.2-bionic    command: mongod --port 20017    restart: always    ports:      - 20017:20017    environment:      - MONGO_INITDB_DATABASE=horton      - MONGO_INITDB_ROOT_USERNAME=admin      - MONGO_INITDB_ROOT_PASSWORD=admin_password  kafka-service:    container_name: horton-kafka-local    image: obsidiandynamics/kafka    restart: always    ports:      - "2181:2181"      - "9092:9092"    environment:      KAFKA_LISTENERS: "INTERNAL://:29092,EXTERNAL://:9092"      KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka-service:29092,EXTERNAL://localhost:9092"      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"      KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNAL"      KAFKA_ZOOKEEPER_SESSION_TIMEOUT: "6000"      KAFKA_RESTART_ATTEMPTS: "10"      KAFKA_RESTART_DELAY: "5"      ZOOKEEPER_AUTOPURGE_PURGE_INTERVAL: "0"  kafdrop:    container_name: horton-kafdrop-local    image: 'obsidiandynamics/kafdrop:latest'    restart: always    ports:      - '9000:9000'    environment:      KAFKA_BROKERCONNECT: kafka-service:29092    depends_on:      - kafka-service

Тут вообще ничего сложного. Для kafka объявили два listener'а: одного (internal) для использования внутри композной сети, а второго (external) для запросов из вне, поэтому пробросили его наружу. 2181 порт zookeeper'а. По остальному, я думаю, ясно.

Готовим скелет проекта

В базовом варианте структура нашего проекта должна выглядеть так:

horton docker-compose.yml horton     agents.py *     alphavantage.py *     app.py *     config.py     database      connect.py      cruds       base.py       __init__.py       security.py *      __init__.py     __init__.py     records.py *     tasks.py *

*Всё что я отметил мы пока не трогаем, а просто создаём пустые файлы.**

Создали структуру. Теперь добавим необходимые зависимости, напишем конфиг и подключение к mongodb. Полный текст файлов приводить в статье не буду, чтобы не затягивать, а сделаю ссылки на нужные версии.

Начнём с зависимостей и мета о проекте pyproject.toml

Далее, запускаем установку зависимостей и создание virtualenv (либо, можете сами создать папку venv и активировать окружение):

pip3 install poetry (если ещё не установлено)poetry install

Теперь создадим config.yml креды и куда стучаться. Сразу туда можно разместить и данные для alphavantage. Ну и переходим к config.py извлекаем данные для приложения из нашего конфига. Да, каюсь, заюзал свою либу sitri.

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

Что будет дальше?

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

Итак, а в этой самой следующей части мы:

1. Напишем небольшой клиентик для alphavantage на aiohttp с запросами на нужные нам эндпоинты.
2. Сделаем агента, который будет собирать данные о ценных бумагах и исторические цены по ним.

Код проекта

Код этой части

package mainimport "fmt"func function(index int) {    fmt.Println("function", index)}func main() {    for i := 0; i < 10; i++ {         go function(i)    }    fmt.Println("end")}

package mainimport "fmt"func function(index int) {    fmt.Println("function", index)}func main() {    for i := 0; i < 10; i++ {         function(i)    }    fmt.Println("end")}

def func1(a, b):    func2(a + b)    await func3(a - b)  # Ошибка, так как await может выполняться только в корутинах

class WebAgent:    def update(self, obj=None, settings=None):        ...        response = self.get_request(path=path, **settings)        ...class AsyncWebAgent:    async def update(self, obj=None, settings=None):        ...        response = await self.get_request(path=path, **settings)        ...

import aiohttpimport requestsimport pysyncasync as psa# Регистрируем функцию для синхронного запроса в google# В скобочках указываем имя для дальнейшего указания в объекте Call@psa.register("google_request")def sync_google_request(query, start):    response = requests.get(        url="http://personeltest.ru/aways/google.com/search",        params={"q": query, "start": start},    )    return response.status_code, dict(response.headers), response.text# Регистрируем корутину для асинхронного запроса в google# В скобочках указываем имя для дальнейшенго указания в объекте Call@psa.register("google_request")async def async_google_request(query, start):    params = {"q": query, "start": start}    async with aiohttps.ClientSession() as session:        async with session.get(url="http://personeltest.ru/aways/google.com/search", params=params) as response:            return response.status, dict(response.headers), await response.text()# Шаблон для получения первых 100 результатовdef google_search(query):    start = 0    while start < 100:        # В Call аргументы передавать можно как угодно, они так же и будут переданы в google_request        call = Call("google_request", query, start=start)        yield call        status, headers, text = call.result        print(status)        start += 10if __name__ == "__main__":    # Синхронный запуск кода    sync_google_search = psa.generate(google_search, psa.SYNC)    sync_google_search("Python sync")    # Асинхронный запуск кода    async_google_search = psa.generate(google_search, psa.ASYNC)    loop = asyncio.get_event_loop()    loop.run_until_complete(async_google_search("Python async"))

def _async_generate(self, template):    async def wrapper(*args, **kwargs):        ...        for call in template(*args, **kwargs):            callback = self._callbacks.get(f"{call.name}:{ASYNC}")            call.result = await callback(*call.args, **call.kwargs)        ...    return wrapper