1、flask demo

from flask import Flask, Blueprint, request
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 数据库
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:[email protected]:3306/flask'
db = SQLAlchemy(app)


class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(), unique=True)
    password = db.Column(db.String(), unique=True)

    def __init__(self, username, password):
        self.username = username
        self.password = password

    def __eq__(self, other):
        # `__eq__` is an instance method, which also accepts one other object as an argument.
        if type(other) == type(self) and other.username == self.username and other.password == self.password:
            return True
        else:
            return False

    def __str__(self):
        return "User(%s,%s)" % (self.username, self.password)


# 蓝图
bp_index = Blueprint("index_page", __name__)
bp_sql = Blueprint("sql_page", __name__)


@bp_index.route("/")
def index():
    return "hello world"


@bp_sql.route("/", methods=['GET'])
def list_users():
    from sqlalchemy import text
    sql = text("SELECT * FROM  user ")
    result = db.engine.execute(sql)
    msg = ''
    for row in result:
        msg = msg + str(row) + "\n"
    return msg


@bp_sql.route("/", methods=['POST'])
def create_user():
    user_json = request.get_json()
    username = user_json.get("username")
    password = user_json.get("password")
    user = User(username, password)
    db.session.add(user)
    db.session.commit()
    return "add success"


@bp_sql.route("/", methods=['PUT'])
def update_user():
    user_json = request.get_json()
    username = user_json.get("username")
    password = user_json.get("password")
    id = user_json.get("id")
    db.session.query(User).filter(User.id == id).update({"username": username, "password": password})
    db.session.commit()
    return "update success"


@bp_sql.route("/", methods=['DELETE'])
def delete_user():
    user_json = request.get_json()
    username = user_json.get("username")
    password = user_json.get("password")
    del_user = User(username, password)
    users = User.query.all()
    for u in users:
        if u == del_user:
            db.session.delete(u)
            db.session.commit()
    return "delete success"


app.register_blueprint(bp_index, url_prefix="/")
app.register_blueprint(bp_sql, url_prefix="/sql")

if __name__ == '__main__':
    app.run()

2、Postman测试

显示用户

添加用户

更新用户

删除用户