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

首页 > 数据库 > 数据库开发技术 > fastapi框架源码分析系列-starlette框架之Debug方式学源码（6）-Response响应报文视频流示例和源码文解

fastapi框架源码分析系列-starlette框架之Debug方式学源码（6）-Response响应报文视频流示例和源码文解

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

1 前言

上一篇中我们是单独对请求参数Request组件分解。本章主要是分解各个响应体的报文封装.

  • Response （父类）
  • HTMLResponse （文本类的HTML）
  • JSONResponse（文本类的JSON）
  • PlainTextResponse （文本类的PlainText）
  • RedirectResponse （重定向写入头部关键信息，自动重新跳转访问）
  • StreamingResponse （数据流类型的响应报文）
  • FileResponse （文件类型（数据流类型）的响应报文）

由此我们大概也能总结出一个问题点，就是我们一个请求响应完整最终点是：

Response中的__call__
所以也可以理解是一个：
await response(scope, receive, send)

1：文本类型-响应报文分析

从前面我们可以得知Response其实也是一个asgiapp协议实现对象，所以我们有额可以直接使用下面的类似示例来启动一个APP：

app2 = PlainTextResponse('Hello, world!')
if __name__ == '__main__':
    import uvicorn
    uvicorn.run('main2:app2')

随意访问任何接口此时都可以返回'Hello, world!'。或者

app2 = JSONResponse(content={'sdsd':34})
if __name__ == '__main__':
    import uvicorn
    uvicorn.run('main2:app2')

随意访问任何接口此时都可以返回'{"sdsd":34}'。或者

app2 = HTMLResponse(content={‘<div>ssssssssssssss</div>>’)
if __name__ == '__main__':
    import uvicorn
    uvicorn.run('main2:app2')

随意访问任何接口此时都可以返回'ssssssssssssss'。

所有响应报文主要是父类，其源码是：

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:
        # 如果设置响应码，则直接赋值，默认是200
        self.status_code = status_code
        # 赋值响应报文类型
        if media_type is not None:
            self.media_type = media_type
        # 是否存在后台任务
        self.background = background
        # 开始封装发到你会  self.body，里面主要是返回内容进行类型对比做对应编码处理而已
        self.body = self.render(content)
        # 根据需要设置自定义请求头信息初始化要返回请求头内容信息
        self.init_headers(headers)

    def render(self, content: typing.Any) -> bytes:
        # 如果响应报文内容是空返回空字节
        if content is None:
            return b""
        # 如果传入是bytes，则直接返回
        if isinstance(content, bytes):
            return content
        # 不是的字节数组类似的就需要编码，注意可以进行encode类型
        return content.encode(self.charset)

    def init_headers(
        self, headers: typing.Optional[typing.Mapping[str, str]] = None
    ) -> None:
        # 如果没有传入自定义而定类似的响应头
        if headers is None:
            # 设置响应头列表为空
            raw_headers: typing.List[typing.Tuple[bytes, bytes]] = []
            populate_content_length = True
            populate_content_type = True
        else:
           # 如果存在自定义响应头，则进行字典推导式的，生成列表
            raw_headers = [
                (k.lower().encode("latin-1"), v.encode("latin-1"))
                for k, v in headers.items()
            ]
            keys = [h[0] for h in raw_headers]
            # 判断头部里面是否存在 content-length等关键的KEY
            populate_content_length = b"content-length" not in keys
            # 判断头部里面是否存在 content-type等关键的KEY
            populate_content_type = b"content-type" not in keys
        
        # 判断我们而定body内容是否已经设置完成，也就是存在对应的属性
        body = getattr(self, "body", None)
        # 如果 body有值，且存在content_length键的值，状态码设置小于200或者是在204 或 304 
        #里面的胡啊，那么我们就需要根据body的长度来设置content_length的长度 ，且
        if (
            body is not None
            and populate_content_length
            and not (self.status_code < 200 or self.status_code in (204, 304))
        ):
            #那么我们就需要根据body的长度来设置content_length的长度 ，且
            content_length = str(len(body))
            # 设置具体content-length值
            raw_headers.append((b"content-length", content_length.encode("latin-1")))
        
        # 设置content_type类型
        content_type = self.media_type
        # 如果content_type有具体的值 且populate_content_type值也有
        if content_type is not None and populate_content_type:
            # 判断是不是文本类型
            if content_type.startswith("text/"):
                # 如果是文本类型的话，那么久设置对应charset
                content_type += "; charset=" + self.charset
            raw_headers.append((b"content-type", content_type.encode("latin-1")))
        # 生成最终的raw_headers
        self.raw_headers = raw_headers

    @property
    def headers(self) -> MutableHeaders:
        if not hasattr(self, "_headers"):
            self._headers = MutableHeaders(raw=self.raw_headers)
        return self._headers

    ....省略cookei的代码分析

    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,
            }
        )
        # 一次性的把self.body进行发送
        await send({"type": "http.response.body", "body": self.body})
        # 如果存在后台任务，那么久执行后台任务
        if self.background is not None:
            await self.background()

