ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 数据库 > 数据库开发技术 > fastapi框架源码分析系列-starlette框架之Debug方式学源码（1）-认识ASGI接口的协议

fastapi框架源码分析系列-starlette框架之Debug方式学源码（1）-认识ASGI接口的协议

原创 数据库开发技术 作者：程序员小钟同学 时间：2022-06-07 13:27:16 0 删除 编辑

说明

由于目前我接触比较多则是fastapi框架，而fastapi是基于starlette扩展而来，所以当我需要深入了解fastapi框架源码的时候，需要基于fastapi扩展自己的插件的是好，感觉有必要还是需要了解starlette框架设计，关于starlette框架源码分析，其实网络上已有大神做想跟源码分析解说，如下大神：

Starlette 源码阅读 系列 https://www.jianshu.com/p/9791a303cc38

然则【撸码人的世界】事必躬亲，才会有自己的体会，出于近期需要了解ASGI框架方面设计思路，所以借鉴优秀框架设计开始，这样从别人的思路中提取有效信息。

需要说明的是，我们当前只是分析一个应用开发框架的使用，而不是分析ASGI 服务器，例如uvicorn、daphne或hypercorn是属于asgi服务器是范畴。后续有机会再展开分析。

starlette 源码分析环境搭建

当我们的需要了解一个框架之前，我们肯定需要先学会这个框架是如何使用，只有知道他是如何使用才容易找到的突破口，starlette框架官方文档本身其实已经提供了丰富的案例。以下学习笔记，主要是参考starlette官网而来。

starlette简介说明

  • 是一个asgi框架工具包
  • 是一个异步http web框架
  • 支持websoket,后台任务
  • 支持中间件自定义

其实上面这些和Fastapi框架差不多。不多叙述！

starlette 安装

其实安装了fastapi的时候就安装了具体starlette所依赖的版本，为方便结合后续fastapi源码分析，这里直接的进行fastapi安装：

pip install fastapi[all]

然后拷贝源码出来，放当前目录以供后续分析。

当前starlette版本信息为：

__version__ = "0.19.1"

关于ASGI服务器-uvicorn

我们知道uvicorn是ASGI服务器，只要符合它一个协议内容的我们都完成可以直接的使用，也就是说实现ASGI接口的协议规范，不管是类还是返回都可以，如下示例代码：

import uvicorn

async def app(scope, receive, send):
    assert scope['type'] == 'http'

    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [
            [b'content-type', b'text/plain'],
        ],
    })
    await send({
        'type': 'http.response.body',
        'body': b'Hello, world!',
    })
if __name__ == "__main__":
    uvicorn.run("main:app", host="127.0.0.1", port=5000, log_level="info")

只要符合async def app(scope, receive, send)方式函数都可以当做一个请求处理，如上代码结果： 当我们启动服务之后，直接随意访问任何一个路由地址怎么都会返回如下图所示结果:

所以可以充分说明了只要符合async def app(scope, receive, send)方式函数的组件或函数都可以被当做一个请求处理。

starlette 初步分析及使用

说明示例直接来自官网，这里启动上只是修改了以下和fastapi一样启动即可。

ASGI接口的协议规中进一步分析-三种方式实现

接前面说的关于 符合app(scope, receive, send)协议则可以当做一个请求处理，那么我们starlette在设计的是其实都及基于这种设计思路展开，不管是路由还是中间件还是app对象本身，他内部都是饶这app(scope, receive, send)协议本身展开，所以一下代码示例也是可以运行的，如下示例代码：

from starlette.responses import PlainTextResponse
async def app(scope, receive, send):
    assert scope['type'] == 'http'
    response = PlainTextResponse('Hello, world!')
    await response(scope, receive, send)
if __name__ == '__main__':
    import uvicorn
    uvicorn.run('main:app')

需要说明的是上面的示例是使用函数的方式来定义，如果是使用的类方式，我们可以参考源码中端点类的方式，所以我们的改为使用类的方式，如下代码示例（基于__await__的方式）：

from starlette.responses import PlainTextResponse
from starlette.types import Message, Receive, Scope, Send
import typing


