flask基本使用

app.py

from flask import g, Flask, url_for, redirect, request, render_template, session
from flask_migrate import Migrate
from exts import db
from decorations import login_required
from models import User
import config

from blueprints.user import bp as user_bp
from blueprints.device import bp as device_bp
from blueprints.data import bp as data_bp
from blueprints.register_requests import bp as register_requests_bp
from blueprints.big_data import bp as big_data_bp

app = Flask(__name__)
app.config.from_object(config)
db.init_app(app)

app.register_blueprint(user_bp)
app.register_blueprint(device_bp)
app.register_blueprint(data_bp)
app.register_blueprint(register_requests_bp)
app.register_blueprint(big_data_bp)

migrate = Migrate(app, db)


@app.before_request
def before_request():
    id_user = session.get("id_user")
    if id_user:
        try:
            user = User.query.get(id_user)
            g.user = user
        except:
            g.user = None


@app.route("/")
@login_required
def index():
    return redirect(url_for('static', filename='layui/examples/layout-admin.html'))


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

config.py

# 数据库
HOSTNAME = '127.0.0.1'
PORT = '3306'
DATABASE = 'xxx'
USERNAME = 'root'
PASSWORD = '123456'
DB_URI = 'mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)

JSON_AS_ASCII = False
SQLALCHEMY_TRACK_MODIFICATIONS = True
SQLALCHEMY_DATABASE_URI = DB_URI

SECRET_KEY = 'jbwkbjevkanvejk3412512515'

# MAX_CONTENT_LENGTH = 16 * 1024 * 1024

exts.py

from flask_sqlalchemy import SQLAlchemy
import time

db = SQLAlchemy()


def delete_sa_instance_state(data):
    del data['_sa_instance_state']
    # data["href"] = "<a href=\"edit.html?id={}\">详情</a>".format(data['id_user'])
    return data


def get_time():
    return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))


def add_name_from_user(data):
    author = data.user.name
    data = vars(data)
    data['author'] = author
    del data["user"]
    return data

models.py

from exts import db


# orm模型
class User(db.Model):
    __tablename__ = "user"
    id_user = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(45), nullable=False, unique=True)
    password = db.Column(db.String(45), nullable=False)
    name = db.Column(db.String(45), nullable=False)
    phone = db.Column(db.String(45), nullable=False)
    role = db.Column(db.String(45), nullable=False)
    power = db.Column(db.Integer, nullable=False)
    grade = db.Column(db.String(45), nullable=False)


class UserExtends(db.Model):
    __tablename__ = "user_extends"
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)

    # 外键
    id_user = db.Column(db.Integer, db.ForeignKey("user.id"))
    user = db.relationship("User", backref=db.backref("user_extends", uselist=False))  # 一对一关系


class RegisterRequests(db.Model):
    __tablename__ = "register_requests"
    id_register_requests = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(45), nullable=False, unique=True)
    password = db.Column(db.String(45), nullable=False)
    name = db.Column(db.String(45), nullable=False)
    phone = db.Column(db.String(45), nullable=False)
    role = db.Column(db.String(45), nullable=False)
    power = db.Column(db.Integer, nullable=False)
    grade = db.Column(db.String(45), nullable=False)


class Device(db.Model):
    __tablename__ = "device"
    id_device = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(45), nullable=False)
    pic = db.Column(db.String(200), nullable=False)
    time = db.Column(db.String(45), nullable=False)

    # 外键
    id_user = db.Column(db.Integer, db.ForeignKey("user.id_user"))
    user = db.relationship("User", backref="devices")


class Data(db.Model):
    __tablename__ = "data"
    id_data = db.Column(db.Integer, primary_key=True, autoincrement=True)
    title = db.Column(db.String(45), nullable=False)
    content = db.Column(db.Text, nullable=False)
    appendix = db.Column(db.String(200), nullable=False)
    appendix_name = db.Column(db.String(45), nullable=False)
    time = db.Column(db.String(45), nullable=False)

    # 外键
    id_user = db.Column(db.Integer, db.ForeignKey("user.id_user"))
    user = db.relationship("User", backref="data_s")


class BigData(db.Model):
    __tablename__ = "big_data"
    id_big_data = db.Column(db.Integer, primary_key=True, autoincrement=True)
    filename = db.Column(db.String(45), nullable=False)
    path = db.Column(db.String(200), nullable=False)
    time = db.Column(db.String(45), nullable=False)
    who_can_see = db.Column(db.Integer, nullable=False)
    size = db.Column(db.String(45), nullable=False)

    # 外键
    id_user = db.Column(db.Integer, db.ForeignKey("user.id_user"))
    user = db.relationship("User", backref="big_data_s")

decorations.py

from functools import wraps
from flask import g,redirect,url_for


