Flask 项目模板

,

经常会用到 flask 框架，因此想做一个模板，方便以后直接进行开发使用。

大体是使用蓝图设计一个项目模板，让其配置文件、数据库文件等单独存放，提升项目可读性。

完整代码：https://github.com/HuTa0kj/flask-template

项目目录及结构

模板项目目录如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
run.py
app
   |__init__.py           蓝图注册文件
   |config.py             配置文件，如 SECRET_KEY、数据库账号密码
   |database.py           数据库文件，放置数据库表对象
   |extensions.py         拓展文件，导入拓展包（flask_sqlalchemy、flask_login），并实例化
   |models.py             模型文件，放置如用户登录模型之类的非数据库模型
   |tasks.py              celery 异步任务
   |static|               静态文件夹
          |bootstrap|     bootstrap 静态文件目录
                    |css
                    |js
   |templates             模板文件夹，如项目较大，可以设置二级目录
             |user        user 的模板
             |admin       admin 的模板
   |views                 视图函数文件夹
          |__init__.py    空文件夹，包占位
          |user           user 视图函数，可以根据对应功能创建，并在蓝图文件中注册
          |admin          admin 视图函数
   |data|
        |mysql-volume     MySQL 挂载目录
        |redis-volume     Redis 挂载目录

主要还是遵循了蓝图的设计模式，拆分了各个部分并存放到不同文件中。

项目代码

app.py

作为项目的启动文件，直接通过 create_app 启动，并加载 DevConfig 配置。app.py 本质上就是启动程序，因此不需要写入过多的内容，保持它专属的特性即可。

1
2
3
4
5
6
7
from app import create_app, config

app, celery = create_app(config.DevConfig)

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

__init__.py

_init_.py 文件负责导入对应的配置，完成蓝图的注册、插件模块的导入。

该文件创建了 create_app 用来创建 app 对象，并加载配置文件，如果没有则默认加载 BaseConfig。

使用 register_extensions 函数用来初始化拓展对象。需要从 extensions 中导入对应的实例化对象。

使用 register_blueprints 注册蓝图。需要从 views 下导入对应的蓝图，这里就导入了 user 和 admin 并进行注册。

load_user 为 flask_login 注册用户使用。

此外，经常会用到 celery 和 redis 构建异步消息队列，因此添加了初始化 celery 配置。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
from flask import Flask
from app.views.user.user import user
from app.models import LoginUser
from app.extensions import lm, db, celery
from app.config import BaseConfig


def create_app(config=BaseConfig):
    app = Flask(__name__)
    # 注册蓝图
    app.config.from_object(config)

    # 注册扩展（包括Celery）
    register_extensions(app)

    # 注册蓝图
    register_blueprints(app)

    celery = make_celery(app)
    celery.set_default()

    return app, celery


# 插件注册
def register_extensions(app):
    lm.init_app(app)
    # 设置登录端点
    lm.login_view = 'user.login'
    db.init_app(app)


# 初始化 Celery
def make_celery(app):
    # 使用Flask应用的配置初始化Celery
    celery.conf.update(
        broker_url=app.config["CELERY_BROKER_URL"],
        result_backend=app.config["CELERY_RESULT_BACKEND"]
    )

    class ContextTask(celery.Task):
        """为 Celery 任务提供 Flask 应用上下文"""

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    # 设置 Celery 使用应用上下文任务
    celery.Task = ContextTask
    return celery


# 蓝图注册
def register_blueprints(app):
    app.register_blueprint(user)


@lm.user_loader
def load_user(user_id):
    user = LoginUser(user_id)
    return user

config.py

config.py 文件用于导入一些配置信息，例如数据库信息、开发模式或者生产模式的配置等。

Dev 模式为开发模式，开启了 debug，关闭了 csrf 保护。

Base 模式为基础设置，配置了数据库字段、秘钥以及 redis 的 broker 和 backend 地址。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class BaseConfig(object):
    DEBUG = False
    SECRET_KEY = 'xxx'
    HOSTNAME = 'mysql'
    PORT = '3306'
    USERNAME = 'root'
    PASSWORD = '123456'
    DATABASE = 'flask_app'
    SQLALCHEMY_DATABASE_URI = f"mysql+pymysql://{USERNAME}:{PASSWORD}@{HOSTNAME}:{PORT}/{DATABASE}?charset=utf8"

    REDIS_HOST = 'redis'
    CELERY_BROKER_URL = f'redis://{REDIS_HOST}:6379/1',
    CELERY_RESULT_BACKEND = f'redis://{REDIS_HOST}:6379/2'


