# GET方法 StreamingResponse





服务器：

```clike
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from starlette.responses import FileResponse
from pydantic import BaseModel
app = FastAPI()
# 创建一个 Base 模型，用于表示其他的 JSON 信息
class AdditionalInfo(BaseModel):
    message: str
    status: str
@app.get("/get_demo_image_with_json")
async def get_demo_image_with_json():
    # 从文件中读取字节流
    file_path = "face.png"
    file_like = open(file_path, mode="rb")
    # 模拟其他的 JSON 信息
    json_info = AdditionalInfo(message="Image loaded successfully", status="OK")
    # 使用 StreamingResponse 返回字节流和其他的 JSON 信息
    return StreamingResponse(file_like, media_type="image/jpeg",
                             headers={"Additional-Info": json_info.model_dump_json()})
if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
```

客户端：

```clike
import requests
url = "http://127.0.0.1:8000/get_demo_image_with_json"
response = requests.get(url)
# 检查请求是否成功
if response.status_code == 200:
    # 获取文件的字节流
    image_data = response.content
    # 处理其他的 JSON 信息
    additional_info = response.headers.get("Additional-Info")
    # 在此处添加您的处理逻辑，例如保存字节流到文件，解析 JSON 信息等
    # ...
    print("Image loaded successfully.")
    print(f"Additional Info: {additional_info}")
else:
    print(f"Failed to fetch image. Status code: {response.status_code}")
```

# POST方法 StreamingResponse

服务器代码：

```clike
import io
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
app = FastAPI()
# 创建一个 Base 模型，用于表示其他的 JSON 信息
class AdditionalInfo(BaseModel):
    message: str
    status: str
@app.post("/get_demo_image_with_json")
async def get_demo_image_with_json():
    #  读取face.png
    image_data = open("face.png", "rb").read()
    # 模拟其他的 JSON 信息
    json_info = AdditionalInfo(message="Image loaded successfully", status="OK")
    # 使用 StreamingResponse 返回字节流和其他的 JSON 信息
    return StreamingResponse(io.BytesIO(image_data), media_type="image/png",
                             headers={"Additional-Info": json_info.model_dump_json()})
if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
```

客户端参数：

```clike
import requests
url = "http://127.0.0.1:8000/get_demo_image_with_json"
response = requests.post(url)
# 检查请求是否成功
if response.status_code == 200:
    # 获取文件的字节流
    image_data = response.content
    # 写入文件
    with open("demo_image_with_json.png", "wb") as fp:
        fp.write(image_data)
    # 打印其他参数
    print(response.headers["Additional-Info"])
else:
    print(f"Failed to fetch image. Status code: {response.status_code}")
```

# 其他

还有FileResponse、base64。

FileResponse太肤浅，base64对于大文件来说太大。


在 FastAPI 中，返回文件字节流的主要方式包括使用 `StreamingResponse` 和 `FileResponse`。这两者都可以用于返回二进制数据，例如图像文件。

1. **StreamingResponse:** 适用于以流式方式发送数据，对于大型文件特别有用，因为它允许在数据生成时就开始发送，而不必等到整个数据集都可用。

    ```python
    from fastapi.responses import StreamingResponse
    @app.get("/get_demo_image")
    async def get_demo_image():
        image_data = open("face.png", "rb").read()
        return StreamingResponse(io.BytesIO(image_data), media_type="image/png")
    ```

2. **FileResponse:** 适用于返回文件，可以从文件系统路径中读取文件内容，也可以通过 `content` 参数直接传递文件内容。

    ```python
    from fastapi.responses import FileResponse
    @app.get("/get_demo_image")
    async def get_demo_image():
        image_data = open("face.png", "rb").read()
        return FileResponse(content=image_data, media_type="image/png")
    ```

这两种方法都是有效的，并且具体的选择可能取决于你的应用程序的需求和性能考虑。如果你希望以异步方式发送文件，你可能会更喜欢 `StreamingResponse`。如果你只是从文件系统中返回文件，`FileResponse` 是一个更简单的选择。

# 上传文件，返回文件的一个示例程序

```clike
from fastapi.responses import StreamingResponse
from fastapi import Depends, FastAPI, Header, Query, Response, UploadFile, Form
# 字节流
@app.post("/image_upscaler_bytes", summary="Image Upscaler Bytes",
          responses={image_upscaler_CustomErrorResponse.error_code:
                         image_upscaler_CustomErrorResponse.to_response_dict()})
def image_upscaler_bytes(image: UploadFile = File(..., title="图像字节流", description="图像字节流"),
                         outscale: float = Form(2, ge=1.0, le=5.0, title="图像超分倍率",
                                                description="图像超分倍率"),
                         requestId: str = Form("", title="Request Id", description="requestId to be returned")):
    """
    Image Upscaler.
    """
    try:
        img = cv2.imdecode(np.fromstring(image.file.read(), np.uint8), cv2.IMREAD_COLOR)
        output, _ = upsampler.enhance(img, outscale=outscale)
        # output = enhance_image(img, outscale=outscale)
        if output is None:
            return image_upscaler_CustomErrorResponse()
        else:
            # 返回字节流
            return StreamingResponse(io.BytesIO(cv2.imencode('.png', output)[1].tobytes()), media_type="image/png",
                                     headers={'requestId': requestId})
    except:
        traceback.print_exc()
        return image_upscaler_CustomErrorResponse(str(traceback.format_exc()))
```

