flask学习2-应用(博客)

发布于:2025-03-07 ⋅ 阅读:(43) ⋅ 点赞:(0)

项目目录

/home/user/Projects/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/
├── pyproject.toml
└── MANIFEST.in
  • .gitignore

    .venv/
    
    *.pyc
    __pycache__/
    
    instance/
    
    .pytest_cache/
    .coverage
    htmlcov/
    
    dist/
    build/
    *.egg-info/
    

应用程序工厂

mkdir flaskr
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',
        DATABASE=os.path.join(app.instance_path, 'flaskr.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

    # a simple page that says hello
    @app.route('/hello')
    def hello():
        return 'Hello, World!'

    return app
  • 在flask-tutorial中执行flask --app flaskr run --debug

  • http://127.0.0.1:5000/hello

连接到数据库

  • flaskr/db.py

    import sqlite3
    from datetime import datetime
    
    import click
    from flask import current_app, g
    
    
    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
    
    
    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')
    def init_db_command():
        """Clear the existing data and create new tables."""
        init_db()
        click.echo('Initialized the database.')
    
    
    sqlite3.register_converter(
        "timestamp", lambda v: datetime.fromisoformat(v.decode())
    )
    
    # 注册应用程序
    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)
    );
    
  • flaskr/init.py

    def create_app():
        app = ...
        # existing code omitted
    
        from . import db
        db.init_app(app)
    
        return app
    
  • 初始化数据库文件

    $ flask --app flaskr init-db
    Initialized the database.
    

蓝图和视图

  • Flaskr 将有两个蓝图,一个用于身份验证函数,另一个 一个用于博客文章功能

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

flaskr/init.py 注册身份验证

def create_app():
    app = ...
    # existing code omitted

    from . import auth
    app.register_blueprint(auth.bp)

    return app

第一个视图:注册

flaskr/auth.py

注册
@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.'

        if error is None:
            try:
                db.execute(
                    "INSERT INTO user (username, password) VALUES (?, ?)",
                    (username, generate_password_hash(password)),
                )
                db.commit()
            except db.IntegrityError:
                error = f"User {username} is already registered."
            else:
                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):
            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查询用户
@bp.before_app_request
def load_logged_in_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'))

创建、编辑和删除博客文章将要求用户 已登录。

flaskr/auth.py

# 此装饰器返回一个包装原始视图的新视图函数 它被应用于。新函数检查用户是否已加载,并且 否则重定向到登录页面。如果用户加载了原始 view 被调用并继续正常。在以下情况下,您将使用此装饰器 编写博客视图。
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

使用蓝图时,蓝图的名称会附加到 name 的函数

模板

基本布局

flaskr/templates/base.html

<!doctype html>
<title>{% block title %}{% endblock %} - Flaskr</title>
<link rel="stylesheet" href="{
    { url_for('static', filename='style.css') }}">