解析请求和处理响应

处理响应

响应纯文本

# 处理响应
from flask import Flask, make_response

app = Flask(__name__)

@app.get("/text")
def r1():
    resp = make_response("hello world")
    resp.headers["Content-Type"] = "text/plain"  # 指定内容类型为纯文本
    return resp

if __name__ == '__main__':
    app.run(host="127.0.0.1",port=5000)

HTML转义

当返回html响应时（flask的默认响应类型），任何模板参数应该进行显式转义，以防止注入攻击。

from markupsafe import escape

@app.route("/<name>")
def hello(name):
    return f"Hello, {escape(name)}"

响应JSON结构体

# 处理响应
from flask import Flask, jsonify

app = Flask(__name__)

@app.get("/json")
def r2():
    data = {
        "name": "zhangsan",
    }
    return jsonify(data), 201 # 返回参数中第二个为状态码, 默认为200

if __name__ == '__main__':
    app.run(host="127.0.0.1",port=5000)

响应文件

# 处理响应
from flask import Flask, send_from_directory

app = Flask(__name__)

@app.get("/file")
def r3():
    filepath = "demo4.py"
    # as_attachment=True 表示让浏览器直接下载文件，而不是打开
    return send_from_directory(".", path=filepath, as_attachment=True)

if __name__ == '__main__':
    app.run(host="127.0.0.1",port=5000)

响应重定向

# 处理响应
from flask import Flask, redirect

app = Flask(__name__)

@app.get("/redirect")
def r4():
    # 301: 永久重定向
    # 302: 临时重定向
    # 307: 表示请求方法不变，临时重定向
    # 308: 表示请求方法不变，永久重定向
    return redirect("http://www.baidu.com", 302)

if __name__ == '__main__':
    app.run(host="127.0.0.1",port=5000)

自定义响应头

# 处理响应
from flask import Flask, make_response

app = Flask(__name__)

@app.get("/headers")
def r5():
    resp = make_response("get headers")
    resp.headers["X-Custom-Header"] = "Custom Value"
    return resp

if __name__ == '__main__':
    app.run(host="127.0.0.1",port=5000)

响应Cookie

# 处理响应
from flask import Flask, make_response

app = Flask(__name__)

@app.get("/cookies")
def r6():
    resp = make_response("get cookies")
    resp.set_cookie("name", "zhangsan", max_age=3600)
    return resp

if __name__ == '__main__':
    app.run(host="127.0.0.1",port=5000)

规范响应类

对于前后端分离的Web服务，后端一般只会响应JSON格式。最好规范响应体内容，统一前后端调用的认知。比如：

{
    "code": 200,
    "data": {
        "content": "this is /a/1"
    },
    "msg": "success"
}

code状态码主要用于表示错误类型区间状态码。如果设计比较简单，可以直接使用HTTP的状态码。如果是一个大型系统，也可以设计一套自定义的状态码。比如：

from enum import Enum

class BizStatus(Enum):
    # custom status code
    OK = 200
    BadRequestA1 = 4001  # 请求参数异常-A情况
    BadRequestA2 = 4002  # 请求参数异常-B情况

message 字段是对当前 code 状态码错误明细的补充说明。通常不同的code状态码会有不同的message描述信息。

data 值通常代表返回的响应体内容。

以下代码定义了一个JSON响应类，api在返回的时候需要引用这个响应类。除此之外，还对404和一般异常做了统一处理，当出现这两类异常时，也会返回JSON结构的响应体。

from flask import Flask, Response
from http import HTTPStatus
from typing import Optional
import json

from datetime import datetime

app = Flask(__name__)

class CustomJSONEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, datetime):
            return o.isoformat()
        return super().default(o)

class JsonResponse(Response):
    def __init__(self, data: Optional[dict] = None, code: HTTPStatus = HTTPStatus.OK, msg: str = "success"):
        headers = dict({"Content-Type": "application/json; charset=utf-8"})
        response = json.dumps({
            "code": code.value,
            "msg": msg,
            "data": data,
        },cls=CustomJSONEncoder)
        super().__init__(response=response, status=code.value, headers=headers)

class Success(JsonResponse):
    """http status code: 200"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"{request.method} {request.path} success"
        super().__init__(code=HTTPStatus.OK, msg=msg, data=data)

class Fail(JsonResponse):
    """http status code: 500"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"Fail to {request.method} {request.path}"
        super().__init__(code=HTTPStatus.INTERNAL_SERVER_ERROR, msg=msg, data=data)

class ArgumentNotFound(JsonResponse):
    """http status code: 400"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"Argument not found when {request.method} {request.path}"
        super().__init__(code=HTTPStatus.BAD_REQUEST, msg=msg, data=data)

class ArgumentInvalid(JsonResponse):
    """http status code: 400"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"Invalid arguments when {request.method} {request.path}"
        super().__init__(code=HTTPStatus.BAD_REQUEST, msg=msg, data=data)

class AuthFailed(JsonResponse):
    """http status code: 401"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"Auth failed when {request.method} {request.path}"
        super().__init__(code=HTTPStatus.UNAUTHORIZED, msg=msg, data=data)

class ResourceConflict(JsonResponse):
    """http status code: 409"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"Resource conflict when {request.method} {request.path}"
        super().__init__(code=HTTPStatus.CONFLICT, msg=msg, data=data)