class app:
    def __init__(self, scope: Scope, receive: Receive, send: Send) -> None:
        assert scope["type"] == "http"
        self.scope = scope
        self.receive = receive
        self.send = send
        self._allowed_methods = [
            method
            for method in ("GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS")
            if getattr(self, method.lower(), None) is not None
        ]

    def __await__(self) -> typing.Generator:
        print("会自动进行调用！")
        return self.dispatch().__await__()

    async def dispatch(self) -> None:
        print("dispatch")
        response = PlainTextResponse('Hello, world!')
        await response(self.scope, self.receive, self.send)


if __name__ == '__main__':
    import uvicorn
    uvicorn.run('dsasd:app')

还可以进一步在简化类方式定义如下（基于__call__的方式）：

class app:
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        response = PlainTextResponse('Hello, world!')
        await response(scope, receive, send)

if __name__ == '__main__':
    import uvicorn
    uvicorn.run('dsasd:app')

此效果和前面的示例一样，只有启动服务任意访问一个地址，都可以返回如下图所示结果:

定义app对象方式来使用

下面开始根据官方来学习基本的使用：

from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.routing import Route, Mount, WebSocketRoute
from starlette.staticfiles import StaticFiles


def homepage_sync(request):
    return PlainTextResponse('Hello, world!')

async def homepage_async(request):
    return PlainTextResponse('Hello, world!')


def user_me(request):
    username = "John Doe"
    return PlainTextResponse('Hello, %s!' % username)

def user(request):
    username = request.path_params['username']
    return PlainTextResponse('Hello, %s!' % username)

async def websocket_endpoint(websocket):
    await websocket.accept()
    await websocket.send_text('Hello, websocket!')
    await websocket.close()

def startup():
    print('Ready to go')


routes = [
    Route('/homepage_sync', homepage_sync),
    Route('/homepage_async', homepage_async),
    Route('/user/me', user_me),
    Route('/user/{username}', user),
    WebSocketRoute('/ws', websocket_endpoint),
    Mount('/static', StaticFiles(directory="static")),
]

app = Starlette(debug=True, routes=routes, on_startup=[startup])

if __name__ == '__main__':
    import uvicorn
    uvicorn.run('main:app')

代码说明：

  • 定义一个同步类型路由函数homepage_sync，并显示传递request作为参数进行传递
  • 定义一个异步类型路由函数homepage_async，并显示传递request作为参数进行传递
  • 定义需要注册的路由列表
routes = [
    Route('/homepage_sync', homepage_sync),
    Route('/homepage_async', homepage_async),
    Route('/user/me', user_me),
    Route('/user/{username}', user),
    WebSocketRoute('/ws', websocket_endpoint),
    Mount('/static', StaticFiles(directory="static")),
]
  • 定义支持websocket的路由websocket_endpoint，并显示传递websocket连接对象
  • 定义Starlette具体的app实例化对象，并开启debug,然后设置app对象具体的路由列表为rroutes=routes
  • 定义在应用程序启动时运行的可调用对象列表on_startup。启动处理程序可调用对象不接受任何参数，并且可以是标准函数或异步函数。
  • 定义在应用程序关闭时运行的可调用项列表on_shutdown。关闭处理程序可调用对象不接受任何参数，并且可以是标准函数或异步函数。
  • 最终通过uvicorn.run('main:app')来启动服务

此时启动服务之后，我们可以通过访问具体的地址得到对应路由返回信息。 如：

Starlette框架请求报文和响应报文封装-Request -Response

首先Request其实对传入参数信息一种解析之后的封装，如下示例代码：

from starlette.requests import Request
from starlette.responses import Response


async def app(scope, receive, send):
    assert scope['type'] == 'http'
    request = Request(scope, receive)
    content = f'返回内容信息为：{request.method},{ request.url.path}'
    response = Response(content, media_type='text/plain')
    await response(scope, receive, send)

if __name__ == '__main__':
    import uvicorn
    uvicorn.run('main:app')

