Рубрики
Без рубрики

DIY Async Web Framework 🙌

Узнайте, как работает современные асинхронные рамки … Теги с WebDev, учебником, Python, Showdev.

Первоначально опубликовано на github.com/hzlmn/diy-Async-web-framework

Если вы когда-нибудь задумывались, насколько современными каркасами нравятся AioHTTP и Санич Работа, тогда это руководство для вас.

Вступление

Асинхронное программирование стало очень популярным в последние несколько лет в сообществе Python. Библиотеки, как AioHTTP Показать невероятное использование растет. Они обрабатывают большое количество одновременного соединения, пока все равно поддерживают хорошую читаемость кода и простоту. Не давно, Джанго совершил При добавлении поддержки Async в следующей основной версии. Так что будущее асинхронного питона довольно яркое, насколько вы можете понять. Однако для большого количества разработчиков, которые пришли из стандартной модели блокировки, рабочий механизм этих инструментов может показаться запутанным. Так что в этом кратком направлении я пытался пойти за сцену и уточнить процесс, повторно наращиваю немного AIOHTTP. клон с нуля. Мы начнем только с базовой выборки от официальной документации и постепенно добавляем все необходимые функциональные возможности, которые нам все нравятся. Итак, начнем.

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

Для нетерпеливых целей, окончательный исходный код доступен в HZLMN/эскиз

Оглавление 📖

  • Asyncio низкоуровневые API, транспорт и протоколы

  • Создание серверного протокола

  • Запрос/объекты ответа

  • Приложение и Urldispatcher.

  • Идти дальше

    • Маршзяр
    • Hiddrwares.
    • Крючки жизненного цикла приложения
    • Лучшие исключения
    • Изящное отключение
  • Пример приложения

  • Заключение

Asyncio низкоуровневые API, транспорт и протоколы

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

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

Из коробки он поддерживает только TCP , UDP , SSL и подпроцессы. Библиотеки реализуют свой собственный уровень (HTTP, FTP и т. Д.) На основе базовых транспортов и доступных API.

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

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

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

server.py

import asyncio

class Server(asyncio.Protocol):
    def connection_made(self, transport):
        self._transport = transport

    def data_received(self, data):
        message = data.decode()

        self._transport.write(data)

        self._transport.close()

loop = asyncio.get_event_loop()

coro = loop.create_server(Server, '127.0.0.1', 8080)
server = loop.run_until_complete(coro)

try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