# 关于压缩

```clike
FastAPI本身没有直接支持响应内容压缩的中间件，但你可以通过使用 Starlette 的中间件来实现这一功能。具体来说，Starlette 提供了 `Middleware` 类，你可以使用它来定义自定义中间件。以下是一个简单的例子，演示如何使用 Gzip 中间件来压缩响应内容：
```python

from fastapi import FastAPI

from fastapi.responses import StreamingResponse

from starlette.responses import FileResponse

from starlette.middleware.gzip import GZipMiddleware

from pydantic import BaseModel

app = FastAPI()

# 使用 Gzip 中间件

app.add_middleware(GZipMiddleware, minimum_size=1000, compress_level=6)

# 创建一个 Base 模型，用于表示其他的 JSON 信息

class AdditionalInfo(BaseModel):

    message: str

    status: str

@app.get("/get_demo_image_with_json")

async def get_demo_image_with_json():

    # 从文件中读取字节流

    file_path = "face.png"

    file_like = open(file_path, mode="rb")

    # 模拟其他的 JSON 信息

    json_info = AdditionalInfo(message="Image loaded successfully", status="OK")

    # 使用 StreamingResponse 返回字节流和其他的 JSON 信息

    return StreamingResponse(file_like, media_type="image/jpeg",

                             headers={"Additional-Info": json_info.json()})
