Aiohttp

Ближе к делу. Выбираем государственную школу

Выбираем учебную программу

Выбираем онлайн школу

1. Написать в тетрадке, сфоткать и отправить учителю. Бр-р-р, не хочу заставлять учителей ломать глаза в попытках прочесть мутные фотки с мобильников, сделанные, как правило, по какому-то неписанному закону в темноте.
   
2. Отправить скан. Полумера, в общем случае невозможная из-за отсутствия у родителей нужного оборудования.
   
3. Оцифровать рукописный ввод с помощью дигитайзера или планшета. Так себе вариант, но об этом чуть позже.
   
4. Напечатать текст. В принципе, допустимо, но вот как ребёнок введёт с клавиатуры, например, математическую или химическую формулу? Никак. Плюс, для более продвинутых деток, проблема с плагиатом.
   
5. Выполнить онлайн тест. Это, безусловно, самый популярный вариант. Полагаю, большинство школ, включая РЭШ и МЭШ, ориентируются на него. На практике это означает скорее дрессировку, чем обучение. Дети учатся ставить галочки в правильном месте. За бортом остаются предметы, требующие любой формы творчества, например, сочинения, а также диктанты и непопулярное теперь по неведомой мне причине чистописание. Сюда же можно отнести умение отстаивать своё мнение.
   

// Выбрать элементelement = $(selector);element = document.querySelector(selector);// Выбрать элемент внутри элементаelement2 = element.find(selector2);element2 = element.querySelector(selector2);// Скрыть элементelement.hide();  // добавляет стиль display: noneelement.classList.add('hidden');

// Слушать событие onClickelement.click(e => { ... });element.onclick = (e) => { ...  }// Переключить классelement.toggleClass(class_name);element.classList.toggle(class_name);// Создать divdiv = $("<div>");div = document.createElement("div");// Вставить созданный div в element// (это не опечатка, можно писать одинаково)element.append(div);element.append(div);