关键核心位置是：

  request = Request(scope, receive)
  --
  response = Response(content, media_type='text/plain')

经过解析后返回解析好的对象，就可以直接通过request进行想跟封装好的方法进行相关的调用，提取具体请求参数信息。 另外的response则是对要返回响应体的内容进行封装，其中它还支持其他类型响应体封装，如下类型：

  • Response
  • HTMLResponse
  • JSONResponse
  • RedirectResponse
  • PlainTextResponse
  • StreamingResponse
  • FileResponse

此时分析Response核心源码内容为：

class Response:
    media_type = None
    charset = "utf-8"

    def __init__(
        self,
        content: typing.Any = None,
        status_code: int = 200,
        headers: typing.Optional[typing.Mapping[str, str]] = None,
        media_type: typing.Optional[str] = None,
        background: typing.Optional[BackgroundTask] = None,
    ) -> None:
        self.status_code = status_code
        if media_type is not None:
            self.media_type = media_type
        self.background = background
        self.body = self.render(content)
        self.init_headers(headers)

    def render(self, content: typing.Any) -> bytes:
        if content is None:
            return b""
        if isinstance(content, bytes):
            return content
        return content.encode(self.charset)

省略部分源码····

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        await send(
            {
                "type": "http.response.start",
                "status": self.status_code,
                "headers": self.raw_headers,
            }
        )
        await send({"type": "http.response.body", "body": self.body})

        if self.background is not None:
            await self.background()

其中关键则是 ‘__call__’方法是在

response = Response('Hello, world!', media_type='text/plain')
await response(scope, receive, send)

当调用response(scope, receive, send)会去执行Response类里面的 ‘__call__’，

此时返回结果如下图所示：

而其中输出信息为：

通过上面分析，大概能知道我们的响应报文的封装解析思路。

Starlette框架实现了ASGI规范的路由类-Router

查看路由类的Router源码：

他是基于BaseRoute扩展而来，在细看BaseRoute里面的源码，则他也是实现了ASGI规范的的，如下示例截图所示：

BaseRoute路由类则也是一个app规范的实现，且需要子类去实现具体handle处理细节。测试我们查看，Router的：

他除主要对一些请求方法的校验处理，做对应的响应处理，如果当前请求的方法不是路由定义的方法范围内，那么直接返回405错误，也就是不允许使用此类的方法进行提交的错误。如果是的话，那么它又这一次的请求的参数信息，往下继续传递给路由自身的self.app对象，那么这个self.app对象是啥呢？

分析路由初始化时候传入的参数信息为：

通过初始化参数分析可以得知，-- 如果他是一个方法或者函数，那么此时的app对象则是处理好的请求和响应的一个ASGIApp，如下request_response代码：

def request_response(func: typing.Callable) -> ASGIApp:
    """
    Takes a function or coroutine `func(request) -> response`,
    and returns an ASGI application.
    """
    is_coroutine = iscoroutinefunction_or_partial(func)

    async def app(scope: Scope, receive: Receive, send: Send) -> None:
        request = Request(scope, receive=receive, send=send)
        if is_coroutine:
            response = await func(request)
        else:
            response = await run_in_threadpool(func, request)
        await response(scope, receive, send)

    return app

相当于我就直接处理路由函数逻辑并且给参数自动传入request的参数，然后继续执行response响应内的call内部逻辑。

如不过不是函数或方法的话，那么直接的继续递归到下一个，有有可能路由里面还嵌套其他符合类型ASGIApp协议规则的类（如子路由，路由中间件等），那么我就不断执行递归知道内部的逻辑处理完成！！！所以这个就可以理解为了大神分析篇里面有一个说法无线套娃····

Starlette框架实现了ASGI规范HTTPEndpoint端点类-

HTTPEndpoint类其实和路由类似，它本身也是用于定义具体路由信息的，只是它实现的方式不是方法或函数，在上面的路由实例中，我们定义路由是使用函数的方式，所以HTTPEndpoint则是使用类的方式来实现，这也印证了，我们的路由中app判断的逻辑处理的规则，当它不是一个类或方法的时候，那么就继续递归到下一个，也就是进入到使用自定义类的方式实现的示例中， 可以回看前面的代码示例： 需要说明的是上面的示例是使用函数的方式来定义，如果是使用的类方式，我们可以参考源码中端点类的方式，所以我们的改为使用类的方式，如下代码示例：