2：数据流类型-响应报文分析

由于流数据类型在框架中比较场景，比如输出文件流等，所以我们借此来具体分析一个输出数据流的过程是怎么样的一个过程，一个简单基于fastapi下的小验证输出流示例的代码：

app = FastAPI()
@app.get('/')
async def index(request: Request):
    # 设置响应类型
    content_type = 'application/octet-stream'
    # 读取文件的起始位置和终止位置
    range_str = request.headers.get('range', '')
    if range_str:
        range_str = range_str.replace("bytes=","")
    asda = b'asssssssssssssssssssssssssssssssssssssss2333333333333333333' \
           b'asdddddddddddddddddddddddddddddddddddddddd' \
           b'asdasdasd'
    returnJsonLength = stat_result_st_size = len(str(asda))
    offset = 0
    endIndex = returnJsonLength - 1
    length = returnJsonLength - offset
    if range_str and range_str.split('-')[0] != 0 and len(range_str.split('-'))>1:
        rangedata = range_str.split('-')
        offset = int(rangedata[0])
        if endIndex >= returnJsonLength:
            endIndex = returnJsonLength - 1
        length = endIndex - offset + 1
    await asyncio.sleep(2)
    # 打开文件从起始位置开始分片读取文件
    return Response(
        content=str(asda),
        media_type=content_type,
        headers={
            'accept-ranges': 'bytes',
            'connection': 'keep-alive',
            'content-length': str(returnJsonLength),
            # Content-Range:bytes[json的开始字节]-[json的结束字节]/[报文的总大小]
            # "bytes " + offset + "-" + endIndex + "/" + returnJsonLength
            'content-range': f'bytes {offset}-{endIndex}/{stat_result_st_size}',
        },
        status_code=206 if length > 0 else 200
    )

示例代码说明：

  • 上面示例主要是想测试关于分段输出流，但是仅仅测试输出的数据比较简单，所以仅仅用于了解如何进行分段流下载而已！

下面我们基于一个播放视频文件流的方式来解读一下，数据流响应报文。

测试视频文件来源：http://clips.vorwaerts-gmbh.de/big_buck_bunny.mp4

播放视频文件示例代码来源：https://github.com/SiluPanda/fastapi-video-stream-example

测试验证代码完整示例：

from fastapi import FastAPI, Header
from fastapi.responses import Response
from os import getcwd, path
from fastapi.responses import HTMLResponse, StreamingResponse

app = FastAPI()
import cv2

PORTION_SIZE = 1024 * 1024
current_directory = getcwd() + "/"
CHUNK_SIZE = 1024 * 1024
from pathlib import Path

video_path = Path("big_buck_bunny.mp4")


@app.get("/", response_class=HTMLResponse)
async def server():
    # 页面渲染
    return """
        <!DOCTYPE html>
        <html lang="en">
        <head>
            <meta charset="UTF-8">
            <meta http-equiv="X-UA-Compatible" content="IE=edge">
            <meta name="viewport" content="width=device-width, initial-scale=1.0">
            <title>Document</title>
        </head>
        <body>
            <video width="1200" controls>
                <source src="http://localhost:8000/video" type="video/mp4">
            </video>
        </body>
        </html>
    """