'use strict';(function () {    const selfView = document.querySelector('#self-view'),        remoteMaster = document.querySelector('#remote-master'),        remoteSlaves = document.querySelector('#remote-slaves');    let localStream,        selfStream = null,        socket = null,        selfId = null,        connections = {};    // ***********************    // UserMedia & DOM methods    // ***********************    const init = async () => {        try {            let stream = await navigator.mediaDevices.getUserMedia({                audio: true, video: {                    width: { max: 640 }, height: { max: 480 }                }            });            localStream = stream;            selfStream = new MediaStream();            stream.getVideoTracks().forEach(track => {                selfStream.addTrack(track, stream); // track.kind == 'video'            });            selfView.querySelector('video').srcObject = selfStream;        } catch (e) {            document.querySelector('#self-view').innerHTML =                '<i>Веб камера и микрофон не найдены</i>';            console.error('Local stream not found: ', e);        }        wsInit();    }    const createRemoteView = (id, username) => {        let iDiv = document.querySelector('#pc' + id);        if (!iDiv) {            iDiv = document.createElement('div');            iDiv.className = 'remote-view';            iDiv.id = 'pc' + id;            let iVideo = document.createElement('video');            iVideo.setAttribute('autoplay', 'true');            iVideo.setAttribute('playsinline', 'true');            let iLabel = document.createElement('span');            iDiv.append(iVideo);            iDiv.append(iLabel);            if (!remoteMaster.querySelector('video')) {                remoteMaster.append(iDiv);                iLabel.textContent = 'Ведущий';            } else {                remoteSlaves.append(iDiv);                iLabel.textContent = username;            }            remoteMaster.style.removeProperty('display');        }    }    // *******************************    // Signaling (Web Socket) methods    // *******************************    const wsInit = () => {        socket = new WebSocket(SIGNALING_SERVER_URL);        socket.onopen = function (e) {            log('[socket open] Соединение установлено');        }        socket.onmessage = function (event) {            log('[socket message] Данные получены с сервера', event);            wsHandle(event.data);        }        socket.onclose = function (event) {            if (event.wasClean) {                log('[close] Соединение закрыто чисто, ' +                    `код=${event.code} причина=${event.reason}`);            } else {                log('[socket close] Соединение прервано', event);            }            clearInterval(socket.timer);        }        socket.onerror = function (error) {            logError('[socket error]', error);        }        socket.timer = setInterval(() => {            socket.send('heartbeat');        }, 10000);    }    const wsHandle = async (data) => {        if (!data) {            return;        }        try {            data = JSON.parse(data);        } catch (e) {            return;        }        switch (data.type) {            case 'handshake':                selfId = data.uid;                if (!Object.keys(data.users).length) {                    createRemoteView(selfId, 'Ведущий');                    remoteMaster.querySelector('video').srcObject =                        selfStream;                    selfView.remove();                    break;                } else {                    selfView.style.removeProperty('display');                }                for (let id in data.users) {                    await pcCreate(id, data.users[id]);                }                break;            case 'offer':                await wsHandleOffer(data);                break;            case 'answer':                await wsHandleAnswer(data)                break;            case 'candidate':                await wsHandleICECandidate(data);                break;            default:                break;        }    }    const wsHandleOffer = async (data) => {        let pc = null;        if (!connections[data.src]) {            await pcCreate(data.src, data.username);        }        pc = connections[data.src].pc;        // We need to set the remote description to the received SDP offer        // so that our local WebRTC layer knows how to talk to the caller.        let desc = new RTCSessionDescription(data.sdp);        pc.setRemoteDescription(desc).catch(error => {            logError('handleOffer', error);        });        await pc.setLocalDescription(await pc.createAnswer());        wsSend({            type: 'answer',            target: data.src,            sdp: pc.localDescription        });        connections[data.src].pc = pc; // ???    }    const wsHandleAnswer = async (data) => {        log('*** Call recipient has accepted our call, answer:', data);        let pc = connections[data.src].pc;        // Configure the remote description,        // which is the SDP payload in our 'answer' message.        let desc = new RTCSessionDescription(data.sdp);        await pc.setRemoteDescription(desc).catch((error) => {            logError('handleAnswer', error);        });    }    const wsHandleICECandidate = async (data) => {        let pc = connections[data.src].pc;        let candidate = new RTCIceCandidate(data.candidate);        log('*** Adding received ICE candidate', candidate);        pc.addIceCandidate(candidate).catch(error => {            logError('handleICECandidate', error);        });    }    const wsSend = (data) => {        if (socket.readyState !== WebSocket.OPEN) {            return;        }        socket.send(JSON.stringify(data));    }    // ***********************    // Peer Connection methods    // ***********************    const pcCreate = async (id, username) => {        if (connections[id]) {            return;        }        try {            let pc = new RTCPeerConnection(PC_CONFIG);            pc.onicecandidate = (event) =>                pcOnIceCandidate(event, id);            pc.oniceconnectionstatechange = (event) =>                pcOnIceConnectionStateChange(event, id);            pc.onsignalingstatechange =  (event) =>                pcOnSignalingStateChangeEvent(event, id);            pc.onnegotiationneeded = (event) =>                pcOnNegotiationNeeded(event, id);            pc.ontrack = (event) =>                pcOnTrack(event, id);            connections[id] = {                pc: pc,                username: username            }            if (localStream) {                try {                    localStream.getTracks().forEach(                        (track) => connections[id].pc.addTransceiver(track, {                            streams: [localStream]                        })                    );                } catch (err) {                    logError(err);                }            } else {                // Start negotiation to listen remote stream only                pcOnNegotiationNeeded(null, id);            }            createRemoteView(id, username);        } catch (error) {            logError('Peer: Connection failed', error);        }    }    const pcOnTrack = (event, id) => {        let iVideo = document.querySelector('#pc' + id + ' video');        iVideo.srcObject = event.streams[0];    }    const pcOnIceCandidate = (event, id) => {        let pc = connections[id].pc;        if (event.candidate && pc.remoteDescription) {            log('*** Outgoing ICE candidate: ' + event.candidate);            wsSend({                type: 'candidate',                target: id,                candidate: event.candidate            });        }    }    const pcOnNegotiationNeeded = async (event, id) => {        let pc = connections[id].pc;        try {            const offer = await pc.createOffer();            // If the connection hasn't yet achieved the "stable" state,            // return to the caller. Another negotiationneeded event            // will be fired when the state stabilizes.            if (pc.signalingState != 'stable') {                return;            }            // Establish the offer as the local peer's current            // description.            await pc.setLocalDescription(offer);            // Send the offer to the remote peer.            wsSend({                type: 'offer',                target: id,                sdp: pc.localDescription            });        } catch(err) {            logError('*** The following error occurred while handling' +                ' the negotiationneeded event:', err);        };    }    const pcOnIceConnectionStateChange = (event, id) => {        let pc = connections[id].pc;        switch (pc.iceConnectionState) {            case 'closed':            case 'failed':            case 'disconnected':                pcClose(id);                break;        }    }    const pcOnSignalingStateChangeEvent = (event, id) => {        let pc = connections[id].pc;        log('*** WebRTC signaling state changed to: ' + pc.signalingState);        switch (pc.signalingState) {            case 'closed':                pcClose(id);                break;        }    }    const pcClose = (id) => {        let remoteView = document.querySelector('#pc' + id);        if (connections[id]) {            let pc = connections[id].pc;            pc.close();            delete connections[id];        }        if (remoteView) {            remoteView.remove();        }    }    // *******    // Helpers    // *******    const log = (msg, data) => {        if (!data) {            data = ''        }        console.log(msg, data);    }    const logError = (msg, data) => {        if (!data) {            data = ''        }        console.error(msg, data);    }    init();})();

import jsonfrom aiohttp.web import WebSocketResponse, Responsefrom aiohttp import WSMsgTypefrom uuid import uuid1from lib.views import BaseViewclass WebSocket(BaseView):    """ Process WS connections """    async def get(self):        username = self.request['current_user'].firstname or 'Аноним'        room_id = self.request.match_info.get('room_id')        if room_id != 'test_room' and            self.request['current_user'].is_anonymous:            self.raise_error('forbidden')  # @TODO: send 4000        if (self.request.headers.get('connection', '').lower() != 'upgrade' or            self.request.headers.get('upgrade', '').lower() != 'websocket'):            return Response(text=self.request.path)  # ???        self.ws = WebSocketResponse()        await self.ws.prepare(self.request)        self.uid = str(uuid1())        if room_id not in self.request.app['web_sockets']:            self.request.app['web_sockets'][room_id] = {}        self.room = self.request.app['web_sockets'][room_id]        users = {}        for id, data in self.room.items():            users[id] = data['name']        ip = self.request.headers.get(            'X-FORWARDED-FOR',            self.request.headers.get('X-REAL-IP',            self.request.remote))        msg = {            'type': 'handshake',            'uid': str(self.uid),            'users': users, 'ip': ip}        await self.ws.send_str(json.dumps(msg, ensure_ascii=False))        self.room[self.uid] = {'name': username, 'ws': self.ws}        try:            async for msg in self.ws:                if msg.type == WSMsgType.TEXT:                    if msg.data == 'heartbeat':                        print('---heartbeat---')                        continue                    try:                        msg_data = json.loads(msg.data)                        if 'target' not in msg_data or                            msg_data['target'] not in self.room:                            continue                        msg_data['src'] = self.uid                        if 'type' in msg_data and 'target' in msg_data:                            if msg_data['type'] == 'offer':                                msg_data['username'] = username                        else:                            print('INVALID DATA:', msg_data)                    except Exception as e:                        print('INVALID JSON', e, msg)                    try:                        await self.room[msg_data['target']]['ws'].send_json(                            msg_data);                    except Exception as e:                        if 'target' in msg_data:                            self.room.pop(msg_data['target'])        finally:            self.room.pop(self.uid)        return self.ws

Тестируем интерактивную доску

Тестируем WebRTC

Медиа сервер

Выводы

В этой статье я опишу один из подходов для создания json api сервиса с валидацией данных.

Сервис будет реализован на aiohttp. Это современный, постоянно развивающийся фреймворк на языке python, использующий asyncio.

Об аннотациях:

Появление аннотаций в python позволило сделать код более понятным. Так же, аннотации открывают некоторые дополнительные возможности. Именно аннотации играют ключевую роль при валидации данных у обработчиков api-методов в этой статье.

Используемые библиотеки:

• aiohttp фреймворк для создания web-приложений
• pydantic классы, которые позволяют декларативно описывать данные и валидировать их
• valdec декоратор для валидации аргументов и возвращаемых значений у функций

Оглавление:

• 1. Файлы и папки приложения
• 2. json middlewares
  
  • 2.1. Простая middleware для json сервиса
    
    • 2.1.1. Объявление обработчика
    • 2.1.2. Класс SimpleHandler для middleware
      
      • 2.1.2.1. Метод middleware
      • 2.1.2.2. Метод для получения данных ответа с ошибкой
      • 2.1.2.3. Метод запуска обработчика
    • 2.1.3. Примеры
      
      • 2.1.3.1. Ответ с кодом 200
      • 2.1.3.2. Ответ с кодом 400
      • 2.1.3.3. Ответ с кодом 500
  • 2.2. middleware для "kwargs-обработчиков"
    
    • 2.2.1. Объявление обработчика
    • 2.2.2. Вспомогательный класс ArgumentsManager
    • 2.2.3. Класс KwargsHandler для middleware
      
      • 2.2.3.1. Метод запуска обработчика
      • 2.2.3.2. Метод формирования словаря с именами аргументов и их значениями
    • 2.2.4. Примеры
      
      • 2.2.4.1. Метод /create
      • 2.2.4.2. Метод /read
      • 2.2.4.3. Метод /info/{info_id}
  • 2.3. middleware c оболочками запроса/ответа и валидацией
    
    • 2.3.1. Класс данных pydantic.BaseModel
      
      • 2.3.1.1. Строгие типы
      • 2.3.1.2. Строгая сигнатура при создании экземпляра
    • 2.3.2. Декоратор valdec.validate
      
      • 2.3.2.1. Функции-валидаторы
      • 2.3.2.2. Настройка декоратора
      • 2.3.2.3. Еще раз про приведение типов
      • 2.3.2.4. Исключения
    • 2.3.3. Базовый класс данных
    • 2.3.4. Объявление обработчика
    • 2.3.5. Классы данных для оболочек
    • 2.3.6. Класс WrapsKwargsHandler для middleware
      
      • 2.3.6.1. Метод запуска обработчика
      • 2.3.6.2. Метод для получения данных ответа с ошибкой
    • 2.3.7. Примеры
      
      • 2.3.7.1. Метод /create
      • 2.3.7.2. Метод /read
      • 2.3.7.3. Метод /info/{info_id}
• 3. О нереализованной документации
• 4. Заключение

1. Файлы и папки приложения

- sources - Папка с кодом приложения    - data_classes - Папка с модулями классов данных        - base.py - базовый класс данных        - person.py - классы данных о персоне        - wraps.py - классы данных оболочек для запросов/ответов    - handlers - Папка с модулями обработчиков запросов        - kwargs.py - обработчики для примера работы с `KwargsHandler.middleware`        - simple.py - обработчики для примера работы с `SimpleHandler.middleware`        - wraps.py - обработчики для примера работы с `WrapsKwargsHandler.middleware`    - middlewares - Папка с модулями для middlewares        - exceptions.py - классы исключений        - kwargs_handler.py - класс `KwargsHandler`        - simple_handler.py - класс `SimpleHandler`        - utils.py - вспомогательные классы и функции для middlewares        - wraps_handler.py - класс `WrapsKwargsHandler`    - requirements.txt - зависимости приложения    - run_kwargs.py - запуск с `KwargsHandler.middleware`    - run_simple.py - запуск c `SimpleHandler.middleware`    - run_wraps.py - запуск c `WrapsKwargsHandler.middleware`    - settings.py - константы с настройками приложения- Dockerfile - докерфайл для сборки образа

Код доступен на гитхаб: https://github.com/EvgeniyBurdin/api_service

2. json middlewares

middleware в aiohttp.web.Application() является оболочкой для обработчиков запросов.

Если в приложении используется middleware, то поступивший запрос сначала попадает в неё, и только потом передается в обработчик. Обработчик формирует и отдает ответ. Этот ответ снова сначала попадает в middleware и уже она отдает его наружу.

Если в приложении используются нескольно middleware, то каждая из них добавляет новый уровень вложенности.

Между middleware и обработчиком не обязательно должны передаваться "запрос" и "ответ" в виде web.Request и web.Response. Допускается передавать любые данные.

Таким образом, в middleware можно выделить действия над запросами/ответами, которые будут одинаковыми для всех обработчиков.

Это довольно упрощенное описание, но достаточное для понимания того что будет дальше.

2.1. Простая middleware для json сервиса

Обычно, объявление обработчика запроса в приложении aiohttp.web.Application() выглядит, примерно, так:

from aiohttp import webasync def some_handler(request: web.Request) -> web.Response:    data = await request.json()    ...    text = json.dumps(some_data)    ...    return web.Response(text=text, ...)

Для доступа к данным обработчику необходимо "вытащить" из web.Request объект, который был передал в json. Обработать его, сформировать объект с данными для ответа. Закодировать ответ в строку json и отдать "наружу" web.Response (можно отдать и сразу web.json_response()).

2.1.1. Объявление обработчика

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

from aiohttp import webasync def some_handler(request: web.Request, data: Any) -> Any:    ...    return some_data

Каждый из обработчиков имеет два позиционных аргумента. В первый будет передан оригинальный экземпляр web.Request (на всякий случай), во второй уже готовый объект python, с полученными данными.

В примере, второй аргумент имеет такое объявление: data: Any. Имя у него может быть любым (как и у первого аргумента), а вот в аннотации лучше сразу указать тип объекта, который "ждет" обработчик. Это пожелание справедливо и для возврата.

То есть, в реальном коде, объявление обработчика может быть таким:

from aiohttp import webfrom typing import Union, Listasync def some_handler(    request: web.Request, data: Union[str, List[str]]) -> List[int]:    ...    return some_data

2.1.2. Класс SimpleHandler для middleware

Класс SimpleHandler реализует метод для самой middleware и методы, которые впоследствии помогут изменять/дополнять логику работы middleware (ссылка на код класса).

Остановлюсь подробнее только на некоторых.

2.1.2.1. Метод middleware

    @web.middleware    async def middleware(self, request: web.Request, handler: Callable):        """ middleware для json-сервиса.        """        if not self.is_json_service_handler(request, handler):            return await handler(request)        try:            request_body = await self.get_request_body(request, handler)        except Exception as error:            response_body = self.get_error_body(request, error)            status = 400        else:            # Запуск обработчика            response_body, status = await self.get_response_body_and_status(                request, handler, request_body            )        finally:            # Самостоятельно делаем дамп объекта python (который находится в            # response_body) в строку json.            text, status = await self.get_response_text_and_status(                request, response_body, status            )        return web.Response(            text=text, status=status, content_type="application/json",        )

Именно этот метод надо будет добавить в список middlewares в процессе создания приложения.

Например, так:

    ...    app = web.Application()    service_handler = SimpleHandler()    app.middlewares.append(service_handler.middleware)    ...

2.1.2.2. Метод для получения данных ответа с ошибкой

Так как у нас json сервис, то, желательно, чтобы ошибки во входящих данных (с кодом 400), и внутренние ошибки сервиса (с кодом 500), отдавались в формате json.

Для этого создан метод формирования "тела" для ответа с ошибкой:

    def get_error_body(self, request: web.Request, error: Exception) -> dict:        """ Отдает словарь с телом ответа с ошибкой.        """        return {"error_type": str(type(error)), "error_message": str(error)}

Хочу обратить внимание на то, что этот метод должен отработать без исключений и вернуть объект с описанием ошибки, который можно кодировать в json. Если работа этого метода завершиться исключением, то мы не увидим json в теле ответа.

2.1.2.3. Метод запуска обработчика

В текущем классе он очень простой:

    async def run_handler(        self, request: web.Request, handler: Callable, request_body: Any    ) -> Any:        """ Запускает реальный обработчик, и возвращает результат его работы.        """        return await handler(request, request_body)

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

2.1.3. Примеры

Имеется такой обработчик:

async def some_handler(request: web.Request, data: dict) -> dict:    return data

Будем посылать запросы на url этого обработчика.

{    "name": "test",    "age": 25}

{    "name": "test",    "age": 25}

{    "name": "test", 111111111111    "age": 25}

{    "error_type": "<class 'json.decoder.JSONDecodeError'>",    "error_message": "Expecting property name enclosed in double quotes: line 2 column 21 (char 22)"}

async def handler500(request: web.Request, data: dict) -> dict:    raise Exception("Пример ошибки 500")    return data

{    "name": "test",    "age": 25}

{    "error_type": "<class 'Exception'>",    "error_message": "Пример ошибки 500"}

2.2. middleware для "kwargs-обработчиков"

middleware из предыдущего раздела уже можно успешно использовать.

Но проблема дублирования кода в обработчиках не решена до конца.

Рассмотрим такой пример:

async def some_handler(request: web.Request, data: dict) -> dict:    storage = request.app["storage"]    logger = request.app["logger"]    user_id = request.match_info["user_id"]    # и т.д. и т.п...    return data

Так как storage, или logger (или что-то еще), могут быть нужны и в других обработчиках, то везде придется "доставать" их одинаковым образом.

2.2.1. Объявление обработчика

Хотелось бы, чтобы обработчики объявлялись, например, так:

async def some_handler_1(data: dict) -> int:    # ...    return some_dataasync def some_handler_2(storage: StorageClass, data: List[int]) -> dict:    # ...    return some_dataasync def some_handler_3(    data: Union[dict, List[str]], logger: LoggerClass, request: web.Request) -> str:    # ...    return some_data

То есть, чтобы нужные для обработчика сущности объявлялись в его сигнатуре и сразу были бы доступны в коде.

2.2.2. Вспомогательный класс ArgumentsManager

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

За регистрацию, хранение и "выдачу" таких сущностей отвечает класс ArgumentsManager. Он объявлен в модуле middlewares/utils.py (ссылка на код класса).

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

Звучит немного запутано, но на самом деле всё просто:

@dataclassclass RawDataForArgument:    request: web.Request    request_body: Any    arg_name: Optional[str] = Noneclass ArgumentsManager:    """ Менеджер для аргументов обработчика.        Связывает имя аргумента с действием, которое надо совершить для        получения значения аргумента.    """    def __init__(self) -> None:        self.getters: Dict[str, Callable] = {}    # Тело json запроса ------------------------------------------------------    def reg_request_body(self, arg_name) -> None:        """ Регистрация имени аргумента для тела запроса.        """        self.getters[arg_name] = self.get_request_body    def get_request_body(self, raw_data: RawDataForArgument):        return raw_data.request_body    # Ключи в request --------------------------------------------------------    def reg_request_key(self, arg_name) -> None:        """ Регистрация имени аргумента который хранится в request.        """        self.getters[arg_name] = self.get_request_key    def get_request_key(self, raw_data: RawDataForArgument):        return raw_data.request[raw_data.arg_name]    # Ключи в request.app ----------------------------------------------------    def reg_app_key(self, arg_name) -> None:        """ Регистрация имени аргумента который хранится в app.        """        self.getters[arg_name] = self.get_app_key    def get_app_key(self, raw_data: RawDataForArgument):        return raw_data.request.app[raw_data.arg_name]    # Параметры запроса ------------------------------------------------------    def reg_match_info_key(self, arg_name) -> None:        """ Регистрация имени аргумента который приходит в параметрах запроса.        """        self.getters[arg_name] = self.get_match_info_key    def get_match_info_key(self, raw_data: RawDataForArgument):        return raw_data.request.match_info[raw_data.arg_name]    # Можно добавить и другие регистраторы...

Регистрация имен аргументов выполняется при создании экземпляра web.Application():

# ...app = web.Application()arguments_manager = ArgumentsManager()# Регистрация имени аргумента обработчика, в который будут передаваться# данные полученные из json-тела запросаarguments_manager.reg_request_body("data")# Регистрация имени аргумента обработчика, в который будет передаваться# одноименный параметр запроса из словаря request.match_infoarguments_manager.reg_match_info_key("info_id")# В приложении будем использовать хранилище# (класс хранилища "взят с потолка" и здесь просто для примера)app["storage"] = SomeStorageClass(login="user", password="123")# Регистрация имени аргумента обработчика, в который будет передаваться# экземпляр хранилищаarguments_manager.reg_app_key("storage")# ...

Теперь экземпляр ArgumentsManager хранит информацию о возможных аргументах обработчиков. Он передается при создании экземпляра класса для middleware:

...service_handler = KwargsHandler(arguments_manager=arguments_manager)app.middlewares.append(service_handler.middleware)...

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

2.2.3. Класс KwargsHandler для middleware

Класс KwargsHandler является наследником SimpleHandler и расширяет его возможности тем, что позволяет создавать обработчики согласно требованию п.2.2.1.

В этом классе переопределяется один метод run_handler, и добавляется еще два make_handler_kwargs и build_error_message_for_invalid_handler_argument (ссылка на код класса).

2.2.3.1. Метод запуска обработчика

Переопределяется метод родительского класса:

    async def run_handler(        self, request: web.Request, handler: Callable, request_body: Any    ) -> Any:        """ Запускает реальный обработчик, и возвращает результат его работы.            (Этот метод надо переопределять, если необходима дополнительная            обработка запроса/ответа/исключений)        """        kwargs = self.make_handler_kwargs(request, handler, request_body)        return await handler(**kwargs)

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

2.2.3.2. Метод формирования словаря с именами аргументов и их значениями

Метод make_handler_kwargs был добавлен в текущий класс. Он реализует заполнение словаря с именами аргументов и их значениями, который будет потом использован при вызове обработчика. Заполнение словаря происходит при помощи уже подготовленного экземпляра ArgumentsManager.

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

Но у этого требования есть одно исключение. А именно, аргумент с экземпляром web.Request может иметь в сигнатуре обработчика любое имя, но он обязательно должен иметь аннотацию типом web.Request (например, r: web.Request или req: web.Request или request: web.Request). То есть, экземпляр web.Request "зарегистрирован" по умолчанию, и может быть использован в любом обработчике.

И еще одно замечание: все аргументы обработчика должны иметь аннотацию.

Метод build_error_message_for_invalid_handler_argument просто формирует строку с сообщением об ошибке. Он создан для возможности изменить сообщение на свой вкус.

2.2.4. Примеры

Сигнатуры методов такие:

async def create(    data: Union[dict, List[dict]], storage: dict,) -> Union[dict, List[dict]]:    # ...async def read(storage: dict, data: str) -> dict:    # ...async def info(info_id: int, request: web.Request) -> str:    # ...

Первые два обслуживают POST запросы, последний GET (просто, для примера)

[    {        "name": "Ivan"    },    {        "name": "Oleg"    }]

[    {        "id": "5730bab1-9c1b-4b01-9979-9ad640ea5fc1",        "name": "Ivan"    },    {        "id": "976d821a-e871-41b4-b5a2-2875795d6166",        "name": "Oleg"    }]

"5730bab1-9c1b-4b01-9979-9ad640ea5fc1"

{    "id": "5730bab1-9c1b-4b01-9979-9ad640ea5fc1",    "name": "Ivan"}

"any json"

"info_id=123 and request=<Request GET /info/123 >"

2.3. middleware c оболочками запроса/ответа и валидацией

Иногда, требования для api-сервиса включают в себя стандартизированные оболочки для запросов и ответов.

Например, тело запроса к методу create может быть таким:

{    "data": [        {            "name": "Ivan"        },        {            "name": "Oleg"        }    ],    "id": 11}

а ответ таким:

{    "success": true,    "result": [        {            "id": "9738d8b8-69da-40b2-8811-b33652f92f1d",            "name": "Ivan"        },        {            "id": "df0fdd43-4adc-43cd-ac17-66534529d440",            "name": "Oleg"        }    ],    "id": 11}

То есть, данные для запроса в ключе data а от ответа в result.

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

Ключ ответа success является признаком успешности запроса.

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

Запрос к методу read:

{    "data":  "ddb0f2b1-0179-44b7-b94d-eb2f3b69292d",    "id": 3}

Ответ:

{    "success": false,    "result": {        "error_type": "<class 'handlers.PersonNotFound'>",        "error_message": "Person whith id=ddb0f2b1-0179-44b7-b94d-eb2f3b69292d not found!"    },    "id": 3}

Уже представленные классы для json middleware позволяют добавить логику работы с оболочками в новый класс для middleware. Надо будет дополнить метод run_handler, и заменить (или дополнить) метод get_error_body.

Таким образом, в обработчики будут "прилетать" только данные, необходимые для их работы (в примере это значение ключа data). Из обработчиков будет возвращаться только положительный результат (значение ключа result). А исключения будет обрабатывать middleware.

Так же, если это необходимо, можно добавить и валидацию данных.

Чтобы "два раза не вставать", я сразу покажу как добавить и оболочки и валидацию. Но сначала необходимо сделать некоторые пояснения по выбранным инструментам.

2.3.1. Класс данных pydantic.BaseModel

pydantic.BaseModel позволяет декларативно объявлять данные.

При создании экземпляра происходит валидация данных по их аннотациям (и не только). Если валидация провалилась поднимается исключение.

Небольшой пример:

from pydantic import BaseModelfrom typing import Union, Listclass Info(BaseModel):    foo: intclass Person(BaseModel):    name: str    info: Union[Info, List[Info]]kwargs = {"name": "Ivan", "info": {"foo": 0}}person = Person(**kwargs)assert person.info.foo == 0kwargs = {"name": "Ivan", "info": [{"foo": 0}, {"foo": 1}]}person = Person(**kwargs)assert person.info[1].foo == 1kwargs = {"name": "Ivan", "info": {"foo": "bar"}}  # <- Ошибка, str не intperson = Person(**kwargs)# Возникло исключение:# ...# pydantic.error_wrappers.ValidationError: 2 validation errors for Person# info -> foo#  value is not a valid integer (type=type_error.integer)# info#  value is not a valid list (type=type_error.list)

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

В аннотациях к полям мы можем использовать алиасы из typing.

Если в аннотации к полю присутствует класс-потомок pydantic.BaseModel, то данные "маппятся" и в него (и так с любой вложенностью хотя, на счет "любой" не проверял).

Провал валидации сопровождается довольно информативным сообщением об ошибке. В примере мы видим, что на самом деле было две ошибки: info.foo не int, и info не list, что соответствует аннотации и сопоставленному с ней значению.

При использовании pydantic.BaseModel есть нюансы, на которые я хочу обратить внимание.

2.3.1.1. Строгие типы

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

kwargs = {"name": "Ivan", "info": {"foo": "0"}}person = Person(**kwargs)assert person.info.foo == 0

То есть, имеем неявное приведение типов. И такое встречается не только с str->int (более подробно про типы pydantic см. в документации).

Приведение типов, в определенных ситуациях, может оказаться полезным, например строка с UUID -> UUID. Но, если приведение некоторых типов недопустимо, то в аннотациях надо использовать типы, наименование у которых начинается со Strict.... Например, pydantic.StrictInt, pydantic.StrictStr, и т.п...

2.3.1.2. Строгая сигнатура при создании экземпляра

Если, для определенных выше классов, попробовать выполнить такой пример:

kwargs = {"name": "Ivan", "info": {"foo": 0}, "bar": "BAR"}person = Person(**kwargs)

То создание экземпляра пройдет без ошибок.

Это тоже может оказаться не тем, что ожидаешь по умолчанию.

Для строгой проверки аргументов, при создании экземпляра, необходимо переопределить базовый класс:

from pydantic import BaseModel, Extra, StrictInt, StrictStrfrom typing import Union, Listclass BaseApi(BaseModel):    class Config:        # Следует ли игнорировать (ignore), разрешать (allow) или        # запрещать (forbid) дополнительные атрибуты во время инициализации        # модели, подробнее:        # https://pydantic-docs.helpmanual.io/usage/model_config/        extra = Extra.forbidclass Info(BaseApi):    foo: StrictIntclass Person(BaseApi):    name: StrictStr    info: Union[Info, List[Info]]kwargs = {"name": "Ivan", "info": {"foo": 0}, "bar": "BAR"}person = Person(**kwargs)# ...# pydantic.error_wrappers.ValidationError: 1 validation error for Person# bar#   extra fields not permitted (type=value_error.extra)

Теперь все нормально, валидация провалилась.

2.3.2. Декоратор valdec.validate

Декоратор valdec.validate позволяет валидировать аргументы и/или возвращаемое значение функции или метода.

Можно валидировать только те аргументы, для которых указана аннотация.

Если у возврата нет аннотации, то считается что функция должна вернуть None (имеет аннотацию -> None:).

Определен декоратор как для обычных функций/методов:

from valdec.decorators import validate@validate  # Валидируем все аргументы с аннотациями, и возвратdef foo(i: int, s: str) -> int:    return i@validate("i", "s")  # Валидируем только "i" и "s"def bar(i: int, s: str) -> int:    return i

так и для асинхронных.

# Импортируем асинхронный вариантfrom valdec.decorators import async_validate as validate@validate("s", "return", exclude=True)  # Валидируем только "i"async def foo(i: int, s: str) -> int:    return int(i)@validate("return")  # Валидируем только возвратasync def bar(i: int, s: str) -> int:    return int(i)

2.3.2.1. Функции-валидаторы

Декоратор получает данные об аргументах/возврате функции, и передает их в функцию-валидатор (это сильно упрощенно, но по сути так), которая и производит, непосредственно, валидацию.

Сигнатура функции-валидатора:

def validator(    annotations: Dict[str, Any],    values: Dict[str, Any],    is_replace: bool,    extra: dict) -> Optional[Dict[str, Any]]:

Аргументы:

• annotations Словарь, который содержит имена аргументов и их аннотации.
• values Словарь, который содержит имена аргументов и их значения.
• is_replace управляет тем, что возвращает функция-валидатор, а именно возвращать отвалидированные значения или нет.
  
  • Если True, то функция должна вернуть словарь с именами отвалидированных аргументов и их значениями после валидации. Таким образом, например, если у аргумента была аннотация с наследником BaseModel и данные для него поступили в виде словаря, то они будут заменены на экземпляр BaseModel, и в декорируемой функции к ним можно будет обращаться "через точку".
  • Если параметр равен False, то функция вернет None, а декорируемая функция получит оригинальные данные (то есть, например, словарь так и останется словарем, а не станет экземпляром BaseModel).
• extra Словарь с дополнительными параметрами.

По умолчанию, в декораторе validate используется функция-валидатор на основе pydantic.BaseModel.

В ней происходит следующее:

• На основании словаря с именами аргументов и их аннотаций создается класс данных (потомок pydantic.BaseModel)
• Создается экземпляр этого класса в который передается словарь с именами и значениями. В этот момент и происходит валидация.
• Возвращает функция аргументы после валидации (которые уже буду содержать значения из созданного экземпляра), или ничего не возвращает, зависит от аргумента is_replace.

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

Функция-валидатор может быть реализована на основе любого валидирующего класса (в репозитарии valdec есть пример реализации на ValidatedDC). Но необходимо учесть следующее: далее в статье, я буду использовать потомков pydantic.BaseModel в аннотациях аргументов у обработчиков. Соответственно, при другом валидирующем классе, в аннотациях необходимо будет указывать потомков этого "другого" класса.

2.3.2.2. Настройка декоратора

По умолчанию, декоратор "подменяет" исходные данные на данные экземпляра валидирующего класса:

from typing import List, Optionalfrom pydantic import BaseModel, StrictInt, StrictStrfrom valdec.decorators import validateclass Profile(BaseModel):    age: StrictInt    city: StrictStrclass Student(BaseModel):    name: StrictStr    profile: Profile@validate("group")def func(group: Optional[List[Student]] = None):    for student in group:        assert isinstance(student, Student)        assert isinstance(student.name, str)        assert isinstance(student.profile.age, int)data = [    {"name": "Peter", "profile": {"age": 22, "city": "Samara"}},    {"name": "Elena", "profile": {"age": 20, "city": "Kazan"}},]func(data)

Обратите внимание на assert'ы.

Это работает и для возврата:

@validate  # Валидируем всёdef func(group: Optional[List[Student]] = None, i: int) -> List[Student]:    #...    return [        {"name": "Peter", "profile": {"age": 22, "city": "Samara"}},        {"name": "Elena", "profile": {"age": 20, "city": "Kazan"}},    ]

Здесь, несмотря на то, что в return явно указан список словарей, функция вернет список экземпляров Student (подмену выполнит декоратор).

Но Нам не всегда надо именно такое поведение. Иногда, бывает полезно отвалидировать, а данные не подменять (например, если речь о входящих данных, чтобы сразу отдать их в БД). И этого можно добиться изменив настройки декоратора:

from valdec.data_classes import Settingsfrom valdec.decorators import validate as _validatefrom valdec.validator_pydantic import validatorcustom_settings = Settings(    validator=validator,     # Функция-валидатор.    is_replace_args=False,   # Делать ли подмену в аргументах    is_replace_result=False, # Делать ли подмену в результате    extra={}                 # Дополнительные параметры, которые будут                             # передаваться в функцию-валидатор)# Определяем новый декораторdef validate_without_replacement(*args, **kwargs):    kwargs["settings"] = custom_settings    return _validate(*args, **kwargs)# Используем@validate_without_replacementdef func(group: Optional[List[Student]] = None, i: int) -> List[Student]:    #...    return [        {"name": "Peter", "profile": {"age": 22, "city": "Samara"}},        {"name": "Elena", "profile": {"age": 20, "city": "Kazan"}},    ]

И теперь func вернет список словарей, так как is_replace_result=False. И получит тоже список словарей, так как is_replace_args=False.

Но сама валидация данных будет работать как и раньше, не будет лишь подмены.

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

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

Как можно заметить, в настройках указывается и функция-валидатор, и если вы захотите использовать свою именно там нужно ее подставить.

2.3.2.3. Еще раз про приведение типов

Рассмотрим такой пример применения декоратора:

from valdec.decorators import validate@validatedef foo(i: int):    assert isinstance(i, int)foo("1")

Мы вызываем функцию и передаем ей строку. Но валидация прошла успешно, и в функцию прилетело целое.

Как я уже говорил, по умолчанию, в декораторе validate, используется функция-валидатор на основе pydantic.BaseModel. В п.2.3.1.1. можно еще раз почитать про неявное приведение типов в этом классе.

В нашем же примере, для того чтобы получить желаемое поведение (ошибку валидации), необходимо сделать так:

from valdec.decorators import validatefrom pydantic import StrictInt@validatedef foo(i: StrictInt):    passfoo("1")# ...# valdec.errors.ValidationArgumentsError: Validation error# <class 'valdec.errors.ValidationError'>: 1 validation error for# argument with the name of:# i#  value is not a valid integer (type=type_error.integer).

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

Не забывайте про это.

2.3.2.4. Исключения

• valdec.errors.ValidationArgumentsError "поднимается" если валидация аргументов функции потерпела неудачу
• valdec.errors.ValidationReturnError если не прошел валидацию возврат

Само сообщение с описанием ошибки берется из валидирующего класса. В нашем примере это сообщение об ошибке от pydantic.BaseModel.

2.3.3. Базовый класс данных

Как я уже говорил, в этой статье используем классы-наследники от pydantic.BaseModel.

Cначала обязательно определим базовый класс данных:

data_classes/base.py

from pydantic import BaseModel, Extraclass BaseApi(BaseModel):    """ Базовый класс данных для api.    """    class Config:        extra = Extra.forbid

2.3.4. Объявление обработчика

Класс для middleware, над созданием которого мы сейчас работаем, позволит объявлять обработчики, например, так:

from typing import List, Unionfrom valdec.decorators import async_validate as validatefrom data_classes.person import PersonCreate, PersonInfo@validate("data", "return")async def create(    data: Union[PersonCreate, List[PersonCreate]], storage: dict,) -> Union[PersonInfo, List[PersonInfo]]:    # ...    return result

Что здесь добавилось (по сравнению с обработчиками из прошлых глав):

• декоратор validate валидирует поступившие данные и ответ, и "подменяет" их на экземпляры валидирующих классов
• в аннотациях у данных указаны уже конкретные классы.

Про оболочки запросов/ответов обработчик ничего не знает, ему это и не надо.

Позволю себе небольшую ремарку: если в аргументах у обработчика нет экземпляра web.Request(), то он является обычной функцией, которую можно использовать не только в wep.Aplication(). На самом деле, даже если он там есть, эту функцию по-прежнему можно будет использовать в приложениях другого типа, если обеспечить совместимый с web.Request() экземпляр данных.

Соответственно, классы данных для этого обработчика могут быть такими:

data_classes/person.py

from uuid import UUIDfrom pydantic import Field, StrictStrfrom data_classes.base import BaseApiclass PersonCreate(BaseApi):    """ Данные для создания персоны.    """    name: StrictStr = Field(description="Имя.", example="Oleg")class PersonInfo(BaseApi):    """ Информация о персоне.    """    id: UUID = Field(description="Идентификатор.")    name: StrictStr = Field(description="Имя.")

2.3.5. Классы данных для оболочек

В самом начале п.2.3. были обозначены тебования к оболочкам запроса и ответа.

Для их выполнения создадим классы данных.

data_classes/wraps.py

from typing import Any, Optionalfrom pydantic import Field, StrictIntfrom data_classes.base import BaseApi_ID_DESCRIPTION = "Идентификатор запроса к сервису."class WrapRequest(BaseApi):    """ Запрос.    """    data: Any = Field(description="Параметры запроса.", default=None)    id: Optional[StrictInt] = Field(description=_ID_DESCRIPTION)class WrapResponse(BaseApi):    """ Ответ.    """    success: bool = Field(description="Статус ответа.", default=True)    result: Any = Field(description="Результат ответа.")    id: Optional[StrictInt] = Field(description=_ID_DESCRIPTION)

Эти классы будут использоваться в классе для middleware при реализации логики оболочек.

2.3.6. Класс WrapsKwargsHandler для middleware

Класс WrapsKwargsHandler является наследником KwargsHandler и расширяет его возможности тем, что позволяет использовать оболочки для данных запросов и ответов и их валидацию (ссылка на код класса).

В этом классе переопределяются два метода run_handler и get_error_body.

2.3.6.1. Метод запуска обработчика

Переопределяется метод родительского класса:

async def run_handler(        self, request: web.Request, handler: Callable, request_body: Any    ) -> dict:        id_ = None        try:            # Проведем валидацию оболочки запроса            wrap_request = WrapRequest(**request_body)        except Exception as error:            message = f"{type(error).__name__} - {error}"            raise InputDataValidationError(message)        # Запомним поле id для ответов        id_ = wrap_request.id        request[KEY_NAME_FOR_ID] = id_        try:            result = await super().run_handler(                request, handler, wrap_request.data            )        except ValidationArgumentsError as error:            message = f"{type(error).__name__} - {error}"            raise InputDataValidationError(message)        # Проведем валидацию оболочки ответа        wrap_response = WrapResponse(success=True, result=result, id=id_)        return wrap_response.dict()

Сначала мы проверим оболочку запроса. Исключение InputDataValidationError поднимется в следующих случаях:

• если в теле запроса не словарь (пусть даже пустой)
• если есть поля с ключами отличными от data и id
• если есть ключ id но его значение не StrictInt и не None

Если в запросе нет ключа id, то wrap_request.id получит значение None. Ключ data может иметь любое значение и валидироваться не будет. Так же, его может вообще не быть во входящих данных, тогда wrap_request.data получит значение None.

Затем мы запоминаем wrap_request.id в request. Это необходимо для формирования ответа с ошибкой на текущий запрос (если она произойдет).

После этого вызывается обработчик, но для его входящих данных передается только wrap_request.data (напомню, что во wrap_request.data сейчас объект python в том виде, как он был получен из json). При этом, исключение InputDataValidationError поднимается если получено исключение valdec.errors.ValidationArgumentsError.

Если обработчик отработал нормально, и был получен результат его работы, то создаем экземпляр класса оболочки ответа WrapResponse в варианте для успешного ответа.

Все просто, но хотел бы обратить внимание на такой момент. Можно было бы обойтись без создания wrap_response, а сразу сформировать словарь (как это и будет сделано для ответа с ошибкой). Но, в случае успешного ответа мы не знаем что пришло в ответе от обработчика, это может быть, например, как список словарей, так и список экземпляров BaseApi. А на выходе из метода мы должны гарантированно отдать объект, готовый для кодирования в json. Поэтому, мы "заворачиваем" любые данные с результом во WrapResponse.result и уже из wrap_response получаем окончательный ответ для метода при помощи wrap_response.dict() (ссылка на документацию).

2.3.6.2. Метод для получения данных ответа с ошибкой

Заменяется метод родительского класса:

def get_error_body(self, request: web.Request, error: Exception) -> dict:        """ Формирует и отдает словарь с телом ответа с ошибкой.        """        result = dict(error_type=str(type(error)), error_message=str(error))        # Так как мы знаем какая у нас оболочка ответа, сразу сделаем словарь        # с аналогичной "схемой"        response = dict(            # Для поля id используется сохраненное в request значение.            success=False, result=result, id=request.get(KEY_NAME_FOR_ID)        )        return response

Здесь можно было бы применить и наследование (вызвать super() для получения result), но для наглядности я оставил так. Вы можете сделать как сочтете нужным.

2.3.7. Примеры

Сигнатуры методов такие:

@validate("data", "return")async def create(    data: Union[PersonCreate, List[PersonCreate]], storage: dict,) -> Union[PersonInfo, List[PersonInfo]]:    # ...@validate("data", "return")async def read(storage: dict, req: web.Request, data: UUID) -> PersonInfo:    # ...@validate("info_id")async def info(info_id: int, request: web.Request) -> Any:    return f"info_id={info_id} and request={request}"

Первые два обслуживают POST запросы, последний GET (просто, для примера)

{    "data": [        {            "name": "Ivan"        },        {            "name": "Oleg"        }    ],    "id": 1}

{    "success": true,    "result": [        {            "id": "af908a90-9157-4231-89f6-560eb6a8c4c0",            "name": "Ivan"        },        {            "id": "f7d554a0-1be9-4a65-bfc2-b89dbf70bb3c",            "name": "Oleg"        }    ],    "id": 1}

{    "data": {        "name": "Eliza"    },    "id": 2}

{    "success": true,    "result": {        "id": "f3a45a19-acd0-4939-8e0c-e10743ff8e55",        "name": "Eliza"    },    "id": 2}

{    "data": 123,    "id": 3}

{    "success": false,    "result": {        "error_type": "<class 'middlewares.exceptions.InputDataValidationError'>",        "error_message": "ValidationArgumentsError - Validation error <class 'valdec.errors.ValidationError'>: 2 validation errors for argument with the name of:\ndata\n  value is not a valid dict (type=type_error.dict)\ndata\n  value is not a valid list (type=type_error.list)."    },    "id": 3}

{    "data": "f3a45a19-acd0-4939-8e0c-e10743ff8e55",    "id": 4}

{    "success": true,    "result": {        "id": "f3a45a19-acd0-4939-8e0c-e10743ff8e55",        "name": "Eliza"    },    "id": 4

{    "some_key": "f3a45a19-acd0-4939-8e0c-e10743ff8e55",    "id": 5}

{    "success": false,    "result": {        "error_type": "<class 'middlewares.exceptions.InputDataValidationError'>",        "error_message": "ValidationError - 1 validation error for WrapRequest\nsome_key\n  extra fields not permitted (type=value_error.extra)"    },    "id": null}

{}

{    "success": true,    "result": "info_id=123 and request=<Request GET /info/123 >",    "id": null}

3. О нереализованной документации

У обработчиков, которые используются с классом WrapsKwargsHandler, есть всё, чтобы автоматически собрать документацию. К ним более не надо ничего добавлять. Так как классы pydantic.BaseModel позволяют получать json-schema, то остается только сделать скрипт сборки документации (если кратко, то надо: перед запуском приложения пройтись по всем обработчикам и у каждого заменить докстринг на swagger-описание, построенное на основе уже имеющегося докстринга и json-схем входящих данных и возврата).

И я эту документацию собираю. Но не стал рассказывать про это в статье. Причина в том, что я не нашел библиотеки для swagger и aiohttp, которая бы работала полностью как надо (или я не нашел способа заставить работать как надо).

Например, библиотека aiohttp-swagger некорректно отображает аргумент (в областях с примерами), если в аннотации есть алиас Union.

Библиотека aiohttp-swagger3, напротив, все прекрасно показывает, но не работает если в приложении есть sub_app.

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

4. Заключение

В итоге у нас имеются три класса для json middleware с разными возможностями. Любой из них можно изменить под свои нужды. Или создать на их основе новый.

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

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

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

При публикации статьи использовал MarkConv

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

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

Как известно, операционная система не позволяет управлять содержимым веб-сайта. Решение о том, как оформить веб-страницу, принимает дизайнер, а не пользователь, и большинство страниц светлые. Конечно, существуют плагины типа Dark Reader, которые пытаются решить эту проблему. Но, к сожалению, все они так или иначе искажают содержимое страниц и, кроме того, могут нарушать работу скриптов сайта.

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

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

Дальше было все просто. Техническая реализация идеи заняла у меня максимум пол-дня, включая тесты и размышления о том, где бы в коде лучше разместить переключатель. Я просто вынес все упоминания цвета из основного CSS-файла в файл light.css, потом скопировал его в файл dark.css и изменил некоторые (даже не все) цвета. Вот что получилось в итоге:

main.py:    async def create_app():        ...        jinja_setup(            app,            context_processors=[BaseHandler().context_processor])views.py:    class BaseHandler:        async def context_processor(self, request):            ...            return {                'theme': 'dark'                    if request.cookies.get('theme') == 'dark' else 'light'}base.html:    <head>        ...        <link rel="stylesheet"            href="{{ static_root_url }}/css/{{ theme }}.css">    </head>    <body>        ...        <div class="nav-block nav-item nav-theme">            {% if theme == 'dark' %}                <a href="javascript:void(0)" class="js-theme"                    data-theme="light">                    Дневной режим                </a>            {% else %}                <a href="javascript:void(0)" class="js-theme"                    data-theme="dark">                    Ночной режим                </a>            {% endif %}        </div>        ...    </body>base.js:    ...    $('.js-theme').on('click', function () {        $.setCookie('theme', $(this).data('theme'));        location.reload();    });

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

А теперь давайте помечтаем о том, как это могло бы быть, если бы было веб-стандартом. Предположим, в настройках ОС одним из первых пунктов идет что-то вроде Look and feel, который предоставляет выбор между темными и светлыми темами. ОС передает эту настройку браузеру, а тот, в свою очередь, веб-странице. Адаптивный сайт (или, если угодно, ресурс с персонализированным интерфейсом), получив такую инструкцию, должен отреагировать на нее переключением цветовой схемы и предоставить посетителю выбор: использовать текущую схему или переключить день на ночь (или наоборот).

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

Мы ведь делаем сайты для людей, да?

1. Что мы будем строить?
2. Подготовка окружения
3. Структура проекта
4. Установка зависимостей
5. Минимальное приложение
6. Giphy API клиент
7. Сервис поиска
8. Подключаем поиск
9. Немного рефакторинга
10. Добавляем тесты
11. Заключение

• Python 3.5+
• Virtual environment

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

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

• Клиент отправляет запрос указывая что искать и сколько результатов вернуть.
• Giphy Navigator возвращает ответ в формате json.
• Ответ включает:
  
  • поисковый запрос
  • количество результатов
  • список url гифок

{    "query": "Dependency Injector",    "limit": 10,    "gifs": [        {            "url": "https://giphy.com/gifs/boxes-dependent-swbf2-6Eo7KzABxgJMY"        },        {            "url": "https://giphy.com/gifs/depends-J56qCcOhk6hKE"        },        {            "url": "https://giphy.com/gifs/web-series-ccstudios-bro-dependent-1lhU8KAVwmVVu"        },        {            "url": "https://giphy.com/gifs/TheBoysTV-friends-friend-weneedeachother-XxR9qcIwcf5Jq404Sx"        },        {            "url": "https://giphy.com/gifs/netflix-a-series-of-unfortunate-events-asoue-9rgeQXbwoK53pcxn7f"        },        {            "url": "https://giphy.com/gifs/black-and-white-sad-skins-Hs4YzLs2zJuLu"        },        {            "url": "https://giphy.com/gifs/always-there-for-you-i-am-here-PlayjhCco9jHBYrd9w"        },        {            "url": "https://giphy.com/gifs/stream-famous-dollar-YT2dvOByEwXCdoYiA1"        },        {            "url": "https://giphy.com/gifs/i-love-you-there-for-am-1BhGzgpZXYWwWMAGB1"        },        {            "url": "https://giphy.com/gifs/life-like-twerk-9hlnWxjHqmH28"        }    ]}

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

mkdir giphynav-aiohttp-tutorialcd giphynav-aiohttp-tutorialpython3 -m venv venv

. venv/bin/activate

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

./ giphynavigator/    __init__.py    application.py    containers.py    views.py venv/ requirements.txt

Установка зависимостей

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

dependency-injectoraiohttpaiohttp-devtoolspyyamlpytest-aiohttppytest-cov

pip install -r requirements.txt

pip install httpie

Минимальное приложение

"""Views module."""from aiohttp import webasync def index(request: web.Request) -> web.Response:    query = request.query.get('query', 'Dependency Injector')    limit = int(request.query.get('limit', 10))    gifs = []    return web.json_response(        {            'query': query,            'limit': limit,            'gifs': gifs,        },    )

"""Application containers module."""from dependency_injector import containersfrom dependency_injector.ext import aiohttpfrom aiohttp import webfrom . import viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = aiohttp.Application(web.Application)    index_view = aiohttp.View(views.index)

"""Application module."""from aiohttp import webfrom .containers import ApplicationContainerdef create_app():    """Create and return aiohttp application."""    container = ApplicationContainer()    app: web.Application = container.app()    app.container = container    app.add_routes([        web.get('/', container.index_view.as_view()),    ])    return app

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

adev runserver giphynavigator/application.py --livereload

[18:52:59] Starting aux server at http://localhost:8001 [18:52:59] Starting dev server at http://localhost:8000 

http http://127.0.0.1:8000/

HTTP/1.1 200 OKContent-Length: 844Content-Type: application/json; charset=utf-8Date: Wed, 29 Jul 2020 21:01:50 GMTServer: Python/3.8 aiohttp/3.6.2{    "gifs": [],    "limit": 10,    "query": "Dependency Injector"}

Giphy API клиент

./ giphynavigator/    __init__.py    application.py    containers.py    giphy.py    views.py venv/ requirements.txt

"""Giphy client module."""from aiohttp import ClientSession, ClientTimeoutclass GiphyClient:    API_URL = 'http://api.giphy.com/v1'    def __init__(self, api_key, timeout):        self._api_key = api_key        self._timeout = ClientTimeout(timeout)    async def search(self, query, limit):        """Make search API call and return result."""        if not query:            return []        url = f'{self.API_URL}/gifs/search'        params = {            'q': query,            'api_key': self._api_key,            'limit': limit,        }        async with ClientSession(timeout=self._timeout) as session:            async with session.get(url, params=params) as response:                if response.status != 200:                    response.raise_for_status()                return await response.json()

• Провайдер Factory будет создавать GiphyClient.
• Провайдер Configuration будет передавать API ключ и таймаут GiphyClient.

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import aiohttpfrom aiohttp import webfrom . import giphy, viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = aiohttp.Application(web.Application)    config = providers.Configuration()    giphy_client = providers.Factory(        giphy.GiphyClient,        api_key=config.giphy.api_key,        timeout=config.giphy.request_timeout,    )    index_view = aiohttp.View(views.index)

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

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

./ giphynavigator/    __init__.py    application.py    containers.py    giphy.py    views.py venv/ config.yml requirements.txt

giphy:  request_timeout: 10

• Загрузить конфигурацию из config.yml
• Загрузить API ключ из переменной окружения GIPHY_API_KEY

"""Application module."""from aiohttp import webfrom .containers import ApplicationContainerdef create_app():    """Create and return aiohttp application."""    container = ApplicationContainer()    container.config.from_yaml('config.yml')    container.config.giphy.api_key.from_env('GIPHY_API_KEY')    app: web.Application = container.app()    app.container = container    app.add_routes([        web.get('/', container.index_view.as_view()),    ])    return app

export GIPHY_API_KEY=wBJ2wZG7SRqfrU9nPgPiWvORmloDyuL0

Для создания собственного ключа Giphy API следуйте этому руководству.

Сервис поиска

• Выполнять поиск
• Форматировать полученный ответ

./ giphynavigator/    __init__.py    application.py    containers.py    giphy.py    services.py    views.py venv/ requirements.txt

"""Services module."""from .giphy import GiphyClientclass SearchService:    def __init__(self, giphy_client: GiphyClient):        self._giphy_client = giphy_client    async def search(self, query, limit):        """Search for gifs and return formatted data."""        if not query:            return []        result = await self._giphy_client.search(query, limit)        return [{'url': gif['url']} for gif in result['data']]

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import aiohttpfrom aiohttp import webfrom . import giphy, services, viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = aiohttp.Application(web.Application)    config = providers.Configuration()    giphy_client = providers.Factory(        giphy.GiphyClient,        api_key=config.giphy.api_key,        timeout=config.giphy.request_timeout,    )    search_service = providers.Factory(        services.SearchService,        giphy_client=giphy_client,    )    index_view = aiohttp.View(views.index)

Подключаем поиск

"""Views module."""from aiohttp import webfrom .services import SearchServiceasync def index(        request: web.Request,        search_service: SearchService,) -> web.Response:    query = request.query.get('query', 'Dependency Injector')    limit = int(request.query.get('limit', 10))    gifs = await search_service.search(query, limit)    return web.json_response(        {            'query': query,            'limit': limit,            'gifs': gifs,        },    )

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import aiohttpfrom aiohttp import webfrom . import giphy, services, viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = aiohttp.Application(web.Application)    config = providers.Configuration()    giphy_client = providers.Factory(        giphy.GiphyClient,        api_key=config.giphy.api_key,        timeout=config.giphy.request_timeout,    )    search_service = providers.Factory(        services.SearchService,        giphy_client=giphy_client,    )    index_view = aiohttp.View(        views.index,        search_service=search_service,    )

adev runserver giphynavigator/application.py --livereload

http http://localhost:8000/ query=="wow,it works" limit==5

HTTP/1.1 200 OKContent-Length: 850Content-Type: application/json; charset=utf-8Date: Wed, 29 Jul 2020 22:22:55 GMTServer: Python/3.8 aiohttp/3.6.2{    "gifs": [        {            "url": "https://giphy.com/gifs/discoverychannel-nugget-gold-rush-rick-ness-KGGPIlnC4hr4u2s3pY"        },        {            "url": "https://giphy.com/gifs/primevideoin-ll1hyBS2IrUPLE0E71"        },        {            "url": "https://giphy.com/gifs/jackman-works-jackmanworks-l4pTgQoCrmXq8Txlu"        },        {            "url": "https://giphy.com/gifs/cat-massage-at-work-l46CzMaOlJXAFuO3u"        },        {            "url": "https://giphy.com/gifs/everwhatproductions-fun-christmas-3oxHQCI8tKXoeW4IBq"        },    ],    "limit": 10,    "query": "wow,it works"}

Немного рефакторинга

• Поисковый запрос по умолчанию
• Лимит количества результатов

"""Views module."""from aiohttp import webfrom .services import SearchServiceasync def index(        request: web.Request,        search_service: SearchService,        default_query: str,        default_limit: int,) -> web.Response:    query = request.query.get('query', default_query)    limit = int(request.query.get('limit', default_limit))    gifs = await search_service.search(query, limit)    return web.json_response(        {            'query': query,            'limit': limit,            'gifs': gifs,        },    )

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import aiohttpfrom aiohttp import webfrom . import giphy, services, viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = aiohttp.Application(web.Application)    config = providers.Configuration()    giphy_client = providers.Factory(        giphy.GiphyClient,        api_key=config.giphy.api_key,        timeout=config.giphy.request_timeout,    )    search_service = providers.Factory(        services.SearchService,        giphy_client=giphy_client,    )    index_view = aiohttp.View(        views.index,        search_service=search_service,        default_query=config.search.default_query,        default_limit=config.search.default_limit,    )

giphy:  request_timeout: 10search:  default_query: "Dependency Injector"  default_limit: 10

Добавляем тесты

./ giphynavigator/    __init__.py    application.py    containers.py    giphy.py    services.py    tests.py    views.py venv/ requirements.txt

"""Tests module."""from unittest import mockimport pytestfrom giphynavigator.application import create_appfrom giphynavigator.giphy import GiphyClient@pytest.fixturedef app():    return create_app()@pytest.fixturedef client(app, aiohttp_client, loop):    return loop.run_until_complete(aiohttp_client(app))async def test_index(client, app):    giphy_client_mock = mock.AsyncMock(spec=GiphyClient)    giphy_client_mock.search.return_value = {        'data': [            {'url': 'https://giphy.com/gif1.gif'},            {'url': 'https://giphy.com/gif2.gif'},        ],    }    with app.container.giphy_client.override(giphy_client_mock):        response = await client.get(            '/',            params={                'query': 'test',                'limit': 10,            },        )    assert response.status == 200    data = await response.json()    assert data == {        'query': 'test',        'limit': 10,        'gifs': [            {'url': 'https://giphy.com/gif1.gif'},            {'url': 'https://giphy.com/gif2.gif'},        ],    }async def test_index_no_data(client, app):    giphy_client_mock = mock.AsyncMock(spec=GiphyClient)    giphy_client_mock.search.return_value = {        'data': [],    }    with app.container.giphy_client.override(giphy_client_mock):        response = await client.get('/')    assert response.status == 200    data = await response.json()    assert data['gifs'] == []async def test_index_default_params(client, app):    giphy_client_mock = mock.AsyncMock(spec=GiphyClient)    giphy_client_mock.search.return_value = {        'data': [],    }    with app.container.giphy_client.override(giphy_client_mock):        response = await client.get('/')    assert response.status == 200    data = await response.json()    assert data['query'] == app.container.config.search.default_query()    assert data['limit'] == app.container.config.search.default_limit()

py.test giphynavigator/tests.py --cov=giphynavigator

platform darwin -- Python 3.8.3, pytest-5.4.3, py-1.9.0, pluggy-0.13.1plugins: cov-2.10.0, aiohttp-0.3.0, asyncio-0.14.0collected 3 itemsgiphynavigator/tests.py ...                                     [100%]---------- coverage: platform darwin, python 3.8.3-final-0 -----------Name                            Stmts   Miss  Cover---------------------------------------------------giphynavigator/__init__.py          0      0   100%giphynavigator/__main__.py          5      5     0%giphynavigator/application.py      10      0   100%giphynavigator/containers.py       10      0   100%giphynavigator/giphy.py            16     11    31%giphynavigator/services.py          9      1    89%giphynavigator/tests.py            35      0   100%giphynavigator/views.py             7      0   100%---------------------------------------------------TOTAL                              92     17    82%

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

Заключение

"""Application containers module."""from dependency_injector import containers, providersfrom dependency_injector.ext import aiohttpfrom aiohttp import webfrom . import giphy, services, viewsclass ApplicationContainer(containers.DeclarativeContainer):    """Application container."""    app = aiohttp.Application(web.Application)    config = providers.Configuration()    giphy_client = providers.Factory(        giphy.GiphyClient,        api_key=config.giphy.api_key,        timeout=config.giphy.request_timeout,    )    search_service = providers.Factory(        services.SearchService,        giphy_client=giphy_client,    )    index_view = aiohttp.View(        views.index,        search_service=search_service,        default_query=config.search.default_query,        default_limit=config.search.default_limit,    )

Что дальше?

• Узнайте больше о 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. Сделаем агента, который будет собирать данные о ценных бумагах и исторические цены по ним.

Код проекта

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

class Categories(JsonEndpoint):    url = "http://127.0.0.1:8888/categories"    params = {"page": range(100), "language": "en"}    headers = {"User-Agent": get_user_agent}    results_key = "*.slug"categories = Categories()class Posts(JsonEndpoint):    url = "http://127.0.0.1:8888/categories/{category}/posts"    params = {"page": range(100), "language": "en"}    url_params = {"category": categories.iter_results()}    results_key = "posts"    async def comments(self, post):        comments = Comments(            self.session,            url_params={"category": post.url.params["category"], "id": post["id"]},        )        return [comment async for comment in comments]posts = Posts()

class LoggingSink(Sink):    def transform(self, obj):        return repr(obj)    async def init(self):        from loguru import logger        self.logger = logger    async def process(self, obj):        self.logger.info(obj)        return Truesink = LoggingSink(num_tasks=1)

worker = Worker(endpoint=posts, sinks=[loggingsink, mongosink])worker.run()

SELECT r.id, r.auth_user_id, r.status, r.updated, r.label, r.content, u.username,    ARRAY_AGG(t.tag) tag, COUNT(*) OVER() cnt,    (        SELECT COUNT(*) FROM record r2            WHERE                r2.parent_id IS NOT NULL                AND r2.parent_id = r.id                AND r2.status = 'new'    ) AS partsFROM record rJOIN auth_user u ON u.id = r.auth_user_idLEFT JOIN tag t ON t.kind_id = r.id AND t.kind = 'rec'WHERE r.parent_id IS NULL AND r.status = 'new'GROUP BY r.id, u.usernameORDER BY r.updated DESCLIMIT 10 OFFSET 0

Выводы