server.close()
loop.run_until_complete(server.wait_closed())
loop.close()
$ curl http://127.0.0.1:8080
GET / HTTP/1.1
Host: 127.0.0.1:8080
User-Agent: curl/7.54.0
Accept: */*

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

Как Http работает над TCP Транспорт Мы уже можем отправить Http Запросы на наш сервер, однако, мы получаем их в формате RAW и работаем с этим, будет раздражать, как вы можете догадаться. Так что следующий шаг нам нужно добавить лучше Http механизм обработки.

Создание серверного протокола

Давайте добавим разборы запроса, чтобы мы могли извлечь некоторые полезные данные, такие как заголовки, тело, путь и работают с ними вместо необработанного текста. Пашина – это сложная тема, и она, безусловно, выходит из объема этого руководства, вот почему мы будем использовать httptools MagicStack для этого, как он быстро быстро, стандартный совместимый и довольно гибкий.

AioHTTP С С другой стороны, имеет собственную рукописную паршуру на основе Python, а также привязки к узлу http-parser Отказ

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

http_parser.py

class HttpParserMixin:
    def on_body(self, data):
        self._body = data

    def on_url(self, url):
        self._url = url

    def on_message_complete(self):
        print(f"Received request to {self._url.decode(self._encoding)}")

    def on_header(self, header, value):
        header = header.decode(self._encoding)
        self._headers[header] = value.decode(self._encoding)

Теперь, когда мы работаем Httpparsermixin. давайте изменим немного нашего Сервер и применять микс.

server.py

import asyncio

from httptools import HttpRequestParser

from .http_parser import HttpParserMixin

class Server(asyncio.Protocol, HttpParserMixin):
    def __init__(self, loop):
        self._loop = loop
        self._encoding = "utf-8"
        self._url = None
        self._headers = {}
        self._body = None
        self._transport = None
        self._request_parser = HttpRequestParser(self)

    def connection_made(self, transport):
        self._transport = transport

    def connection_lost(self, *args):
        self._transport = None

    def data_received(self, data):
        # Pass data to our parser
        self._request_parser.feed_data(data)

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

server.py

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    serv = Server(loop)
    server = loop.run_until_complete(loop.create_server(lambda: serv, port=8080))

    try:
        print("Started server on ::8080")
        loop.run_until_complete(server.serve_forever())
    except KeyboardInterrupt:
        server.close()
        loop.run_until_complete(server.wait_closed())
        loop.stop()

> python server.py
Started server on ::8080
> curl http://127.0.0.1:8080/hello

Запрос/объекты ответа

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

Давайте создадим база Запрос Класс, который будет объединить все входящие Http запросить информацию. Мы будем использовать Ярл Библиотека для устранения URL-адресов, убедитесь, что вы установили его с PIP.

request.py

import json

from yarl import URL

class Request:
    _encoding = "utf_8"

    def __init__(self, method, url, headers, version=None, body=None, app=None):
        self._version = version
        self._method = method.decode(self._encoding)
        self._url = URL(url.decode(self._encoding))
        self._headers = headers
        self._body = body

    @property
    def method(self):
        return self._method

    @property
    def url(self):
        return self._url

    @property
    def headers(self):
        return self._headers

    def text(self):
        if self._body is not None:
            return self._body.decode(self._encoding)

    def json(self):
        text = self.text()
        if text is not None:
            return json.loads(text)

    def __repr__(self):
        return f""

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

response.py

import http.server

web_responses = http.server.BaseHTTPRequestHandler.responses

class Response:
    _encoding = "utf-8"

    def __init__(
        self,
        body=None,
        status=200,
        content_type="text/plain",
        headers=None,
        version="1.1",
    ):
        self._version = version
        self._status = status
        self._body = body
        self._content_type = content_type
        if headers is None:
            headers = {}
        self._headers = headers

    @property
    def body(self):
        return self._body

    @property
    def status(self):
        return self._status

    @property
    def content_type(self):
        return self._content_type

    @property
    def headers(self):
        return self._headers

    def add_body(self, data):
        self._body = data

    def add_header(self, key, value):
        self._headers[key] = value

    def __str__(self):
        """We will use this in our handlers, it is actually generation of raw HTTP response,
        that will be passed to our TCP transport
        """
        status_msg, _ = web_responses.get(self._status)

        messages = [
            f"HTTP/{self._version} {self._status} {status_msg}",
            f"Content-Type: {self._content_type}",
            f"Content-Length: {len(self._body)}",
        ]

        if self.headers:
            for header, value in self.headers.items():
                messages.append(f"{header}: {value}")

        if self._body is not None:
            messages.append("\r\n" + self._body)

        return "\r\n".join(messages)

    def __repr__(self):
        return f""

Как вы можете видеть код, довольно проплее, мы инкафулируем все наши данные и предоставим правильные GetTers. Также у нас мало помощников для чтения тела текст и json. это будет использоваться позже. Нам также нужно обновить наши Сервер на самом деле построить Запрос объект из сообщения.

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

http_parser.py

class HttpParserMixin:
    ...

    def on_message_complete(self):
        self._request = self._request_class(
            version=self._request_parser.get_http_version(),
            method=self._request_parser.get_method(),
            url=self._url,
            headers=self._headers,
            body=self._body,
        )

    ...

Сервер также нужен небольшая модификация для создания Ответ объект и пройти кодированное значение для асинсио. Транспорт Отказ

server.py

from .response import Response
...

class Server(asyncio.Protocol, HttpParserMixin):
    ...

    def __init__(self, loop):
        ...
        self._request = None
        self._request_class = Request

    ...

    def data_received(self, data):
        self._request_parser.feed_data(data)

        resp = Response(body=f"Received request on {self._request.url}")
        self._transport.write(str(resp).encode(self._encoding))

        self._transport.close()

Сейчас бегу на наших Server.py. Мы сможем увидеть Полученный запрос на/путь В ответ на Curl Call http: localhost: 8080/Путь Отказ

Приложение и Urldispatcher.

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

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

router.py

from .response import Response

class UrlDispatcher:
    def __init__(self):
        self._routes = {}

    async def _not_found(self, request):
         return Response(f"Not found {request.url} on this server", status=404)

    def add_route(self, method, path, handler):
        self._routes[(method, path)] = handler

    def resolve(self, request):
        key = (request.method, request.url.path)
        if key not in self._routes:
            return self._not_found
        return self._routes[key]

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

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

import asyncio

from .router import UrlDispatcher
from .server import Server
from .response import Response

class Application:
    def __init__(self, loop=None):
        if loop is None:
            loop = asyncio.get_event_loop()

        self._loop = loop
        self._router = UrlDispatcher()

    @property
    def loop(self):
        return self._loop

    @property
    def router(self):
        return self._router

    def _make_server(self):
        return Server(loop=self._loop, handler=self._handler, app=self)

    async def _handler(self, request, response_writer):
        """Process incoming request"""
        handler = self._router.resolve(request)
        resp = await handler(request)

        if not isinstance(resp, Response):
            raise RuntimeError(f"expect Response instance but got {type(resp)}")

        response_writer(resp)

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

server.py

class Server(asyncio.Protocol, HttpParserMixin):
    ...

    def __init__(self, loop, handler, app):
        self._loop = loop
        self._url = None
        self._headers = {}
        self._body = None
        self._transport = None
        self._request_parser = HttpRequestParser(self)
        self._request = None
        self._request_class = Request
        self._request_handler = handler
        self._request_handler_task = None

    def response_writer(self, response):
        self._transport.write(str(response).encode(self._encoding))
        self._transport.close()

    ...

http_parser.py

class HttpParserMixin:
    def on_body(self, data):
        self._body = data

    def on_url(self, url):
        self._url = url

    def on_message_complete(self):
        self._request = self._request_class(
            version=self._request_parser.get_http_version(),
            method=self._request_parser.get_method(),
            url=self._url,
            headers=self._headers,
            body=self._body,
        )

        self._request_handler_task = self._loop.create_task(
            self._request_handler(self._request, self.response_writer)
        )

    def on_header(self, header, value):
        header = header.decode(self._encoding)
        self._headers[header] = value.decode(self._encoding)

Наконец, когда у нас есть основные функциональные возможности готовы, можете зарегистрировать новые маршруты и обработчики, давайте добавим простой помощник для фактического запуска нашего экземпляра приложения (похоже на Web.run_app в AioHTTP ).

application.py

def run_app(app, host="127.0.0.1", port=8080, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()

    serv = app._make_server()
    server = loop.run_until_complete(
        loop.create_server(lambda: serv, host=host, port=port)
    )

    try:
        print(f"Started server on {host}:{port}")
        loop.run_until_complete(server.serve_forever())
    except KeyboardInterrupt:
        server.close()
        loop.run_until_complete(server.wait_closed())
        loop.stop()

И теперь, время, чтобы сделать простое приложение с нашим свежим инструментом.

app.py

import asyncio

from .response import Response
from .application import Application, run_app

app = Application()

async def handler(request):
    return Response(f"Hello at {request.url}")

app.router.add_route("GET", "/", handler)

if __name__ == "__main__":
    run_app(app)

Если вы запустите его, а затем сделайте Получить Запрос на / , вы сможете увидеть Здравствуйте в/ и 404 Ответ на все другие маршруты. Ура, мы сделали это, однако, все еще есть большая комната для улучшений.

$ curl 127.0.0.1:8080/
Hello at /

$ curl 127.0.0.1:8080/invalid
Not found /invalid on this server

Идти дальше

До сих пор у нас есть все основные функциональные возможности и бега, но нам все еще нужно изменить определенные вещи в нашей «рамках». Прежде всего, как мы обсуждали ранее, ваш роутер отсутствует параметризованные маршруты, он «должен иметь» особенность всех современных библиотек. Далее нам нужно добавить поддержку для Ambrapares, она также очень распространена и мощная концепция. Отличная вещь о AioHTTP что я влюбился в это Крючки жизненного цикла приложений (например, on_startup , On_shutdown , on_cleanup. ) Поэтому мы, безусловно, должны попытаться реализовать его.

Маршзяр

В настоящее время наше Urdispatcher Довольно наклоняется, и он работает с зарегистрированными путями URL в качестве строки. Во-первых, что нам нужно, на самом деле добавляет поддержку образцов, таких как /user/{имя пользователя} нашему решить метод. Также нам нужно _format_pattern Помощник, который будет нести ответственность за создание фактического регулярного выражения из параметризованной строки. Также, как вы можете отметить, у нас есть другой помощник _method_not_allowed и методы более простых определений Получить , Пост и т. Д. Маршруты.

router.py

import re

from functools import partialmethod

from .response import Response

class UrlDispatcher:
    _param_regex = r"{(?P\w+)}"

    def __init__(self):
        self._routes = {}

    async def _not_found(self, request):
        return Response(f"Could not find {request.url.raw_path}")

    async def _method_not_allowed(self, request):
        return Response(f"{request.method} not allowed for {request.url.raw_path}")

    def resolve(self, request):
        for (method, pattern), handler in self._routes.items():
            match = re.match(pattern, request.url.raw_path)

            if match is None:
                return None, self._not_found

            if method != request.method:
                return None, self._method_not_allowed

            return match.groupdict(), handler

    def _format_pattern(self, path):
        if not re.search(self._param_regex, path):
            return path

        regex = r""
        last_pos = 0

        for match in re.finditer(self._param_regex, path):
            regex += path[last_pos: match.start()]
            param = match.group("param")
            regex += r"(?P<%s>\w+)" % param
            last_pos = match.end()

        return regex

    def add_route(self, method, path, handler):
        pattern = self._format_pattern(path)
        self._routes[(method, pattern)] = handler

    add_get = partialmethod(add_route, "GET")

    add_post = partialmethod(add_route, "POST")

    add_put = partialmethod(add_route, "PUT")

    add_head = partialmethod(add_route, "HEAD")

    add_options = partialmethod(add_route, "OPTIONS")

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

application.py

class Application:
    ...
    async def _handler(self, request, response_writer):
        """Process incoming request"""
        match_info, handler = self._router.resolve(request)

        request.match_info = match_info

        ...

Hiddrwares.

Для тех, кто не знаком с этой концепцией, простыми словами промежуточное программное обеспечение это просто COROUTENINE, который может модифицировать объект входящего запроса или изменять ответ обработчика. Он будет уволен перед каждым запросом к серверу. Реализация довольно тривиальна для наших потребностей. Прежде всего, нам нужно добавить список зарегистрированных потрачений внутри наших Приложение Объект и менять немного Application._Handler пробежать через них. Каждое промежуточное программное обеспечение должно работать с результатом предыдущего в цепочке.

application.py

from functools import partial
...

class Application:
    def __init__(self, loop=None, middlewares=None):
        ...
        if middlewares is None:
            self._middlewares = []

    ...

    async def _handler(self, request, response_writer):
        """Process incoming request"""
        match_info, handler = self._router.resolve(request)

        request.match_info = match_info

        if self._middlewares:
            for md in self._middlewares:
                handler = partial(md, handler=handler)

        resp = await handler(request)

        ...

Теперь давайте попробуем добавить запрос на промежуточное программное обеспечение на наше простое приложение.

app.py

import asyncio

from .response import Response
from .application import Application, run_app

async def log_middleware(request, handler):
    print(f"Received request to {request.url.raw_path}")
    return await handler(request)

app = Application(middlewares=[log_middleware])

async def handler(request):
    return Response(f"Hello at {request.url}")

app.router.add_route("GET", "/", handler)

if __name__ == "__main__":
    run_app(app)

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

Крючки жизненного цикла приложения

Следующий шаг Давайте добавим поддержку для выполнения определенных Ciboutines в ответ на события, такие как начальный сервер и останавливая его. Это довольно аккуратная особенность AioHTTP Отказ Есть много сигналов, таких как on_startup , On_shutdown , on_response_prepared Чтобы назвать несколько, но для нашей потребности Давайте сохраним его просто и просто реализуйте Запуск & Выключение помощники.

Внутри Приложение Нам нужно добавить список фактических обработчиков для каждого события с надлежащей инкапсуляцией и предоставлять GetTers. Тогда актуально Запуск и Выключение COROUTINES и добавить соответствующие вызовы на run_app помощник.

application.py

class Application:
    def __init__(self, loop=None, middlewares=None):
        ...
        self._on_startup = []
        self._on_shutdown = []

    ... 

    @property
    def on_startup(self):
        return self._on_startup

    @property
    def on_shutdown(self):
        return self._on_shutdown

    async def startup(self):
        coros = [func(self) for func in self._on_startup]
        await asyncio.gather(*coros, loop=self._loop)

    async def shutdown(self):
        coros = [func(self) for func in self._on_shutdown]
        await asyncio.gather(*coros, loop=self._loop)

    ...

def run_app(app, host="127.0.0.1", port=8080, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()

    serv = app._make_server()

    loop.run_until_complete(app.startup())

    server = loop.run_until_complete(
        loop.create_server(lambda: serv, host=host, port=port)
    )

    try:
        print(f"Started server on {host}:{port}")
        loop.run_until_complete(server.serve_forever())
    except KeyboardInterrupt:
        loop.run_until_complete(app.shutdown())
        server.close()
        loop.run_until_complete(server.wait_closed())
        loop.stop()

Лучшие исключения

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

Итак, во-первых, давайте создадим нашу базу HttpException класс и несколько помощников, основываясь на этом, что нам может понадобиться как Httpnotfound Для непризнанных путей Httpbadrequest Для пользовательских проблем и Httpfound для перенаправления.

from .response import Response

class HTTPException(Response, Exception):
    status_code = None

    def __init__(self, reason=None, content_type=None):
        self._reason = reason
        self._content_type = content_type

        Response.__init__(
            self,
            body=self._reason,
            status=self.status_code,
            content_type=self._content_type or "text/plain",
        )

        Exception.__init__(self, self._reason)


class HTTPNotFound(HTTPException):
    status_code = 404


class HTTPBadRequest(HTTPException):
    status_code = 400


class HTTPFound(HTTPException):
    status_code = 302

    def __init__(self, location, reason=None, content_type=None):
        super().__init__(reason=reason, content_type=content_type)
        self.add_header("Location", location)

Тогда нам нужно немного модифицировать наши Application._Handler на самом деле ловить веб-исключения.

application.py

class Application:
    ...
    async def _handler(self, request, response_writer):
        """Process incoming request"""
        try:
            match_info, handler = self._router.resolve(request)

            request.match_info = match_info

            if self._middlewares:
                for md in self._middlewares:
                    handler = partial(md, handler=handler)

            resp = await handler(request)
        except HTTPException as exc:
            resp = exc

        ...

Также теперь мы можем бросить _not_found & _method_not_allowed помощники от нашего Urdispatcher. и вместо этого просто поднять правильные исключения.

router.py

class UrlDispatcher:
    ...
    def resolve(self, request):
        for (method, pattern), handler in self._routes.items():
            match = re.match(pattern, request.url.raw_path)

            if match is None:
                raise HTTPNotFound(reason=f"Could not find {request.url.raw_path}")

            if method != request.method:
                raise HTTPBadRequest(reason=f"{request.method} not allowed for {request.url.raw_path}")

            return match.groupdict(), handler

        ...

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

helpers.py

import traceback

from .response import Response

server_exception_templ = """

500 Internal server error

Server got itself in trouble : {exc}

{traceback}

""" def format_exception(exc): resp = Response(status=500, content_type="text/html") trace = traceback.format_exc().replace("\n", "") msg = server_exception_templ.format(exc=str(exc), traceback=trace) resp.add_body(msg) return resp

Так же просто, и теперь просто поймай все Исключение внутри нашего Application._Handler и генерировать реальный ответ HTML с нашим помощником.

application.py

class Application:
    ...
    async def _handler(self, request, response_writer):
        """Process incoming request"""
        try:
            match_info, handler = self._router.resolve(request)

            request.match_info = match_info

            if self._middlewares:
                for md in self._middlewares:
                    handler = partial(md, handler=handler)

            resp = await handler(request)
        except HTTPException as exc:
            resp = exc
        except Exception as exc:
            resp = format_exception(exc)
        ...

Изящное отключение

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

application.py

...

def run_app(app, host="127.0.0.1", port=8080, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()

    serv = app._make_server()

    loop.run_until_complete(app.startup())

    server = loop.run_until_complete(
        loop.create_server(lambda: serv, host=host, port=port)
    )

    loop.add_signal_handler(
        signal.SIGTERM, lambda: asyncio.ensure_future(app.shutdown())
    )

    ...

Пример приложения

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

app.py

from .application import Application, run_app

async def on_startup(app):
    # you may query here actual db, but for an example let's just use simple set.
    app.db = {"john_doe",}

async def log_middleware(request, handler):
    print(f"Received request to {request.url.raw_path}")
    return await handler(request)

async def handler(request):
    username = request.match_info["username"]
    if username not in request.app.db:
        raise HTTPNotFound(reason=f"No such user with as {username} :(")

    return Response(f"Welcome, {username}!")

app = Application(middlewares=[log_middleware])

app.on_startup.append(on_startup)

app.router.add_get("/{username}", handler)

if __name__ == "__main__":
    run_app(app)

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

Заключение

Суммируя его вверх, в ~ 500 строк мы с ручным составом довольно простые, но мощные микроформа Micro Frame, вдохновленные AioHTTP & Санич Отказ Конечно, это не готовое программное обеспечение для производства, поскольку он по-прежнему пропускает много полезных и важных функций, таких как более надежный сервер, лучшая поддержка HTTP для полной корреляции со спецификацией, веб-сокетами называть несколько. Однако я считаю, что через этот процесс мы разрабатываем лучше понять, как построили такие инструменты. Поскольку знаменитый физик Ричард Фейнман сказал: «То, что я не могу создать, я не понимаю». Поэтому я надеюсь, что вам понравилось это руководство, увидеть Я! 👋

Оригинал: “https://dev.to/hzlmn/diy-async-web-framework-4eh3”