def login_required(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        if hasattr(g, 'user'):
            return func(*args, **kwargs)
        else:
            # return "未登录"
            return redirect(url_for("user.login"))

    return wrapper


def root_required(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        if hasattr(g, 'user') and g.user.power <= 0:
            return func(*args, **kwargs)
        else:
            # print("权限不足")
            return "权限不足"

    return wrapper


def teacher_required(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        if hasattr(g, 'user') and g.user.power <= 1:
            return func(*args, **kwargs)
        else:
            # print("权限不足")
            return "权限不足"

    return wrapper

blueprint: a demo

user.py

from flask import g, Blueprint, redirect, url_for, request, jsonify, session
from models import User
from exts import db, delete_sa_instance_state, get_time
from sqlalchemy import or_
from decorations import root_required, login_required, teacher_required
from flask import current_app

bp = Blueprint("user", __name__, url_prefix="/user")


@bp.route("/login", methods=['GET', 'POST'])
def login():
    if request.method == 'GET':
        return redirect(url_for('static', filename='login.html'))
    else:
        username = request.values.get("username")
        password = request.values.get("password")
        user = User.query.filter_by(username=username).first()
        if user is None:
            return "没有此用户"
            # return redirect(url_for("user.login"))
        else:
            if user.password == password:
                session['id_user'] = user.id_user
                return redirect(url_for('index'))
            else:
                return "密码错误"
                # return redirect(url_for("user.login"))


@bp.route('/get_user_count', methods=['GET', 'POST'])
@teacher_required
def get_user_count():
    user_list = db.session.query(User).filter(User.power >= g.user.power).all()
    data = {"count": len(user_list)}
    return data


@bp.route('/get_user_data', methods=['GET', 'POST'])
@teacher_required
def get_user_data():
    page = request.values.get('page')
    limit = request.values.get('limit')
    paginate = db.session.query(User).filter(User.power >= g.user.power).paginate(page=int(page), per_page=int(limit))
    return jsonify({"data": list(map(delete_sa_instance_state, map(vars, paginate.items)))})


@bp.route('/user_data_is_login', methods=['GET', 'POST'])
def user_data_is_login():
    try:
        user = g.user
    except:
        return "error"
    return jsonify(delete_sa_instance_state((vars(user))))


@bp.route('/delete_user', methods=['GET', 'POST'])
@root_required
def delete_user():
    id_user = request.values.get("id_user")
    User.query.filter_by(id_user=int(id_user)).delete()
    db.session.commit()
    return "success"


@bp.route('/update_user_and_pwd', methods=['GET', 'POST'])
@login_required
def update_user_and_pwd():
    username = request.values.get("username")
    password = request.values.get("password")
    user = g.user
    user.username = username
    user.password = password
    db.session.commit()
    return "success"


@bp.route("/find_user_data_html", methods=['GET', 'POST'])
@teacher_required
def find_user_data_html():
    name = request.values.get("name")
    role = request.values.get("role")
    grade = request.values.get("grade")

    res_or = db.session.query(User).filter(or_(User.name == name, User.role == role, User.grade == grade)).all()
    str1 = ""
    for user in res_or:
        print(vars(user))
        str1 += f'''
            <tr>
            <td>{user.id_user}</td>
            <td>{user.name}</td>
            <td>{user.username}</td>
            <td>{user.phone}</td>
            <td>{user.grade}</td>
            <td>{user.role}</td>
            <td>{user.password}</td>
            </tr>
            '''
    return str1


@bp.route("/login_out", methods=['GET', 'POST'])
@login_required
def login_out():
    """删除session数据"""
    if session.get("id_user"):
        del session["id_user"]
    return "success"


@bp.route("/teacher_register", methods=['GET', 'POST'])
@root_required
def teacher_register():
    value = request.values
    user = User(username=value.get("username"),
                password=value.get("password"),
                name=value.get("name"),
                phone=value.get("phone"),
                role="教师",
                power=1,
                grade="null",
                )
    db.session.add(user)
    db.session.commit()
    return "注册成功"

flask pip依赖

flask_requirements.txt

alembic==1.5.6
asgiref==3.3.1
beautifulsoup4==4.9.3
bleach==3.3.0
blinker==1.4
certifi==2020.12.5
chardet==4.0.0
click==7.1.2
coverage==5.4
cycler==0.10.0
dnspython==2.1.0
dominate==2.6.0
email-validator==1.1.2
Faker==6.6.3
Flask
Flask-Bootstrap
Flask-Images
Flask-Login
Flask-Mail
Flask-Migrate
Flask-Moment
Flask-PageDown
Flask-Script
Flask-SQLAlchemy
Flask-Uploads
Flask-WTF
ForgeryPy==0.1
freetype-py==2.2.0
future==0.18.2
idna==2.10
image==1.5.33
imbalanced-learn==0.7.0
importlib-metadata==3.4.0
itsdangerous==1.1.0
Jinja2==2.11.3
jinja2-markdown==0.0.3
joblib==1.0.1
json-tricks==3.15.5
jsonify==0.5
Mako==1.1.4
Markdown==3.3.3
MarkupSafe==1.1.1
mccabe==0.6.1
numpy==1.19.5
olefile==0.46
packaging==20.9
Pillow==8.2.0
pillowcase==2.0.0
pyglet==1.5.14
Pygments==2.8.1
PyMySQL==1.0.2
pyparsing==2.4.7
python-dateutil==2.8.1
python-dotenv==0.15.0
python-editor==1.0.4
pytz==2021.1
requests==2.25.1
scikit-learn==0.24.1
scipy==1.5.4
six==1.15.0
soupsieve==2.2.1
SQLAlchemy==1.3.23
sqlparse==0.4.1
text-unidecode==1.3
threadpoolctl==2.1.0
typed-ast==1.4.1
typing-extensions==3.7.4.3
unrar==0.4
urllib3==1.26.3
visitor==0.1.3
webencodings==0.5.1
Werkzeug==1.0.1
wincertstore==0.2
wrapt==1.11.2
WTForms==2.3.3
zipp==3.4.0