Python使用Web框架Flask开发项目

下面我会详细地讲解如何使用Python的Web框架Flask进行项目开发。

安装Flask

首先，在开始使用Flask之前，需要先安装Flask。可以使用pip安装Flask，方法如下：

pip install Flask

创建Flask应用

接下来，我们需要创建一个Flask应用，可以通过下面的代码创建一个最简单的Flask应用：

from flask import Flask

app = Flask(__name__)

@app.route('/')
def index():
    return 'Hello, Flask!'

这段代码创建了一个Flask应用，并在根路由下定义了一个简单的视图函数index()，用于返回一个字符串。

运行Flask应用

Flask应用创建完成后，需要运行应用才能在浏览器中看到效果。运行Flask应用的代码如下：

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

在运行Flask应用之前，需要在命令行中导出FLASK_APP环境变量。假设应用代码保存在app.py文件中，那么运行命令如下：

export FLASK_APP=app.py

然后就可以在命令行中运行Flask应用：

flask run

Flask应用的默认端口是5000，可以通过访问http://localhost:5000/来查看效果。

动态路由

除了根路由之外，Flask应用还可以定义其他路由，如下所示：

@app.route('/user/<username>')
def show_user_profile(username):
    return 'User %s' % username

@app.route('/post/<int:post_id>')
def show_post(post_id):
    return 'Post %d' % post_id

这里定义了两个动态路由，可以匹配类似/user/john和/post/123这样的URL。其中，和是动态参数，可以在视图函数中使用。

使用模板

Flask应用通常需要渲染HTML模板，Flask默认支持Jinja2模板引擎。可以使用Jinja2模板引擎渲染模板，如下所示：

from flask import render_template

@app.route('/hello/')
@app.route('/hello/<name>')
def hello(name=None):
    return render_template('hello.html', name=name)

例如，在templates文件夹下定义一个hello.html文件，包含如下代码：

{% if name %}
    <h1>Hello, {{ name }}!</h1>
{% else %}
    <h1>Hello, World!</h1>
{% endif %}

然后，浏览器访问http://localhost:5000/hello/john，即可以看到页面上输出Hello, john!。

使用数据库

除了渲染模板，Flask应用还可以与数据库交互。Flask支持多种数据库，如SQLite、MySQL、PostgreSQL等。以下是使用SQLite数据库的示例代码：

import sqlite3
from flask import g

DATABASE = '/tmp/flaskr.db'

def get_db():
    if not hasattr(g, 'sqlite_db'):
        g.sqlite_db = sqlite3.connect(DATABASE)
        g.sqlite_db.row_factory = sqlite3.Row
    return g.sqlite_db

@app.teardown_appcontext
def close_db(error):
    if hasattr(g, 'sqlite_db'):
        g.sqlite_db.close()

上述代码定义了一个get_db()函数，用于获取数据库连接。其中，g是Flask应用上下文对象，可用于共享数据。另外，还定义了一个close_db()函数，用于关闭数据库连接。

示例1：博客应用

下面我来举一个简单的示例，说明如何使用Flask开发一个博客应用。

创建数据库

首先，需要创建一个SQLite数据库，保存文章的标题和内容。可以使用下面的SQL语句创建表：

create table entries (
  id integer primary key autoincrement,
  title string not null,
  content string not null
);

编写模板

然后，需要编写一个模板用于渲染文章列表和文章详情页面。可以在templates文件夹下编写一个base.html和一个index.html模板。

base.html模板如下所示：

<!doctype html>
<html>
  <head>
    <title>{% block title %}{% endblock %}</title>
  </head>
  <body>
    <div class="content">
      {% block content %}{% endblock %}
    </div>
  </body>
</html>

index.html模板如下所示：

{% extends "base.html" %}
{% block content %}
  <h1>Article List</h1>
  {% for entry in entries %}
    <h2><a href="{{ url_for('entry', id=entry['id']) }}">{{ entry['title'] }}</a></h2>
    <p>{{ entry['content'] }}</p>
  {% endfor %}
{% endblock %}

编写视图函数

然后，需要编写视图函数，用于获取文章数据，渲染模板并处理表单提交。可以使用下面的代码实现。