@app.get('/video')
def stream_video(range: str = Header(None)):
    # 判断请求头是存在range存在着说明浏览器自动进行分片请求了
    if range:
        start, end = range.replace("bytes=", "").split("-")
    else:
        # 如果没有说明是第一次请求开始播放
        start = 0
        # 计算整个视频文件的结束的位置
        end = video_path.stat().st_size - 1
    # 记录开始的位置
    start = int(start)
    # 如果没有end 数据，那么我就从开始位置读取start + CHUNK_SIZE的量，作为计算结束的
    end = int(end) if end else start + CHUNK_SIZE
    # 如果要读取已经超过了video_path.stat().st_size，那么已经是读取到末尾了，那么久直接使用整个文件最大长度,也就是我们不能读取超过video_path.stat().st_size - 1,超过取最小的，
    end = min(end, video_path.stat().st_size - 1)
    # 开始打开视频文件
    with open(video_path, "rb") as video:
        # 设置读取文件偏移量
        video.seek(start)
        # 根据计算读取多少的数据量
        data = video.read(end - start + 1)
        # 计算整个filesize的长度
        filesize = str(video_path.stat().st_size)
        # 设置响应流文件的头部信息
        headers = {
            # 需要有这个响应头
            'Content-Range': f'bytes {str(start)}-{str(end)}/{filesize}',
            'Accept-Ranges': 'bytes',
            'Content-Length': str(end - start + 1),
            'Content-Type': 'video/mp4',
        }
        # 返回的状态码
        return Response(data, status_code=206, headers=headers, media_type="video/mp4")


# ==========================================

def read_in_chunks():
    # 读取视频位置
    videoPath = "./big_buck_bunny.mp4"  # 读取视频路径
    # 打开视频
    cap = cv2.VideoCapture(videoPath)
    # 判断是否打开成功
    suc = cap.isOpened()  # 是否成功打开
    # 循环开始读取数据流
    while suc:
        # 读取数据帧
        suc, output_frame = cap.read()
        # 装在数据帧
        if output_frame is not None:
            # 把数据帧转化为图片
            _, encodedImage = cv2.imencode(".jpg", output_frame)
            # 设置播放帧的速度等待时间
            cv2.waitKey(1)
            # 迭代返回对应的数据帧
            yield (b'--frame\r\n' b'Content-Type: image/jpeg/\r\n\r\n' + bytearray(encodedImage) + b'\r\n')
        else:
            break
    # 释放
    cap.release()


@app.get("/streamvideo")
def main():
    # 迭代的方式返回流数据
    return StreamingResponse(read_in_chunks(), media_type="multipart/x-mixed-replace;boundary=frame")


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

代码说明，在上面的代码中，分别通过两种方式来实现了视频的流的分段处理。

  • 第一种：通过浏览器自带的播放器效果，进行使用Response进行请求头分片响应处理
  • 第二种：通过把把视频转化为图片结合StreamingResponse进行流输出

这里查看第一种方案时候的浏览器对应的响应图示：第一批次：

中间都一样，最后一个批次的请求大小：

StreamingResponse源码分析则： 

