#  FastAPI 服务





下面是代码，

`@app.post("/files/")` 路由让我们可以传文件。

`@app.post("/uploadfiles/")`路由让我们可以传多个文件。

需要保证：

（1） uvicorn.run中写的是文件名：app名，确保自己的文件名。

（2）`# python 3.8   pip install fastapi uvicorn python-multipart`

```python
# -*- coding:utf-8 -*-
import traceback
from typing import List, Optional, Union
import uvicorn
from fastapi.middleware.cors import CORSMiddleware
from fastapi import FastAPI, File, Form, UploadFile
import cv2
import numpy as np
app = FastAPI(
    title='FastAPI Tutorial',
    description='FastAPI教程',
    version='1.0.0',
    docs_url='/docs',
    redoc_url='/redocs',
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
@app.post("/files/")
async def alg_file(
        fileb: UploadFile = File(...),
        token: str = Form(...)
):
    try:
        img = cv2.imdecode(np.fromstring(fileb.file.read(), np.uint8), cv2.IMREAD_COLOR)
        return {
            "token": token,
            "filename": fileb.filename,
            "content_type": fileb.content_type,
            "size": img.shape
        }
    except:
        traceback.print_exc()
        return -1
@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile] = File(...)):
    return {"filenames": [file.filename for file in files]}
@app.post("/filesper/")
def filesper(
        file: Optional[bytes] = File(None),
        fileb: Optional[UploadFile] = File(None),
        filepath: Optional[str] = Form(None),
):
    try:
        if file:
            img = cv2.imdecode(np.frombuffer(file, dtype=np.uint8), cv2.IMREAD_COLOR)
            return {
                "size": img.shape,
                "style": "bytes"
            }
        if fileb:
            img = cv2.imdecode(np.fromstring(fileb.file.read(), np.uint8), cv2.IMREAD_COLOR)
            return {
                "filename": fileb.filename,
                "content_type": fileb.content_type,
                "size": img.shape,
                "style": "UploadFile"
            }
        if filepath:
            return {
                "str": filepath,
                "style": "str"
            }
    except:
        traceback.print_exc()
        return -1
if __name__ == '__main__':
    uvicorn.run('yolo5fastapi:app', host='0.0.0.0', port=8001, reload=False, debug=False, workers=1)
```


#  requests 测试

## requests 单一文件上传

测试上传一个图片文件：

```python
import requests
url = "http://0.0.0.0:8001/files/"
with open("/home/kevin_xie/yifeinfs/pycharm_k/yolov5_v2/kevin/picture/realcamera.jpg", "rb") as f:
    res = requests.post(url=url, data={"token": "sdas"}, files={"fileb": f}, timeout=None)
    print(res.json())
```

返回：

```python
{'token': 'sdas', 'filename': 'realcamera.jpg', 'content_type': '', 'size': [720, 1280, 3]}
```

## requests 多个文件上传

注意给文件参数的方式，元组和列表均可，可自由尝试。
```python
import requests
url = "http://0.0.0.0:8001/uploadfiles/"
with open("/home/kevin_xie/yifeinfs/pycharm_k/yolov5_v2/kevin/picture/realcamera.jpg", "rb") as f:
    res = requests.post(url=url, files=[("files", f), ("files", f), ("files", f)], timeout=None)
    print(res.json())
```

Python fastapi requests 传文件方式 上传文件 上传多个文件 UploadFile模板方式 上传图片

# 1 Flask 路由阻塞





当启动一个Flask APP如下，在同一浏览器开2个窗口访问这个服务（我的是http://172.18.43.32:8001/hello），前一个请求没返回时，另外一个窗口的相同请求不会有反应。

这是浏览器的锅，同一请求会顺序执行，此时可以用同一台电脑的不同浏览器测试，可以发现不会存在路由阻塞。
```python
import time
from flask import Flask
app = Flask(__name__)
@app.route('/hello')
def hello_world():
    print("receive !")
    time.sleep(20)
    return 'Hello World!'
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8001)
```

# 2 requests访问

- 换个方式，使用requests访问，线程池中100个线程同时执行访问：

```python
import requests
from concurrent.futures import ThreadPoolExecutor
def task():
    res = requests.request(method="GET", url="http://172.18.43.32:8001/hello", timeout=None)
executor = ThreadPoolExecutor(max_workers=100)
for i in range(100):
    executor.submit(task)
```

还是可以看见不会阻塞。


# 3 FastAPI 路由阻塞

下面的情况下，访问就会阻塞。


