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>

<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">

<nav>

<h1>Flaskr</h1>

<ul>

{% if g.user %}

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

<li><a href="{{ url_for('auth.logout') }}">Log Out</a>

{% else %}

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

<li><a href="{{ url_for('auth.login') }}">Log In</a>

{% endif %}

</ul>

</nav>

<section class="content">

<header>

{% 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 %}

<form method="post">

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

<input name="username" id="username" required>

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

<input type="password" name="password" id="password" required>

<input type="submit" value="Register">

</form>

{% endblock %}

 

# 登录

flaskr/templates/auth/login.html

{% extends 'base.html' %}

 

{% block header %}

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

{% endblock %}

 

{% block content %}

<form method="post">

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

<input name="username" id="username" required>

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

<input type="password" name="password" id="password" required>

<input type="submit" value="Log In">

</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 %}

<a class="action" href="{{ url_for('blog.create') }}">New</a>

{% endif %}

{% endblock %}

 

{% block content %}

{% for post in posts %}

<article class="post">

<header>

<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'] %}

<a class="action" href="{{ url_for('blog.update', id=post['id']) }}">Edit</a>

{% endif %}

</header>

<p class="body">{{ post['body'] }}</p>

</article>

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

<hr>

{% endif %}

{% endfor %}

{% endblock %}

 

# 创建模板

{% extends 'base.html' %}

 

{% block header %}

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

{% endblock %}

 

{% block content %}

<form method="post">

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

<input name="title" id="title" value="{{ request.form['title'] }}" required>

<label for="body">Body</label>

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

<input type="submit" value="Save">

</form>

{% endblock %}

 

# 更新/删除模板

{% extends 'base.html' %}

{% block header %}

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

{% endblock %}

{% block content %}

<form method="post">

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

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

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

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

<label for="body">Body</label>

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

<input type="submit" value="Save">

</form>

<hr>

<form action="{{ url_for('blog.delete', id=post['id']) }}" method="post">

<input class="danger" type="submit" value="Delete" οnclick="return confirm('Are you sure?');">

</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'

 

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

继续开发

 类似资料: