Python - Flask - 进阶

颜熙云

2023-12-01

项目布局

flaskr/ ，一个包含应用代码和文件的 Python 包。

tests/ ，一个包含测试模块的文件夹。

venv/ ，一个 Python 虚拟环境，用于安装 Flask 和其他依赖的包。

flask-tutorial/ ├── flaskr/ │ ├── __init__.py │ ├── db.py │ ├── schema.sql │ ├── auth.py │ ├── blog.py │ ├── templates/ │ │ ├── base.html │ │ ├── auth/ │ │ │ ├── login.html │ │ │ └── register.html │ │ └── blog/ │ │ ├── create.html │ │ ├── index.html │ │ └── update.html │ └── static/ │ └── style.css ├── tests/ │ ├── conftest.py │ ├── data.sql │ ├── test_factory.py │ ├── test_db.py │ ├── test_auth.py │ └── test_blog.py ├── venv/ ├── setup.py └── MANIFEST.in

==========================================

应用设置

应用工厂

flaskr/__init__.py

import os from flask import Flask def create_app(test_config=None): # create and configure the app app = Flask(__name__, instance_relative_config=True) # 创建实例，告诉配置文件是相对路径。 app.config.from_mapping( SECRET_KEY='dev', # 被 Flask 和扩展用于保证数据安全的。 DATABASE=os.path.join(app.instance_path, 'flaskr.sqlite'), # SQLite 数据文件存放路径 ) if test_config is None: # load the instance config, if it exists, when not testing app.config.from_pyfile('config.py', silent=True) # else: # load the test config if passed in app.config.from_mapping(test_config) # ensure the instance folder exists try: os.makedirs(app.instance_path) # 确保文件夹存在 except OSError: pass from . import db db.init_app(app)

from . import auth

app.register_blueprint(auth.bp) # 导入并注册蓝图。新的代码放在工厂函数的尾部返回应用之前。

from . import blog

app.register_blueprint(blog.bp)

app.add_url_rule('/', endpoint='index') #下文的 index 视图的端点会被定义为 blog.index 。一些验证视图会指定向普通的 index 端点。

# 我们使用 app.add_url_rule() 关联端点名称 'index' 和 / URL ，这样 url_for('index') 或 url_for('blog.index') 都会有效，会生成同样的 / URL 。 return app

运行应用

export FLASK_APP=flaskr export FLASK_ENV=development flask run --host='0.0.0.0' --port=5000

==========================================

定义操作数据库

flaskr/db.py

import sqlite3 import click from flask import current_app, g from flask.cli import with_appcontext # 连接数据库 def get_db(): if 'db' not in g: # g 是一个特殊对象，独立于每一个请求。在处理请求过程中，它可以于储存可能多个函数都会用到的数据。把连接储存于其中，

g.db = sqlite3.connect( # 可以多次使用，而不用在同一个请求中每次调用 get_db 时都创建一个新的连接。 current_app.config['DATABASE'], # 一个特殊对象，该对象指向处理请求的 Flask 应用。这里使用了应用工厂，那么在其余的代码中就不会出现应用对象。

detect_types=sqlite3.PARSE_DECLTYPES #当应用创建后，在处理一个请求时， get_db 会被调用。这样就需要使用 current_app 。 ) g.db.row_factory = sqlite3.Row # 告诉连接返回类似于字典的行，这样可以通过列名称来操作数据。 return g.db def close_db(e=None): db = g.pop('db', None) if db is not None: db.close()

# 初始化数据库

def init_db():

db = get_db()

with current_app.open_resource('schema.sql') as f:

db.executescript(f.read().decode('utf8'))

@click.command('init-db') # 定义一个名为 init-db 命令行，它调用 init_db 函数，并为用户显示一个成功的消息。更多关于如何写命令行的内容请参阅 ref:cli 。

@with_appcontext

def init_db_command():

"""Clear the existing data and create new tables."""

init_db()

click.echo('Initialized the database.')

# 在应用中注册

def init_app(app):

app.teardown_appcontext(close_db)

app.cli.add_command(init_db_command)

*****************

flaskr/schema.sql

DROP TABLE IF EXISTS user;

DROP TABLE IF EXISTS post;

CREATE TABLE user (

id INTEGER PRIMARY KEY AUTOINCREMENT,

username TEXT UNIQUE NOT NULL,

password TEXT NOT NULL

);

CREATE TABLE post (

id INTEGER PRIMARY KEY AUTOINCREMENT,

author_id INTEGER NOT NULL,

created TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,

title TEXT NOT NULL,

body TEXT NOT NULL,

FOREIGN KEY (author_id) REFERENCES user (id)

);

*****************

初始化数据库文件

flask init-db

==========================================

蓝图和视图

视图是一个应用对请求进行响应的函数。 Flask 通过模型把进来的请求 URL 匹配到对应的处理视图。视图返回数据， Flask 把数据变成出去的响应。 Flask 也可以反过来，根据视图的名称和参数生成 URL 。

Blueprint 是一种组织一组相关视图及其他代码的方式。与把视图及其他代码直接注册到应用的方式不同，蓝图方式是把它们注册到蓝图，然后在工厂函数中把蓝图注册到应用。

flaskr/auth.py

import functools

from flask import (

Blueprint, flash, g, redirect, render_template, request, session, url_for

)

from werkzeug.security import check_password_hash, generate_password_hash

from flaskr.db import get_db

bp = Blueprint('auth', __name__, url_prefix='/auth') # url_prefix 会添加到所有与该蓝图关联的 URL 前面。

# 注册

@bp.route('/register', methods=('GET', 'POST'))

def register():

if request.method == 'POST':

username = request.form['username']

password = request.form['password']

db = get_db()

error = None

if not username:

error = 'Username is required.'

elif not password:

error = 'Password is required.'

elif db.execute(

'SELECT id FROM user WHERE username = ?', (username,)

).fetchone() is not None:

error = 'User {} is already registered.'.format(username)

if error is None:

db.execute(

'INSERT INTO user (username, password) VALUES (?, ?)',

(username, generate_password_hash(password)) # 为了安全原因，不能把密码明文储存在数据库中。相代替的，使用 generate_password_hash() 生成安全的哈希值并储存到数据库中。

)

db.commit()

return redirect(url_for('auth.login')) # 用户数据保存后将转到登录页面。 url_for() 根据登录视图的名称生成相应的 URL 。与写固定的 URL 相比，

#这样做的好处是如果以后需要修改该视图相应的 URL ，那么不用修改所有涉及到 URL 的代码。 redirect() 为生成的 URL 生成一个重定向响应。

flash(error) # 如果验证失败，那么会向用户显示一个出错信息。 flash() 用于储存在渲染模块时可以调用的信息。

return render_template('auth/register.html')

# 登录

@bp.route('/login', methods=('GET', 'POST'))

def login():

if request.method == 'POST':

username = request.form['username']

password = request.form['password']

db = get_db()

error = None

user = db.execute(

'SELECT * FROM user WHERE username = ?', (username,)

).fetchone()

if user is None:

error = 'Incorrect username.'

elif not check_password_hash(user['password'], password): # check_password_hash() 以相同的方式哈希提交的密码并安全的比较哈希值。如果匹配成功，那么密码就是正确的。

error = 'Incorrect password.'

if error is None:

session.clear()

session['user_id'] = user['id']

return redirect(url_for('index'))

flash(error)

return render_template('auth/login.html')

# 现在用户的 id 已被储存在 session 中，可以被后续的请求使用。请每个请求的开头，如果用户已登录，那么其用户信息应当被载入，以使其可用于其他视图。

@bp.before_app_request # 注册一个在视图函数之前运行的函数

def load_logged_in_user(): # 检查用户 id 是否已经储存在 session 中，并从数据库中获取用户数据，然后储存在 g.user 中。 g.user 的持续时间比请求要长。

user_id = session.get('user_id')

if user_id is None:

g.user = None

else:

g.user = get_db().execute(

'SELECT * FROM user WHERE id = ?', (user_id,)

).fetchone()

# 注销

@bp.route('/logout')

def logout():

session.clear()

return redirect(url_for('index'))

# 其他视图中验证

# 用户登录以后才能创建、编辑和删除博客帖子。在每个视图中可以使用装饰器来完成这个工作。

# 装饰器返回一个新的视图，该视图包含了传递给装饰器的原视图。新的函数检查用户是否已载入。如果已载入，那么就继续正常执行原视图，否则就重定向到登录页面。

def login_required(view):

@functools.wraps(view)

def wrapped_view(**kwargs):

if g.user is None:

return redirect(url_for('auth.login'))

return view(**kwargs)

return wrapped_view

==========================================

模板

# 基础布局

flaskr/templates/base.html

<!doctype html>

<title>{% block title %}{% endblock %} - Flaskr</title>

<nav>

<h1>Flaskr</h1>

<ul>

{% if g.user %}

<li><span>{{ g.user['username'] }}</span>

{% else %}

<li><a href="{{ url_for('auth.register') }}">Register</a>

{% endif %}

</ul>

</nav>

{% block header %}{% endblock %}

</header>

{% for message in get_flashed_messages() %}

<div class="flash">{{ message }}</div>

{% endfor %}

{% block content %}{% endblock %}

</section>

# 注册

flaskr/templates/auth/register.html

{% extends 'base.html' %}

{% block header %}

<h1>{% block title %}Register{% endblock %}</h1>

{% endblock %}

{% block content %}

<label for="username">Username</label>

<label for="password">Password</label>

</form>

{% endblock %}

# 登录

flaskr/templates/auth/login.html

{% extends 'base.html' %}

{% block header %}

<h1>{% block title %}Log In{% endblock %}</h1>

{% endblock %}

{% block content %}

<label for="username">Username</label>

<label for="password">Password</label>

</form>

{% endblock %}

==========================================

静态文件

flaskr/static/style.css

**************

html { font-family: sans-serif; background: #eee; padding: 1rem; }

body { max-width: 960px; margin: 0 auto; background: white; }

h1 { font-family: serif; color: #377ba8; margin: 1rem 0; }

a { color: #377ba8; }

hr { border: none; border-top: 1px solid lightgray; }

nav { background: lightgray; display: flex; align-items: center; padding: 0 0.5rem; }

nav h1 { flex: auto; margin: 0; }

nav h1 a { text-decoration: none; padding: 0.25rem 0.5rem; }

nav ul { display: flex; list-style: none; margin: 0; padding: 0; }

nav ul li a, nav ul li span, header .action { display: block; padding: 0.5rem; }

.content { padding: 0 1rem 1rem; }

.content > header { border-bottom: 1px solid lightgray; display: flex; align-items: flex-end; }

.content > header h1 { flex: auto; margin: 1rem 0 0.25rem 0; }

.flash { margin: 1em 0; padding: 1em; background: #cae6f6; border: 1px solid #377ba8; }

.post > header { display: flex; align-items: flex-end; font-size: 0.85em; }

.post > header > div:first-of-type { flex: auto; }

.post > header h1 { font-size: 1.5em; margin-bottom: 0; }

.post .about { color: slategray; font-style: italic; }

.post .body { white-space: pre-line; }

.content:last-child { margin-bottom: 0; }

.content form { margin: 1em 0; display: flex; flex-direction: column; }

.content label { font-weight: bold; margin-bottom: 0.5em; }

.content input, .content textarea { margin-bottom: 1em; }

.content textarea { min-height: 12em; resize: vertical; }

input.danger { color: #cc2f2e; }

input[type=submit] { align-self: start; min-width: 10em; }

**************

==========================================

博客蓝图

flaskr/blog.py

from flask import (

Blueprint, flash, g, redirect, render_template, request, url_for

)

from werkzeug.exceptions import abort

from flaskr.auth import login_required

from flaskr.db import get_db

# 蓝图

bp = Blueprint('blog', __name__)

# 索引

@bp.route('/')

def index():

db = get_db()

posts = db.execute(

'SELECT p.id, title, body, created, author_id, username'

' FROM post p JOIN user u ON p.author_id = u.id'

' ORDER BY created DESC'

).fetchall()

return render_template('blog/index.html', posts=posts)

# 创建

@bp.route('/create', methods=('GET', 'POST'))

@login_required

def create():

if request.method == 'POST':

title = request.form['title']

body = request.form['body']

error = None

if not title:

error = 'Title is required.'

if error is not None:

flash(error)

else:

db = get_db()

db.execute(

'INSERT INTO post (title, body, author_id)'

' VALUES (?, ?, ?)',

(title, body, g.user['id'])

)

db.commit()

return redirect(url_for('blog.index'))

return render_template('blog/create.html')

# 更新

def get_post(id, check_author=True):

post = get_db().execute(

'SELECT p.id, title, body, created, author_id, username'

' FROM post p JOIN user u ON p.author_id = u.id'

' WHERE p.id = ?',

(id,)

).fetchone()

if post is None:

abort(404, "Post id {0} doesn't exist.".format(id))

if check_author and post['author_id'] != g.user['id']:

abort(403)

return post

@bp.route('/<int:id>/update', methods=('GET', 'POST'))

@login_required

def update(id):

post = get_post(id)

if request.method == 'POST':

title = request.form['title']

body = request.form['body']

error = None

if not title:

error = 'Title is required.'

if error is not None:

flash(error)

else:

db = get_db()

db.execute(

'UPDATE post SET title = ?, body = ?'

' WHERE id = ?',

(title, body, id)

)

db.commit()

return redirect(url_for('blog.index'))

return render_template('blog/update.html', post=post)

# 删除

@bp.route('/<int:id>/delete', methods=('POST',))

@login_required

def delete(id):

get_post(id)

db = get_db()

db.execute('DELETE FROM post WHERE id = ?', (id,))

db.commit()

return redirect(url_for('blog.index'))

*************************

模板

# 索引模板

{% extends 'base.html' %}

{% block header %}

<h1>{% block title %}Posts{% endblock %}</h1>

{% if g.user %}

{% endif %}

{% endblock %}

{% block content %}

{% for post in posts %}

<div>

<h1>{{ post['title'] }}</h1>

<div class="about">by {{ post['username'] }} on {{ post['created'].strftime('%Y-%m-%d') }}</div>

</div>

{% if g.user['id'] == post['author_id'] %}

{% endif %}

</header>

</article>

{% if not loop.last %} # 它用于在每个博客帖子后面显示一条线来分隔帖子，最后一个帖子除外。

<hr>

{% endif %}

{% endfor %}

{% endblock %}

# 创建模板

{% extends 'base.html' %}

{% block header %}

<h1>{% block title %}New Post{% endblock %}</h1>

{% endblock %}

{% block content %}

<label for="title">Title</label>

<textarea name="body" id="body">{{ request.form['body'] }}</textarea>

</form>

{% endblock %}

# 更新/删除模板

{% extends 'base.html' %}

{% block header %}

<h1>{% block title %}Edit "{{ post['title'] }}"{% endblock %}</h1>

{% endblock %}

{% block content %}

<label for="title">Title</label>

<input name="title" id="title"

value="{{ request.form['title'] or post['title'] }}" required> # 选择在表单显示什么数据。当表单还未提交时，显示原 post 数据。

# 但是，如果提交了非法数据，然后需要显示这些非法数据以便于用户修改时，就显示 request.form 中的数据。

<textarea name="body" id="body">{{ request.form['body'] or post['body'] }}</textarea>

</form>

<hr>

</form>

{% endblock %}

==========================================

项目可安装化

setup.py 文件描述项目及其从属的文件。

setup.py

from setuptools import find_packages, setup

setup(

name='flaskr',

version='1.0.0',

packages=find_packages(),

include_package_data=True,

zip_safe=False,

install_requires=[

'flask',

)

MANIFEST.in

include flaskr/schema.sql

graft flaskr/static

graft flaskr/templates

global-exclude *.pyc

==========================================

测试覆盖

==========================================

部署产品

构建安装

# 安装所需 python 包

pip install wheel

# 进入到代码文件夹中，构建文件

python setup.py bdist_wheel

生成文件

dist/flaskr-1.0.0-py2-none-any.whl

# 在其他环境安装

# 复制到其他的环境中。安装

pip install flaskr-1.0.0-py2-none-any.whl

# 初始化数据库

export FLASK_APP=flaskr

flask init-db

配置秘钥

运行产品服务器

# 安装所需 python 包

pip install waitress

需要把应用告知 Waitree ，但是方式与 flask run 那样使用 FLASK_APP 不同。需要告知 Waitree 导入并调用应用工厂来得到一个应用对象。

waitress-serve --call 'flaskr:create_app'

==========================================

继续开发

Python - Flask - 进阶

相关阅读

相关文章

相关问答

相关文档