```

在上述代码中，通过添加 `GZipMiddleware` 到 FastAPI 应用中，你启用了 Gzip 压缩。在 `StreamingResponse` 中返回的字节流会在传输过程中被压缩。请注意，Gzip 压缩可能会增加 CPU 使用，但通常可以显著减小传输的数据量，提高性能。你可以根据需求调整 `minimum_size` 和 `compress_level` 参数。

FastAPI如何返回文件字节流？并且附带一些json参数

问题：





hon -I/tmp/pip-build-env-eoaar12h/overlay/lib/python3.10/site-packages/numpy/core/include -I/home/xd/stable-diffusion-webui/venv/include -I/usr/include/python3.10 -c insightface/thirdparty/face3d/mesh/cython/mesh_core.cpp -o build/temp.linux-x86_64-cpython-310/insightface/thirdparty/face3d/mesh/cython/mesh_core.o

      x86_64-linux-gnu-gcc -Wno-unused-result -Wsign-compare -DNDEBUG -g -fwrapv -O2 -Wall -g -fstack-protector-strong -Wformat -Werror=format-security -g -fwrapv -O2 -fPIC -Iinsightface/thirdparty/face3d/mesh/cython -I/tmp/pip-build-env-eoaar12h/overlay/lib/python3.10/site-packages/numpy/core/include -I/home/xd/stable-diffusion-webui/venv/include -I/usr/include/python3.10 -c insightface/thirdparty/face3d/mesh/cython/mesh_core_cython.cpp -o build/temp.linux-x86_64-cpython-310/insightface/thirdparty/face3d/mesh/cython/mesh_core_cython.o

      insightface/thirdparty/face3d/mesh/cython/mesh_core_cython.cpp:36:10: fatal error: Python.h: No such file or directory

         36 | #include "Python.h"

            |          ^~~~~~~~~~

      compilation terminated.

      error: command '/usr/bin/x86_64-linux-gnu-gcc' failed with exit code 1

      [end of output]

  note: This error originates from a subprocess, and is likely not a problem with pip.

  ERROR: Failed building wheel for insightface

Failed to build insightface

ERROR: Could not build wheels for insightface, which is required to install pyproject.toml-based projects



It looks like there is a compilation error while trying to build the `insightface` package, and the error is related to the absence of the `Python.h` file. This file is part of the Python development headers and is needed to build Python C extensions.


解决：

To resolve this issue, you need to install the Python development headers. The process to install these headers varies depending on your operating system.


### Ubuntu/Debian Linux:

```bash
sudo apt-get install python3-dev
```

！！！！！！！！！！！！！！！！！！

<mark>我是python3.10，我后面尝试了sudo apt-get install python3.10-dev，非常好用。

### Red Hat/Fedora Linux:

```bash
sudo dnf install python3-devel
```

### macOS:

```bash
xcode-select --install
```

After installing the Python development headers, try installing the `insightface` package again:

```bash
pip install insightface
```

This should resolve the issue you're facing. If you encounter any additional errors or have specific requirements for the `insightface` package, you may need to install additional dependencies as indicated in the package documentation.

安装 insightface Python.h: No such file or directory 36 | #include “Python.h“

根源就是js和ccs文件访问路由的问题，首先你要有本地的文件，详情看[https://qq742971636.blog.csdn.net/article/details/134587010](https://qq742971636.blog.csdn.net/article/details/134587010)。






其次，你需要这么写：

/unicontorlblip就是我配置的mLB网关路由。

```python
app = FastAPI(
    title='outpainting_captioning_upscaler',
    description='outpainting images, captioning images,upscaler images',
    version='1.0.0',
    docs_url=None,
    redoc_url=None,  # 设置 ReDoc 文档的路径
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html():
    return get_swagger_ui_html(
        openapi_url="/unicontorlblip/openapi.json",
        title="xx",
        # oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url,
        swagger_js_url='/unicontorlblip/static/swagger/swagger-ui-bundle.js',
        swagger_css_url='/unicontorlblip/static/swagger/swagger-ui.css',
        swagger_favicon_url='/unicontorlblip/static/swagger/img.png',
    )
```

![在这里插入图片描述](/static/img/images/03780.png)

Python，FastAPI，mLB网关，无法访问/docs

使用fastapi的时候，swagger-ui.css 、swagger-ui-bundle.js、redoc.standalone.js 有时候无法加载（国内环境原因或者是局域网屏蔽），此时就需要自己用魔法下载好对应文件，然后替换到fastapi里面去。





fastapi里面依靠这2个函数实现docs和redoc：

```bash
from fastapi.openapi.docs import get_swagger_ui_html, get_redoc_html
```

fastapi里面官网给的解决办法：

[https://github.com/tiangolo/fastapi/blob/ac93277d3b506f4076aee1af6ae5a86406f545c6/docs_src/custom_docs_ui/tutorial001.py#L4](https://github.com/tiangolo/fastapi/blob/ac93277d3b506f4076aee1af6ae5a86406f545c6/docs_src/custom_docs_ui/tutorial001.py#L4)

实践起来就是，首先下载好这些文件：

![在这里插入图片描述](/static/img/images/03782.png)

其次复现出/docs路由和/redoc路由：

```python
import os
import uvicorn
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.openapi.docs import get_swagger_ui_html, get_redoc_html
app = FastAPI(
    title='outpainting_captioning_upscaler',
    description='outpainting images, captioning images,upscaler images',
    version='1.0.0',
    docs_url=None,
    redoc_url=None,  # 设置 ReDoc 文档的路径
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html():
    return get_swagger_ui_html(
        openapi_url="/openapi.json",
        title="xx",
        # oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url,
        swagger_js_url='/static/swagger/swagger-ui-bundle.js',
        swagger_css_url='/static/swagger/swagger-ui.css',
        swagger_favicon_url='/static/swagger/img.png',
    )
@app.get("/redoc", include_in_schema=False)
async def redoc_html():
    return get_redoc_html(
        openapi_url=app.openapi_url,
        title=app.title + " - ReDoc",
        redoc_js_url="/static/swagger/redoc.standalone.js",
    )
# 写个测试接口
@app.get('/xx')
async def test():
    return {'message': 'test success'}
if __name__ == '__main__':
    uvicorn.run(f'{os.path.basename(__file__).split(".")[0]}:app',
                host='0.0.0.0',
                port=7777,
                reload=False,
                workers=1)
```

其中文件也可以去这里下载：

```bash
下载:https://wwsm.lanzouy.com/i682Q1frjnmd 密码:9zw2
```



成功：

![在这里插入图片描述](/static/img/images/03783.png)

【Python】Fastapi swagger-ui.css 、swagger-ui-bundle.js 无法加载，docs无法加载，redocs无法使用

Python





```python
import copy
import logging
import sys
class ColoredFormatter(logging.Formatter):
    COLORS = {
        "DEBUG": "\033[0;36m",  # CYAN
        "INFO": "\033[0;32m",  # GREEN
        "WARNING": "\033[0;33m",  # YELLOW
        "ERROR": "\033[0;31m",  # RED
        "CRITICAL": "\033[0;37;41m",  # WHITE ON RED
        "RESET": "\033[0m",  # RESET COLOR
    }
    def format(self, record):
        colored_record = copy.copy(record)
        levelname = colored_record.levelname
        seq = self.COLORS.get(levelname, self.COLORS["RESET"])
        colored_record.levelname = f"{seq}{levelname}{self.COLORS['RESET']}"
        return super().format(colored_record)
# Create a new logger
logger = logging.getLogger("user_server")
logger.propagate = False
# Add handler if we don't have one.
if not logger.handlers:
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(
        ColoredFormatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
    )
    logger.addHandler(handler)
# Configure logger
loglevel = getattr(logging, "INFO", "info")
logger.setLevel(loglevel)
logger.info("sdfasdasdas")
logger.warning("sdfasdasdas")
logger.error("sdfasdasdas")
```


![在这里插入图片描述](/static/img/images/03790.png)

【python】【logging】python如何使用logging模块，一个完美的logging模块

我是一个专注技术分享的博主,尤其爱搞深度学习和单片机电子。你现在看到的是我的个人博客网站。