from flask import Flask, request, session, g, redirect, url_for, \
     abort, render_template, flash, jsonify
import sqlite3

app = Flask(__name__)

# 数据库配置
DATABASE = './entries.db'
DEBUG = True
SECRET_KEY = 'development key'
USERNAME = 'admin'
PASSWORD = 'default'

app.config.from_object(__name__)

def connect_db():
    return sqlite3.connect(app.config['DATABASE'])

# 创建数据库表
def init_db():
    with app.app_context():
        db = get_db()
        with app.open_resource('schema.sql', mode='r') as f:
            db.cursor().executescript(f.read())
        db.commit()

# 获取数据库连接
def get_db():
    if not hasattr(g, 'sqlite_db'):
        g.sqlite_db = connect_db()
        g.sqlite_db.row_factory = sqlite3.Row
    return g.sqlite_db

# 关闭连接
@app.teardown_appcontext
def close_db(error):
    if hasattr(g, 'sqlite_db'):
        g.sqlite_db.close()

# 首页，显示文章列表
@app.route('/')
def index():
    db = get_db()
    cur = db.execute('select title, content from entries order by id desc')
    entries = cur.fetchall()
    return render_template('index.html', entries=entries)

# 显示文章详情
@app.route('/entry/<int:id>')
def entry(id):
    db = get_db()
    cur = db.execute('select title, content from entries where id = ?', [id])
    entry = cur.fetchone()
    if entry is None:
        abort(404)
    return render_template('entry.html', entry=entry)

# 添加文章
@app.route('/add', methods=['POST'])
def add():
    if not session.get('logged_in'):
        abort(401)
    db = get_db()
    db.execute('insert into entries (title, content) values (?, ?)',
                 [request.form['title'], request.form['content']])
    db.commit()
    flash('New entry was successfully posted')
    return redirect(url_for('index'))

# 登录
@app.route('/login', methods=['GET', 'POST'])
def login():
    error = None
    if request.method == 'POST':
        if request.form['username'] != app.config['USERNAME']:
            error = 'Invalid username'
        elif request.form['password'] != app.config['PASSWORD']:
            error = 'Invalid password'
        else:
            session['logged_in'] = True
            flash('You were logged in')
            return redirect(url_for('index'))
    return render_template('login.html', error=error)

# 注销
@app.route('/logout')
def logout():
    session.pop('logged_in', None)
    flash('You were logged out')
    return redirect(url_for('index'))

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

这段代码定义了4个路由，分别用于显示文章列表，显示文章详情，添加文章和登录。

启动应用

最后，使用flask run命令启动应用，访问http://localhost:5000即可查看效果。

示例2：电影推荐应用

下面我来举另一个简单的示例，说明如何使用Flask开发一个电影推荐应用。

数据集和模型

首先，需要准备一个电影的评分数据集和基于该数据集训练的推荐模型。可以使用MovieLens数据集，下载地址为：https://grouplens.org/datasets/movielens/

基于MovieLens数据集可以训练出多种推荐模型，如基于用户的协同过滤、基于物品的协同过滤、矩阵分解等。这里我们使用矩阵分解模型，代码如下所示：

import numpy as np
import pandas as pd
import os
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error

class MatrixFactorization:
    def __init__(self, n_factors=10, alpha=0.01, lamda=0.1, n_epochs=10, random_state=42):
        self.n_factors = n_factors
        self.alpha = alpha
        self.lamda = lamda
        self.n_epochs = n_epochs
        self.random_state = random_state

    def fit(self, X):
        users = X['userId'].unique()
        items = X['movieId'].unique()
        n_users = users.shape[0]
        n_items = items.shape[0]
        user_idx = dict(zip(users, range(n_users)))
        item_idx = dict(zip(items, range(n_items)))

        X['userId'] = X['userId'].apply(lambda x: user_idx[x])
        X['movieId'] = X['movieId'].apply(lambda x: item_idx[x])

        U = np.random.normal(size=(n_users, self.n_factors))
        V = np.random.normal(size=(n_items, self.n_factors))
        for epoch in range(self.n_epochs):
            train_X, valid_X = train_test_split(X)
            for i, x in train_X.iterrows():
                u = x['userId']
                v = x['movieId']
                r = x['rating']
                error = r - np.dot(U[u], V[v])
                U[u] += self.alpha * (error * V[v] - self.lamda * U[u])
                V[v] += self.alpha * (error * U[u] - self.lamda * V[v])
            y_true = valid_X['rating'].values
            y_pred = np.array([np.dot(U[x['userId']], V[x['movieId']]) for i, x in valid_X.iterrows()])
            rmse = np.sqrt(mean_squared_error(y_true, y_pred))
            print('Epoch:', epoch, 'RMSE:', rmse)
        self.U = U
        self.V = V

    def predict(self, user, item):
        return np.dot(self.U[user], self.V[item])

以上代码定义了一个MatrixFactorization类，用于训练矩阵分解模型。

Flask应用

接下来，我们需要使用Flask构建一个推荐应用，并提供用户界面和推荐API。

应用的文件夹结构如下所示：

movie_recommender/
|- app/
|  |- templates/
|  |  |- index.html
|  |- __init__.py
|- data/
|  |- ratings.csv
|  |- movies.csv
|- model/
|  |- model.pkl
|- tests/
|  |- test_model.py
|- Dockerfile
|- docker-compose.yml
|- requirements.txt
|- run.py

其中，app文件夹下包含应用的代码，data文件夹下包含训练数据，model文件夹下包含训练好的模型。

编写视图函数

下面的代码定义了一个推荐应用，提供电影评分界面和基于矩阵分解模型的电影推荐API。

from flask import Flask, request, jsonify, render_template
import pandas as pd
import numpy as np
import joblib
import os
from app.my_model import MatrixFactorization

app = Flask(__name__)

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/predict', methods=['POST'])
def predict():
    userId = int(request.form['userId'])
    movieId = int(request.form['movieId'])
    rating = float(request.form['rating'])
    print(userId, movieId, rating)

    # 加载模型
    model_path = os.path.join(os.getcwd(), 'model', 'model.pkl')
    model = joblib.load(model_path)

    # 预测评分
    score = model.predict(userId, movieId)
    print(score)

    # 更新模型
    data_path = os.path.join(os.getcwd(), 'data', 'ratings.csv')
    X = pd.read_csv(data_path)
    X = X.append({'userId': userId, 'movieId': movieId, 'rating': rating}, ignore_index=True)
    model.fit(X)
    joblib.dump(model, model_path)

    # 返回结果
    return jsonify({'score': round(score, 2)})

以上的代码定义了两个路由，一个用于渲染页面，在页面中展示电影列表和评分表单；另一个用于接收表单提交，预测评分并更新模型。

编写模型训练脚本

训练矩阵分解模型的代码如下所示：

import pandas as pd
import numpy as np
import joblib
import os
from app.my_model import MatrixFactorization

# 加载数据
data_path = os.path.join(os.getcwd(), 'data', 'ratings.csv')
X = pd.read_csv(data_path)

# 训练模型
model = MatrixFactorization(n_factors=10, alpha=0.01, lamda=0.1, n_epochs=10)
model.fit(X)

# 保存模型
model_path = os.path.join(os.getcwd(), 'model', 'model.pkl')
joblib.dump(model, model_path)

以上代码定义了一个脚本，用于加载数据，训练模型并保存训练好的模型。

总结

以上就是使用Flask开发项目的完整攻略。Flask是一个轻量的Web框架，易于上手，支持多种数据库和模板引擎，并且非常灵活。使用Flask，可以快速实现各种Web应用，也可以作为学习Web开发的入门工具。

安装Flask

创建Flask应用

运行Flask应用

动态路由

使用模板

使用数据库

示例1：博客应用

创建数据库

编写模板

编写视图函数

启动应用

示例2：电影推荐应用

数据集和模型

Flask应用

编写视图函数

编写模型训练脚本

总结

你可能也喜欢

深入理解Python中的元类(metaclass)

详解Python 函子和应用型函子

Python3 单行多行万能正则匹配方法