class DevConfig(BaseConfig):
    DEBUG = True
    ASSETS_DEBUG = True
    WTF_CSRF_ENABLED = False

database.py

database.py 主要创建数据库模型，并配置数据表的对应字段属性，通过模型处理数据库。

创建了数据表模型，这里设置了 user 表，并添加了对应字段的信息，后续有新增的数据表也需要在此文件中添加。如果哪个视图函数需要对应的数据表操作，只需要导入对应的类就可以。

1
2
3
4
5
6
7
8
9
10
11
12
13
from app.extensions import db

"""
编写数据库连接
"""


class UserTable(db.Model):
    __tablename__ = "user"
    # 主键、自增长
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(100), nullable=False)
    password = db.Column(db.String(100), nullable=False)

extensions.py

扩展导入文件，这里导入扩展 flask_login 和 flask_sqlalchemy，并实例化对象。用于管理扩展插件，但导入新拓展后需要在蓝图文件中初始化。

1
2
3
4
5
6
7
8
from celery import Celery
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager

db = SQLAlchemy()
lm = LoginManager()
# 配置 celery 启动配置
celery = Celery()

models.py

放置非数据库的模型文件，这里放入 LoginUser 是用来在用户登录时进行注册的。

1
2
3
4
5
6
7
8
9
10
11
12
13
from flask_login import UserMixin


class LoginUser(UserMixin):
    def __init__(self, user_id):
        self.id = user_id

    def get_id(self):
        return str(self.id)  # 返回id属性的字符串表示

    def is_authenticated(self):
        return True  # 这里简单返回True，表示所有用户都已经通过了身份验证

tasks.py

放置 celery 要执行的异步函数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import time
from datetime import datetime
from app.extensions import celery


@celery.task
def log_access_time(user_id):
    start_time = datetime.now()
    print(f"用户 {user_id} 访问时间: {start_time}")

    # 模拟长时间任务，延迟30秒
    time.sleep(30)

    end_time = datetime.now()
    print(f"用户 {user_id} 结束时间: {end_time}")
    return start_time, end_time

views/user/user.py

user 文件，通过 Blueprint 引入，并在蓝图文件注册，对应着 user 的一些功能。

需要数据库操作，则需要从 database 中引入模型，如果需要其他模型操作，需要从 models 下引入。

其中异步任务依赖 app.tasks 里面的异步任务函数，使用 delay 方法异步执行，并可通过 task_status/ 查询状态和结果。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
from flask import Blueprint, request, redirect, render_template, jsonify
from flask_login import login_required, login_user, logout_user, current_user
from celery.result import AsyncResult
from app.tasks import log_access_time
from app.database import UserTable
from app.models import LoginUser

user = Blueprint('user', __name__)


# 登录
@user.route('/login/', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':  # 判断是否是 POST 请求
        # 获取表单数据
        username = request.form.get('username')  # 传入表单对应输入字段的 name 值
        password = request.form.get('password')
        users = UserTable.query.filter_by(username=username).first()

        if users:
            if users.password == password:
                print(users.username)
                print(users.password)

                # 设置用户登录
                login_user(LoginUser(users.id))

                # 调用 Celery 任务，延迟执行
                task = log_access_time.delay(current_user.id)

                return jsonify({
                    'task_id': task.id,  # 返回任务ID
                    'status': 'Login successful'
                })
            else:
                return jsonify({'status': 'Invalid password'}), 400
        else:
            return jsonify({'status': 'User not found'}), 404

    return render_template('/user/login.html')


# 查询任务状态
@user.route('/task_status/<task_id>', methods=['GET'])
@login_required
def get_task_status(task_id):
    task = AsyncResult(task_id)  # 查询任务结果
    if task.state == 'PENDING':
        response = {
            'state': task.state,
            'status': 'Pending...',
        }
    elif task.state == 'SUCCESS':
        response = {
            'state': task.state,
            'result': task.result,  # 返回任务的执行结果
        }
    else:
        response = {
            'state': task.state,
            'status': str(task.info),  # 若任务失败，返回异常信息
        }
    return jsonify(response)


@user.route('/dashboard/')
@login_required
def dashboard():
    # 获取当前用户信息
    user_id = current_user.id
    user = UserTable.query.get(user_id)
    return f"欢迎 {user.username} 到你的仪表盘！"


# 注销
@user.route('/login_out/')
@login_required
def login_out():
    logout_user()
    # 刷新页面的时候，做重定向，不要直接修改
    response = redirect('/')
    return response