class StreamingResponse(Response):
    def __init__(
        self,
        content: typing.Any,
        status_code: int = 200,
        headers: typing.Optional[typing.Mapping[str, str]] = None,
        media_type: typing.Optional[str] = None,
        background: typing.Optional[BackgroundTask] = None,
    ) -> None:
        # 如果流内容信息是一个AsyncIterable异步可迭代的对象
        if isinstance(content, typing.AsyncIterable):
            # 直接赋值给当前，标识为一个可以异步迭代的对象
            self.body_iterator = content
        else:
            # 在线程里面进行安全的迭代--iterate_in_threadpool这个可以看看，内部主要一个调用的转换
            self.body_iterator = iterate_in_threadpool(content)
        # 设置状态码
        self.status_code = status_code
        # 设置响应类型
        self.media_type = self.media_type if media_type is None else media_type
        # 是否有后台任务
        self.background = background
        # 初始化对应的响应头信息
        self.init_headers(headers)

    async def listen_for_disconnect(self, receive: Receive) -> None:
        # 循环
        while True:
            # 接收要发送消息
            message = await receive()
            # 如果消息内容是连接关闭则退出
            if message["type"] == "http.disconnect":
                break

    async def stream_response(self, send: Send) -> None:
        # 开始执行发送响应头信息开始标志
        await send(
            {
                "type": "http.response.start",
                "status": self.status_code,
                "headers": self.raw_headers,
            }
        )
        # 循环进行我们响应体消息内容，进行循环的迭代进行分片发送
        async for chunk in self.body_iterator:
            # 如果要发送的chunk不是直接数据类型的数据
            if not isinstance(chunk, bytes):
                # 编码处理一下
                chunk = chunk.encode(self.charset)
            # 不断的发送字节数组数据--注意"more_body": True标记的是True表示后面还有后续的数据要发送
            await send({"type": "http.response.body", "body": chunk, "more_body": True})
        # 标记要发送的self.body已经发送完成，执行 "body": b"", "more_body": False 发送处理
        await send({"type": "http.response.body", "body": b"", "more_body": False})

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        # 创建异步执行任务组
        # anyio是一个异步兼容性api，它允许针对 它在asyncio、curio和trio上未经修改运行。
        # 它连接以下功能：
        #
        # 任务组
        # 取消
        # 线程
        # 信号处理
        # 异步文件I/O
        # 同步原语（锁、条件、事件、信号量、队列）
        # 高级网络（TCP、UDP和Unix套接字）
        async with anyio.create_task_group() as task_group:
            # 定义内部执行函数装饰器---这里其实和asyncio 中的wait 和 gather差不多
            # 当然---wait 和 gather 的区别
            # 如 gather更加高层，可以将我们task分组
            # group1 = [get_html("htttp://") for i in range()]
            # group2 = [get_html("htttp://") for i in range()]
            # loop.run_until_complete(asyncio.gather(*group1,*group2))
            async def wrap(func: typing.Callable[[], typing.Coroutine]) -> None:
                await func()
                task_group.cancel_scope.cancel()

            # 通过偏函数执行wrap里面的函数的调用，也就是执行self.stream_response() 和send 同时处理
            task_group.start_soon(wrap, partial(self.stream_response, send))
            # 监听请求连接关闭处理
            await wrap(partial(self.listen_for_disconnect, receive))
        # 执行完成所有处理再执行后台任务
        # 判断是否存在后台任务
        if self.background is not None:
            await self.background()

FileResponse源码分析则： 


