flask的使用

# flask/flaskr/__init__.py
import os
from flask import Flask

# 创建app工厂函数,并设置默认配置
def create_app(test_config=None):
    # instance_relative_config:Flask中的选项，用来判断是否关联自定义文件
    app = Flask(__name__, instance_relative_config=True)
    # 配置字典
    app.config.from_mapping(
        SECRET_KET = ‘dev‘,  # 随机字符串
        DATABASE = os.path.join(app.instance_path, ‘flaskr.sqlite‘)  # app.instance_path:获取当前文件夹的路径,构造"当前路径/instace"文件路径。这里的目的是创建数据库文件地址
    )
    if test_config is None:
        # 判断是否配置了默认配置，如果没有，就执行"config.py"默认配置文件
        # from_pyfile:重载配置文件。silent参数默认为false表示如果不存在就报错
        app.config.from_pyfile(‘config.py‘, silent=True)
    else:
        # 执行默认配置
        app.config.from_mapping(test_config)
    try:
        # 尝试创建路径
        os.makedirs(app.instance_path)
    except OSError:
        pass
    # 显示hello
    @app.route("/")
    def hello():
        return "hello"

    # 调用数据库初始化
    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‘)  # 导航栏访问:http://127.0.0.1:5000/，所以必须注释上面的hello输出
    return app

# flask/flaskr/db.py
# 对数据库的操作
import sqlite3
import click
# g:专门用来存储用户信息的对象，功能和session相似，但又不相同
# current_app:表示当前运行程序(也就是调用的app)的实例对象。
from flask import current_app, g
from flask.cli import with_appcontext

# 创建一个sqlite3链接，并保存到g.db中
def get_db():
    if ‘db‘ not in g:
        g.db = sqlite3.connect(
            current_app.config[‘DATABASE‘],
            detect_types = sqlite3.PARSE_DECLTYPES
        )
        g.db.row_factory = sqlite3.Row
    return g.db

# 关闭g中创建的链接
def close_db(e=None):
    db = g.pop(‘db‘, None)
    if db is not None:
        db.close()

# 如果数据库没有初始化,那么执行初始化
def init_db():
    db = get_db()
    # open_resource：打开一个文件，该文件名是相对于 flaskr 包的
    with current_app.open_resource(‘schema.sql‘) as f:
        # 执行sql语句创建内容
        db.executescript(f.read().decode(‘utf8‘))

# click.command()定义一个名为init-db命令行
@click.command(‘init-db‘)
@with_appcontext
def init_db_command():
    init_db()  # 调用init_db
    click.echo(‘Initialized the database.‘)  # 显示初始化信息

# 注：close_db 和 init_db_command 函数需要在应用实例中注册，否则无法使用
def init_app(app):
    app.teardown_appcontext(close_db)  # app.teardown_appcontext()告诉Flask在返回响应后进行清理的时候调用此函数
    app.cli.add_command(init_db_command)  # app.cli.add_command()添加一个新的可以与flask一起工作的命令。

# flask/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‘)  # 创建一个auth的蓝图,当导航栏使用".../auth/"访问时，都会跳到该视图下面

# 蓝图内的路由
@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:
            # db.execute(‘SELECT id FROM user WHERE username = ?‘, (username,)).fetchone()：表示查找一条username的数据
            error = ‘User {} is already registered.‘.format(username)
        if error is None:
            # 如果没有错误执行插入操作
            #  generate_password_hash()生成安全的哈希值并储存到数据库
            db.execute(‘INSERT INTO user (username, password) VALUES (?, ?)‘, (username, generate_password_hash(password)))
            db.commit()  # sqlites必须进行提交
            return redirect(url_for(‘auth.login‘))  # 跳转登录
        # 如果有错误，错误信息闪现
        flash(error)
    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()将user[‘password‘]密码转换为hash值，并与password比较
            error = ‘Incorrect password.‘
        if error is None:
            session.clear()  # 清空session
            session[‘user_id‘] = user[‘id‘]  # 设置session
            return redirect(url_for(‘index‘))  # 跳转首页
        flash(error)
    return render_template(‘auth/login.html‘)

# before_app_request：在每次请求之前执行该函数，同样的还有before_request
@bp.before_app_request
def load_logged_in_user():
    user_id = session.get(‘user_id‘, None)  # 获取session
    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()  # 清空session
    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

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‘))

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));

<!doctype html>
<!-- html5代码,block表示可代替的标题 -->
<title>{% block title %}{% endblock %} - Flaskr</title>
<!-- 调用静态文件 -->
<link rel="stylesheet" href="{{ url_for(‘static‘, filename=‘style.css‘) }}">
<!-- 在html5中nav表示头部 -->
<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>
<!-- session表示主体 -->
<section class="content">
  <header>
    {% block header %}{% endblock %}
  </header>
  <!-- 获取消息闪现信息 -->
  {% for message in get_flashed_messages() %}
    <div class="flash">{{ message }}</div>
  {% endfor %}
  {% block content %}{% endblock %}
</section>