```python
# -*- coding:utf-8 -*-
import asyncio
import time
import uvicorn
from fastapi.middleware.cors import CORSMiddleware
from fastapi import FastAPI, File, Form, UploadFile
import cv2
import numpy as np
app = FastAPI(
    title='FastAPI Tutorial',
    description='FastAPI教程',
    version='1.0.0',
    docs_url='/docs',
    redoc_url='/redocs',
)
@app.get("/hello")
async def hl():
    print("receive !")
    time.sleep(20)
    return {
        "token": "ads"
    }
def fastweb():
    uvicorn.run('deal:app', host='0.0.0.0', port=8001, reload=False, debug=False, workers=1)
if __name__ == '__main__':
    fastweb()
```

原因：在async标识的函数中使用同步方法time.sleep(20)，这导致了一个异步程序之间的同步问题，出现了错误！这种写法本身就是错误的，加async 是想利用Python的协程机制 将路由变成异步请求， 进一步提高并发量！只要考虑加async ，那么就得好好设计函数内部， 多用一些异步方法，比如可以写成：

```python
@app.get("/hello")
async def hl():
    print("receive !")
    await asyncio.sleep(20)
    return {
        "token": "ads"
    }
```

设计合理就会极大利用起协程异步，从而极大提高并发量，设计不合理就是呆瓜。若是分不清，还是老老实实和Flask一样写法：

```python
@app.get("/hello")
def hl():
    print("receive !")
    time.sleep(20)
    return {
        "token": "ads"
    }
```

这样也是出现不了路由阻塞的。

# 4 总结：同步并发 异步并发

1 Flask 中的属于同步并发，可以无阻塞访问！但要注意别让浏览器欺骗你！

2 FastAPI 支持异步框架，支持是支持，前提是你得用好它，不然就呆瓜。用好了并发量就提升了！

3 FastAPI 中同样支持Flask 那样子的同步并发，和Flask 一样一样的，但要注意写法！

4 FastAPI 貌似比 Flask 更强大！

# 参考

java web

golang 协程

Flask FastAPI 路由阻塞 Web Python 访问同一url出现阻塞 同步 异步 并发

# 何为bytes？





是Python3的数据类型，详见：[http://c.biancheng.net/view/2175.html](http://c.biancheng.net/view/2175.html)。

# np.frombuffer

返回一维数组：

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

# cv2.imdecode

cv2.imdecode()函数从指定的内存缓存中读取数据，并把数据转换(解码)成图像格式;主要用于从网络传输数据中恢复出图像。

cv2.imencode()函数是将图片格式转换(编码)成流数据，赋值到内存缓存中;主要用于图像数据格式的压缩，方便网络传输。

# bytes转图片numpy矩阵
```python
img = cv2.imdecode(np.frombuffer(file, dtype=np.uint8), cv2.IMREAD_COLOR)
```

# 逆运算 numpy矩阵转bytes

这里的img_bytes 就是bytes的file
```python
# 对数组的图片格式进行编码
success, encoded_image = cv2.imencode(".jpg", img)
# 将数组转为bytes
img_bytes = encoded_image.tostring()
```

# bytes转图片numpy矩阵


# 完整代码：

```python
# -*- coding:utf-8 -*-
import uvicorn
from fastapi.middleware.cors import CORSMiddleware
from fastapi import FastAPI, File, Form, UploadFile
import cv2
import numpy as np
from io import BytesIO
app = FastAPI(
    title='FastAPI Tutorial',
    description='FastAPI教程',
    version='1.0.0',
    docs_url='/docs',
    redoc_url='/redocs',
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
@app.post("/files/")
async def alg_file(
        file: bytes = File(...),
        token: str = Form(...)
):
    img = cv2.imdecode(np.frombuffer(file, dtype=np.uint8), 1)
    return {
        "token": token,
        "size": img.shape
    }
if __name__ == '__main__':
    uvicorn.run('run1:app', host='0.0.0.0', port=8002, reload=True, debug=True, workers=1)
```

FastAPI bytes转矩阵 bytes转图片 bytes opencv URL传文件 POST FORM Python

Opencv 读取的图片是BGR 矩阵：




```python
import cv2 
import matplotlib.pyplot as plt
img=cv2.imread('./res.jpg')
plt.imshow(img)
plt.show()
```

plt显示方法imshow是默认矩阵是RGB的，

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

想要正确显示，需要：

```python
imgrgb= cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
plt.imshow(imgrgb)
plt.show()
```

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

Opencv Python 图片通道 BGR 转 RGB imgrgb= cv2.cvtColor(img, cv2.COLOR_BGR2RGB)

读取




```python
import cv2
import numpy as np 
img = cv2.imdecode(np.fromfile("./人物-1.jpg", dtype=np.uint8), 1) # img是矩阵
```

写入：

```python
import cv2
cv2.imencode('.jpg', img)[1].tofile(os.path.join(self.keyword, self.keyword + str(count) + ".jpg"))
```