from starlette.responses import PlainTextResponse
from starlette.types import Message, Receive, Scope, Send
import typing


class app:
    def __init__(self, scope: Scope, receive: Receive, send: Send) -> None:
        assert scope["type"] == "http"
        self.scope = scope
        self.receive = receive
        self.send = send
        self._allowed_methods = [
            method
            for method in ("GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS")
            if getattr(self, method.lower(), None) is not None
        ]

    def __await__(self) -> typing.Generator:
        print("会自动进行调用！")
        return self.dispatch().__await__()

    async def dispatch(self) -> None:
        print("dispatch")
        response = PlainTextResponse('Hello, world!')
        await response(self.scope, self.receive, self.send)


if __name__ == '__main__':
    import uvicorn
    uvicorn.run('dsasd:app')

其实HTTPEndpoint端点类就是上面一个app类而已。 如图示：

Starlette框架实现了ASGI规范中间件见类-

示例代码：

from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.endpoints import HTTPEndpoint
from starlette.routing import Route
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.cors import CORSMiddleware

class Homepage(HTTPEndpoint):
    async def get(self, request):
        return PlainTextResponse(f"Hello, world!")


class User(HTTPEndpoint):
    async def get(self, request):
        username = request.path_params['username']
        return PlainTextResponse(f"Hello, {username}")

routes = [
    Route("/", Homepage),
    Route("/{username}", User)
]

middleware = [
  Middleware(CORSMiddleware, allow_origins=['*']),
  Middleware(TrustedHostMiddleware, allowed_hosts=['example.com', '*.example.com']),
  Middleware(HTTPSRedirectMiddleware)
]
app = Starlette(routes=routes, middleware=middleware)
if __name__ == '__main__':
    import uvicorn
    uvicorn.run('main:app')

代码说明：

  • 上面的代码中主要是引入middleware中间件列表，里面注册了CORSMiddleware、TrustedHostMiddleware和HTTPSRedirectMiddleware中间件

其中我们查看一下跨域中间件处理CORSMiddleware源码（只看关键的地方）：

其中中间件也实现的ASGI协议规范，且是在__call__进行想跟处理的， 这个和前面说我们的定义一个类的方式实现协议规范一样，如下代码示例：


class app:
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        response = PlainTextResponse('Hello, world!')
        await response(scope, receive, send)

if __name__ == '__main__':
    import uvicorn
    uvicorn.run('dsasd:app')

所以又侧面证明了我们的中间件其实也是一个实现的ASGI协议规范而定对象而已！！没啥！！！

总结

  • 只要实现的ASGI协议规范都都被处理，当时显示的方式是套娃调用实现。
  • 路由也罢，中间件也罢其实都是实现的ASGI协议规范对象，只不过在一个框架中他们怕处于不同的层级，扮演的角色层次不一样。
  • 只要实现的ASGI协议规范内部，我们都可以通过解析对应的请求参数和封装具体的响应报文参数来想跟结果转回给我们的asgi服务器。

预告

下一篇开始进行断点调试一个包含了路由和中间件一个请求处理全过程是怎么样！！

以上仅仅是个人结合自己的实际需求，做学习的实践笔记！如有笔误！欢迎批评指正！感谢各位大佬！

结尾

END

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

公众号：微信搜【程序员小钟同学】

小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822


原文作者：程序员小钟同学；如有侵权，请联系管理员删除。

0
0
上一篇： fastapi框架源码分析系列-starlette框架之Debug方式学源码（2）-ASGI协议套娃图解
下一篇： fastapi框架系列- 新开线程用全局request遇到LookupError:
请登录后发表评论 登录
全部评论
程序员小钟同学

注册时间：2021-02-20

  • 博文量
    224
  • 访问量
    111686

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593