class FileResponse(Response):
    chunk_size = 64 * 1024

    def __init__(
        self,
        path: typing.Union[str, "os.PathLike[str]"],
        status_code: int = 200,
        headers: typing.Optional[typing.Mapping[str, str]] = None,
        media_type: typing.Optional[str] = None,
        background: typing.Optional[BackgroundTask] = None,
        filename: typing.Optional[str] = None,
        stat_result: typing.Optional[os.stat_result] = None,
        method: typing.Optional[str] = None,
        # 处理响应内容的方式-默认是attachment 附件的方式
        content_disposition_type: str = "attachment",
    ) -> None:
        # 要发送的文件路径
        self.path = path
        # 设置响应码
        self.status_code = status_code
        # 发送文件的名称
        self.filename = filename
        # 是否只是发送头文
        self.send_header_only = method is not None and method.upper() == "HEAD"
        # 如果没设置要发送类似是什么
        if media_type is None:
            # 进行类似的猜测逻辑推理guess_type，否则默认就是text/plain
            media_type = guess_type(filename or path)[0] or "text/plain"
        # 设置最终的类型
        self.media_type = media_type
        # 对否存在后台任务
        self.background = background
        # 响应头的初始化花式
        self.init_headers(headers)
        # 如果有设置文件名称
        if self.filename is not None:
            # 进行URL转义
            content_disposition_filename = quote(self.filename)
            # 如果转义后和文件名称不一样
            if content_disposition_filename != self.filename:
                # 重新设置content_disposition如何处理响应内容
                ##content-disposition 指示如何处理响应内容一般有两种方式：
                # inline：直接在页面显示
                # attchment：以附件形式下载
                content_disposition = "{}; filename*=utf-8''{}".format(
                    content_disposition_type, content_disposition_filename
                )
            else:
                content_disposition = '{}; filename="{}"'.format(
                    content_disposition_type, self.filename
                )
            # 设置content-disposition类型
            self.headers.setdefault("content-disposition", content_disposition)
        # 类os.stat_result---获取到文件一些元数据信息用于设置响应头内容
        # 类os.stat_result的属性与内核中stat结构体的成员相对应。其中包含的重要属性主要包括：
        #st_mode，文件类型与文件模式
        #st_ino，对于UNIX系统是文件的inode
        #st_nlink，硬链接的数量
        #st_size，文件的字节大小
        #st_atime、st_mtime、st_ctime，文件时间属性
        self.stat_result = stat_result
        # 如果stat_result有数据
        if stat_result is not None:
           # 设置对应的文件响应体响应头信息
            self.set_stat_headers(stat_result)

    def set_stat_headers(self, stat_result: os.stat_result) -> None:
        # 设置响应报文的长度
        content_length = str(stat_result.st_size)
        # 设置最后修改值信息，一般用户缓存
        last_modified = formatdate(stat_result.st_mtime, usegmt=True)
        @ 最后修改前面值信息
        etag_base = str(stat_result.st_mtime) + "-" + str(stat_result.st_size)
        # Etag是服务端的一个资源的标识，在 HTTP 响应头中将其传送到客户端，主要是用于缓存
        # 其中把Last-Modified和ETags请求的http报头一起使用，这样可利用客户端（例如浏览器）的缓存。
        #因为服务器首先产生Last-Modified/Etag标记
        #服务器可在稍后使用它来判断页面是否已经被修改。
        #本质上，客户端通过将该记号传回服务器要求服务器验证其（客户端）缓存
        etag = md5_hexdigest(etag_base.encode(), usedforsecurity=False)
        
        # 设置长度
        self.headers.setdefault("content-length", content_length)
        # 设置最后更新
        self.headers.setdefault("last-modified", last_modified)
        设置最后修改
        self.headers.setdefault("etag", etag)
        # 后续前端是通过If-None-Match来发送etag的值信息来计算是否返回304状态值，还觉得使用使用缓存。

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if self.stat_result is None:
            try:
                # 异步执行获取对应的文件信息
                stat_result = await anyio.to_thread.run_sync(os.stat, self.path)
                self.set_stat_headers(stat_result)
            except FileNotFoundError:
                raise RuntimeError(f"File at path {self.path} does not exist.")
            else:
                # 设置文件模型
                mode = stat_result.st_mode
                if not stat.S_ISREG(mode):
                    raise RuntimeError(f"File at path {self.path} is not a file.")
        # 开始标记发送响应报文开始
        await send(
            {
                "type": "http.response.start",
                "status": self.status_code,
                "headers": self.raw_headers,
            }
        )
        # 如果只是发送头部信息
        if self.send_header_only:
            #那么返回空的字节数据
            await send({"type": "http.response.body", "body": b"", "more_body": False})
        else:
           # 要有内容的话，那么久开始打开文件进行读取文件
            async with await anyio.open_file(self.path, mode="rb") as file:
                # 标记后续的文件内容是一直需要读取和要发送
                more_body = True
                while more_body:
                    #读取文件内容，按大小的读取
                    chunk = await file.read(self.chunk_size)
                    #哦安短当前读取内容的长度来设置是否还有后续的内容要发送 
                    more_body = len(chunk) == self.chunk_size
                    # 开始执行发送具体的字节内容
                    await send(
                        {
                            "type": "http.response.body",
                            "body": chunk,
                            "more_body": more_body,
                        }
                    )
        # 如果存在后台任务
        if self.background is not None:
            await self.background()

RedirectResponse重定向类型的响应报文 

class RedirectResponse(Response):
    def __init__(
        self,
        url: typing.Union[str, URL],
        status_code: int = 307,
        headers: typing.Optional[typing.Mapping[str, str]] = None,
        background: typing.Optional[BackgroundTask] = None,
    ) -> None:
        super().__init__(
            content=b"", status_code=status_code, headers=headers, background=background
        )
        # 仅仅只是在响应头中添加location，此时就客户端就会自动的重定向处理
        self.headers["location"] = quote(str(url), safe=":/%#?=@[]!$&'()*+,;")

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

结尾

END

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

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

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

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


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

0
0
上一篇： fastapi框架源码分析系列-（9）HTTP协议中的Chunked transfer encoding
下一篇： fastapi框架源码分析系列-starlette框架之Debug方式学源码（5）fastapi中间件消费body消息体再次消费
请登录后发表评论 登录
全部评论
程序员小钟同学

注册时间：2021-02-20

  • 博文量
    224
  • 访问量
    111689

最新文章

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

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

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

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