Python Flask教程

Flask

Doc: https://rest-apis-flask.teclado.com/docs/course_intro/what_is_rest_api/
Github: https://github.com/tecladocode/rest-apis-flask-python

在这里插入图片描述

1. 最简单的应用

最小应用

from flask import Flask

app = Flask(__name__)

@app.route("/")
def hello_world():
    return "<p>Hello, World!</p>"

启动

flask run --host=0.0.0.0 --port=8080

代码热加载

新建一个文件，名为：.flaskenv , 在文件内开启debug模式。（需要下载doenv）库

FLASK_DEBUG=true

1.1 url传参规则

参数规则：

from markupsafe import escape

@app.route('/user/<username>')
def show_user_profile(username):
    # show the user profile for that user
    return f'User {escape(username)}'

@app.route('/post/<int:post_id>')
def show_post(post_id):
    # show the post with the given id, the id is an integer
    return f'Post {post_id}'

@app.route('/path/<path:subpath>')
def show_subpath(subpath):
    # show the subpath after /path/
    return f'Subpath {escape(subpath)}'

`string`	(default) accepts any text without a slash
`int`	accepts positive integers
`float`	accepts positive floating point values
`path`	like `string` but also accepts slashes
`uuid`	accepts UUID strings

1.2 URL访问规则

URL 规则

/root/name/ 可通过url /root/name/和/root/name访问

/root 仅可通过/root访问，不可通过/root/访问

1.3 url中区分HTTP方法

方法一

from flask import request

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        return do_the_login()
    else:
        return show_the_login_form()

方法二

@app.get('/login')
def login_get():
    return show_the_login_form()

@app.post('/login')
def login_post():
    return do_the_login()

1.4 cookies

username = request.cookies.get('username')
resp.set_cookie('username', 'the username')

1.5 重定向和报错

from flask import abort, redirect, url_for

@app.route('/')
def index():
    return redirect(url_for('login'))

@app.route('/login')
def login():
    abort(401)
    this_is_never_executed()

2. 使用docker运行

dockerfile

FROM python:3.10
EXPOSE 5000
WORKDIR /app
COPY requirments.txt .
RUN pip install -r requirments.txt
COPY . .
CMD ["flask", "run", "--host", "0.0.0.0"]

# Dockerfile ---build--> docker image ---run--> docker container

run cmd

#!/bin/bash

# build images
docker build -t flask_smorest_api .

# run images
docker run -dp 5005:5000 -w /app -v "$(pwd):/app" flask_smorest_api

3. Restful api之Flask-Smorest

restful 的插件有很多，三选一，选择了smorest，因为它使用了marshmallow。

理由如下：

容易使用和学习
可维护性和可扩展性
项目活跃度
文档和最佳练习
开发体验

Flask-RESTful
Flask-RESTX
Flask-Smorest 官网url

# pip install flask-smorest
# 设置api的title 和 访问地址
app.config["API_TITLE"] = "Stores REST API"
app.config["API_VERSION"] = "v1"
app.config["OPENAPI_VERSION"] = "3.0.3"
app.config["OPENAPI_URL_PREFIX"] = "/"
# web页面的api访问地址
app.config["OPENAPI_SWAGGER_UI_PATH"] = "/cds"
app.config["OPENAPI_SWAGGER_UI_URL"] = "https://cdn.jsdelivr.net/npm/swagger-ui-dist/"

3. 1 methodview

from flask.views import MethodView
import marshmallow as ma
from flask_smorest import Api, Blueprint, abort

@blp.route("/")
class Pets(MethodView):
    @blp.arguments(PetQueryArgsSchema, location="query")
    @blp.response(200, PetSchema(many=True))
    def get(self, args):
        """List pets"""
        return Pet.get(filters=args)

    @blp.arguments(PetSchema)
    @blp.response(201, PetSchema)
    def post(self, new_data):
        """Add a new pet"""
        item = Pet.create(**new_data)
        return item

3.2 bluepoint

bp = Blueprint('store', __name__ , description='Store API')

@bp.route('/store/<string:stroe_id>')
class store(MethodView):

    def get(self, stroe_id):
        return "================="

比如如上的蓝图，注册后访问的网址就是：

4. 数据验证之marshmallow

数据校验模块 marshmallow官网API

flask
flask-smorest
python-dotenv

语法

from marshmallow import Schema, fields

# 基础1
class PlainItemSchema(Schema):
    id = fields.Int(dump_only=True)
    name = fields.Str(required=True)
    price = fields.Float(required=True)

# 基础2
class PlainStoreSchema(Schema):
    id = fields.Int(dump_only=True)
    name = fields.Str()

# 继承基础1
class ItemSchema(PlainItemSchema):
  	# 只有入参需要 
    store_id = fields.Int(required=True, load_only=True)
    # 只有出参需要， 嵌套基础2
    store = fields.Nested(PlainStoreSchema(), dump_only=True)


class ItemUpdateSchema(Schema):
    name = fields.Str()
    price = fields.Float()


class StoreSchema(PlainStoreSchema):
    # 嵌套
    items = fields.List(fields.Nested(PlainItemSchema()), dump_only=True)

dump_only: 只有在回应http请求的时候需要，接受http请求的时候不是必须的。
requiered=true： http发起请求和接受请求都需要该参数

# 所有字段在传入和传出的时候，都是可选的。
class ItemUpdateSchema(Schema):
    name = fields.Str()
    price = fields.Float()

输入入参数

@blp.arguments(ItemUpdateSchema)

输出参数

@blp.response(200, ItemSchema)
# 输出参数改数组
@blp.response(200, ItemSchema(many=True))

5. 数据库之SQL Alchemy

sqlalchemy
flask-sqlalchemy

Flask-SQLAlchemy官网

官网-查询的实例

主键：表中可以唯一区别的列

外键：其它表中可以唯一区别的列

5. 1 Alchemy使用流程

初始化

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import DeclarativeBase

# 基类有2个，一般情况用DeclarativeBase
# ：DeclarativeBase 或 DeclarativeBaseNoMeta 。

class Base(DeclarativeBase):
  pass

db = SQLAlchemy(model_class=Base)

初始化后 db 对象允许你访问 db.Model 类来定义模型，并访问 db.session 来执行查询。SQLAlchemy 对象还需要其他参数来自定义它管理的对象。

配置扩展

下一步是将扩展连接到 Flask 应用。唯一需要的 Flask 应用配置是 SQLALCHEMY_DATABASE_URI 键。这是一个连接字符串，它告诉 SQLAlchemy 要连接到哪个数据库。

创建 Flask 应用程序对象，加载任何配置，然后通过调用 db.init_app 使用应用程序初始化 SQLAlchemy 扩展类。此示例连接到存储在应用的实例文件夹中的 SQLite 数据库。

# create the app
app = Flask(__name__)
# configure the SQLite database, relative to the app instance folder
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///project.db"
# initialize the app with the extension
db.init_app(app)

定义模型

子类 db.Model 来定义模型类。该模型将通过将 CamelCase 类名转换为 snake_case 来生成表名。

from sqlalchemy import Integer, String
from sqlalchemy.orm import Mapped, mapped_column

# 子类 db.Model 来定义模型类。该模型将通过将 CamelCase 类名转换为 snake_case 来生成表名。
class User(db.Model):
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    username: Mapped[str] = mapped_column(String, unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String)

创建表

定义所有模型和表后，调用 SQLAlchemy.create_all() 在数据库中创建表架构。这需要应用程序上下文。由于此时您不在请求中，因此请手动创建一个。

with app.app_context():
    db.create_all()

# 如果使用flask_migrate， 则不需要上面的二行代码。
from flask_migrate import Migrate
migrate = Migrate(app, db)

如果在其他模块中定义模型，则必须在调用 create_all 之前导入它们，否则 SQLAlchemy 将不知道它们。

create_all 不会更新表中的表（如果它们已在数据库中）。如果更改模型的列，请使用迁移库（如带有 Flask-Alembic 或 Flask-Migrate 的 Alembic）来生成更新数据库架构的迁移。

查询数据

@app.route("/user-by-id/<int:id>")
def user_by_id(id):
    user = db.get_or_404(User, id)
    return render_template("show_user.html", user=user)

@app.route("/user-by-username/<username>")
def user_by_username(username):
    user = db.one_or_404(db.select(User).filter_by(username=username))
    return render_template("show_user.html", user=user)

继承db.Model的类，会自动注册到数据库。

# 子类 db.Model 来定义模型类。该模型将通过将 CamelCase 类名转换为 snake_case 来生成表名。
class User(db.Model):
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    username: Mapped[str] = mapped_column(String, nullable=False)
    email: Mapped[str] = mapped_column(String)

数据表类是大驼峰，然后会自动创建小驼峰的表在数据库中。

比如User 类，会创建user比数据表。

6. 用户认证JWT

JSON Web Token（JWT）是一个非常轻巧的规范。这个规范允许我们使用JWT在两个组织之间传递安全可靠的信息。

解析: jwt (JSON WEB TOKEN) 是一个JSON格式的规范
之前学过的认证方案: session的认证方案
特点: 非常轻巧
使用jwt替代session的原因:
session是基于cookie的，所以在android和ios中，并不通用。为了统一前端认证方案，使用jwt。
备注：
我们一般说的jwt指的是jws。

jws是一个jwt的一种实现方式。
数据格式:
	header.payload.signature
	1.header(头部):头部用于描述关于该JWT的最基本的信息，例如其类型以及签名所用的算法等。 JSON内容要经Base64 编码生成字符串成为Header。
	2.payload(载荷): 可以简单的理解为我们自己要传输的数据
	3.signature(签名)
备注:
	1.JWS的主要目的是保证了数据在传输过程中不被修改，验证数据的完整性。
	2.但由于仅采用Base64对消息内容编码，因此不保证数据的不可泄露性。所以不适合用于传输敏感数据。

flask-jwt-extended

j w t-extened官网文档

# 1. 注册进flask
from flask_jwt_extended import JWTManager
# JWT的设置
# Here you can globally configure all the ways you want to allow JWTs to
# be sent to your web application. By default, this will be only headers.
app.config["JWT_TOKEN_LOCATION"] = ["headers", "cookies", "json", "query_string"]

# If true this will only allow the cookies that contain your JWTs to be sent
# over https. In production, this should always be set to True
app.config["JWT_COOKIE_SECURE"] = True

# Change this in your code!
app.config["JWT_SECRET_KEY"] = "super-secret"

jwt = JWTManager(app)
  
# 2. 创建jwt  
from flask_jwt_extended import create_access_token
access_token = create_access_token(identity=user.id)

# 3. 检验jwt
from flask_jwt_extended import jwt_required
@jwt_required()
def get(self, item_id):
        item = ItemModel.query.get_or_404(item_id)
        return item
    
# 4. 设置jwt    
from flask import Flask, jsonify
@jwt.expired_token_loader
def expired_token_callback(jwt_header, jwt_payload):
    return (
        jsonify({"message": "The token has expired.", "error": "token_expired"}),
        401,
    )
    
# 添加jwt
@app.route("/login")
def login():
    response = jsonify({"msg": "login successful"})
    access_token = create_access_token(identity="example_user")
    set_access_cookies(response, access_token)
    return response

# 删除jwt
@app.route("/logout")
def logout():
    response = jsonify({"msg": "logout successful"})
    unset_jwt_cookies(response)
    return response

@jwt.invalid_token_loader
def invalid_token_callback(error):
    return (
        jsonify(
            {"message": "Signature verification failed.", "error": "invalid_token"}
        ),
        401,
    )

@jwt.unauthorized_loader
def missing_token_callback(error):
    return (
        jsonify(
            {
                "description": "Request does not contain an access token.",
                "error": "authorization_required",
            }
        ),
        401,
    )

7. 数据库迁移工具Flask-migrate

Flask- migrate的官方文档

pip install flask-migrate

from flask_migrate import Migrate
db.init_app(app)
migrate = Migrate(app, db)

# 不在需要下面这二行
#with app.app_context():
#    db.create_all()

# 初始化 migrations 文件夹创建一个迁移环境：
flask db init
# 生成迁移的中间文件 自动生成迁移脚本：
flask db migrate -m "add note timestamp"
# 执行迁移文件，作用与数据库 upgrade子命令即可更新数据库：
flask db upgrade


# 其它相关指令可以参考： 
flask db --help

8. Deploy reset apis

9. task queues and rq and e-mail

部署

Gunicorn

总结

flask
flask-smorest
python-dotenv
sqlalchemy
flask-sqlalchemy
flask-jwt-extended
passlib
flask-migrate
gunicorn

常用插件

在 Flask 网页开发中，有许多常用的插件可以增强 Flask 应用程序的功能和用户体验。以下是一些常用的 Flask 插件：

Flask-SQLAlchemy：用于管理数据库连接和模型定义的插件。它简化了数据库操作，并提供了方便的 ORM 功能。
Flask-WTF：提供了表单验证和提交功能的插件。它支持多种表单控件，如文本框、单选框、复选框等。
Flask-Login：用于用户身份验证和会话管理的插件。它提供了方便的函数和装饰器，用于处理用户登录和注销操作。
Flask-Bcrypt：用于密码哈希和验证的插件。它使用 Bcrypt 算法对密码进行哈希，提高了密码的安全性。
Flask-Mail：用于发送电子邮件的插件。它提供了简单的接口，用于发送 HTML 或文本邮件。
Flask-DebugToolbar：一个调试工具栏插件，可以在开发时提供有用的应用程序信息。它可以帮助开发人员诊断和解决应用程序问题。
Flask-Assets：用于管理静态文件和编译静态资源的插件。它支持多种静态文件类型，如 CSS、JavaScript、图片等。
Flask-Cache：用于缓存的插件。它可以提高应用程序的性能，并减少对数据库和服务器的请求。
Flask-Migrate：用于数据库迁移的插件。它提供了方便的接口，用于管理数据库模型的版本控制和迁移。

knowlege

http verbs

get: get a item or a list or items.
post: create an item
Delete: delete an item
Put: update an item

http head

Content-type: media type that slient sent to server,
- no default value,will put error when no value setted.
- application/json
Accept: media types that client willing to accpet
- Default is json