class ResourceNotFound(JsonResponse):
    """http status code: 404"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"{request.method} {request.path} not found"
        super().__init__(code=HTTPStatus.NOT_FOUND, msg=msg, data=data)

class ResourceForbidden(JsonResponse):
    """http status code: 403"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"Resource forbidden when {request.method} {request.path}"
        super().__init__(code=HTTPStatus.FORBIDDEN, msg=msg, data=data)

def error_handler_notfound(error):
    return ResourceNotFound()

def error_handler_generic(error):
    print(str(traceback.format_exc()))
    return Fail()

# 使用示例
@app.get("/")
def index():
    data = {
        "now": datetime.now(),
        "sth": "hahahaha",
    }
    return JsonResponse(data=data)

app.errorhandler(Exception)(error_handler_generic)
app.errorhandler(404)(error_handler_notfound)

if __name__ == "__main__":
    app.run(host="127.0.0.1", port=5000, debug=False)

解析请求

URL参数

from markupsafe import escape

@app.route('/user/<username>')
def show_user_profile(username):
    return f'User {escape(username)}'

# 指定url参数为整数类型
@app.route('/post/<int:post_id>')
def show_post(post_id):
    return f'Post {post_id}'

# 指定url参数为路径
@app.route('/path/<path:subpath>')
def show_subpath(subpath):
    # show the subpath after /path/
    return f'Subpath {escape(subpath)}'

URL中的参数类型还有以下几种

参数类型	描述
string	字符串类型，可以接收除/以外的字符
int	整型，可以接收能通过int()方法转换的字符
float	浮点类型
path	路径，类似string，但可以接收/
uuid	UUID类型
any	any类型，指备选值中的任何一个

any用法：用户名只能是zhangsan、lisi、wangwu中的其中一个

@app.route("/user/<any(zhangsan,lisi,wangwu):name>")
def get_user(name):
    return f'username: {name}'

如果需要设置默认参数，比如默认分页为1，这样就让url更加简洁。

@app.get("/posts/<int:post_id>")
@app.get("/posts/<int:post_id>/<int:page>")
def get_posts(post_id, page=1):
    return f'Post {post_id} page {page}'

查询参数

from flask import request

# http://127.0.0.1:5000/query?username=zhangsan&group=student
@app.get("/query")
def query_args():
    username = request.args.get('username')
    group = request.args.get('group')
    return f"username: {username}, group: {group}"

表单参数

from flask import request

username = request.form["username"]
password = request.form["password"]

JSON请求体

# 解析请求
from flask import Flask, jsonify, request

app = Flask(__name__)

@app.post("/json")
def post_json():
    try:
        req: dict = request.get_json()
    except Exception as e:
        raise Exception(f"request body is not json format, error: {e}\n") from e

    username = req.get("username")
    password = req.get("password")

    return jsonify(
        {
            "u": username,
            "p": password,
        }
    )

if __name__ == "__main__":
    app.run(host="127.0.0.1", port=5000)

文件上传

参考: https://www.cnblogs.com/geekspeng/p/14961317.html

from flask import request

@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
    if request.method == 'POST':
        f = request.files['the_file']
        f.save('/var/www/uploads/uploaded_file.txt')

请求头参数

from flask import Flask, request

app = Flask(__name__)

@app.route('/headers')
def get_headers():
    user_agent = request.headers.get('User-Agent')
    content_type = request.headers.get('Content-Type')
    return f'User-Agent: {user_agent}, Content-Type: {content_type}'

Cookie参数

from flask import Flask, request

app = Flask(__name__)

@app.route('/cookies')
def get_cookies():
    session_id = request.cookies.get('session_id')
    return f'Session ID: {session_id}'

request对象

参考 - flask官方文档 - incoming-request-data

常用属性

属性	说明
args	解析后的客户端查询参数
query_string	未解析后的客户端查询参数
url	客户端请求的url
base_url	客户端请求的url，不包含查询字符串
host_url	类似base_url
host	客户端请求时用的域名
remote_addr	客户端IP
headers	请求头
json	客户端发送的json格式请求体
is_secure	是否用HTTPS或WSS协议请求
path	请求的URL中path部分
method	请求方法
authorization	请求头中的Authorization

server side示例

from flask import Flask, request, Response
from http import HTTPStatus

app = Flask(__name__)

@app.post("/")
def index():
    print(f"args: {request.args}")
    print(f"query_string: {request.query_string}")
    print(f"form: {request.form}")
    print(f"url: {request.url}")
    print(f"base_url: {request.base_url}")
    print(f"path: {request.path}")
    print(f"method: {request.method}")
    print(f"headers: {request.headers}")
    print(f"cookies: {request.cookies}")
    print(f"host: {request.host}")
    print(f"host_url: {request.host_url}")
    print(f"remote_addr: {request.remote_addr}")
    print(f"user_agent: {request.user_agent}")
    print(f"is_secure: {request.is_secure}")
    print(f"json: {request.json}")
    print(f"auth: {request.authorization}")

    return Response(mimetype="text/plain", response="Hello World", status=HTTPStatus.OK)

if __name__ == '__main__':
    app.run(host="127.0.0.1",port=8000,debug=False)

client side

import requests
import json

url = "http://127.0.0.1:8000"

req_body = {
    "k1": "v1"
}

headers = {
    "Content-Type": "application/json"
}

cookies = {
    "username": "zhangsan"
}

resp = requests.post(url=f"{url}/?q1=s1", data=json.dumps(req_body), headers=headers, cookies=cookies)
print(resp.status_code, resp.text)