「学习笔记」Flask入门到实战

文章目录

Flask 是一个轻量级的 Python Web 应用框架,由 Pallets 组织开发维护。

Flask 被称作微框架,核心原因是本身不绑定任何专用工具与依赖库,原生不内置数据库抽象层、表单验证等通用 Web 组件。

Flask 具备良好的扩展支持能力,可按需额外加装功能插件;目前已有 ORM 对象关系映射、表单校验、文件上传、开放认证及各类配套开发工具的成熟扩展,能按需补齐项目所需功能。

1. Flask简介与安装

1.1 什么是Flask

Flask是一个基于Werkzeug和Jinja2的微型Web框架:

  • Werkzeug:WSGI工具库,处理HTTP请求和响应
  • Jinja2:模板引擎,用于渲染HTML页面

Flask的特点:

  • 轻量级,核心功能精简
  • 高度可扩展,通过插件扩展功能
  • 灵活的URL路由系统
  • 内置开发服务器和调试工具

1.2 安装Flask

使用pip安装Flask:

pip install flask

验证安装:

import flask
print(flask.__version__)  # 输出当前Flask版本

2. 第一个Flask应用

2.1 创建基础应用

创建app.py文件:

from flask import Flask

app = Flask(__name__)

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

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

运行应用:

python app.py

访问http://localhost:5000即可看到"Hello, Flask!"。

2.2 应用基本结构

app = Flask(__name__)  # 创建Flask应用实例
# __name__表示当前模块名,Flask用它来确定资源位置

@app.route('/')  # 路由装饰器,将URL映射到函数
def index():     # 视图函数,处理请求并返回响应
    return 'Hello, Flask!'

app.run(         # 启动开发服务器
    debug=True,  # 调试模式,代码修改自动重启
    host='0.0.0.0',  # 允许外部访问
    port=5000    # 指定端口
)

3. URL路由与视图函数

3.1 基本路由

@app.route('/')
def index():
    return '首页'

@app.route('/about')
def about():
    return '关于我们'

@app.route('/contact')
def contact():
    return '联系方式'

3.2 动态路由

@app.route('/user/<username>')
def user_profile(username):
    return f'用户: {username}'

@app.route('/post/<int:post_id>')
def show_post(post_id):
    return f'文章ID: {post_id}'

@app.route('/path/<path:subpath>')
def show_path(subpath):
    return f'路径: {subpath}'

支持的转换器类型:

  • string:默认,匹配除斜杠外的字符串
  • int:匹配整数
  • float:匹配浮点数
  • path:匹配包含斜杠的路径

3.3 HTTP方法

from flask import request

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        # 处理表单提交
        username = request.form['username']
        password = request.form['password']
        return f'登录: {username}'
    else:
        # 返回登录表单
        return '''
            <form method="post">
                <input type="text" name="username">
                <input type="password" name="password">
                <button type="submit">登录</button>
            </form>
        '''

3.4 URL构建

from flask import url_for

@app.route('/')
def index():
    # 使用url_for生成URL
    login_url = url_for('login')
    user_url = url_for('user_profile', username='admin')
    return f'<a href="{login_url}">登录</a> | <a href="{user_url}">用户</a>'

4. 模板与静态文件

4.1 使用Jinja2模板

创建templates目录,新建index.html

<!DOCTYPE html>
<html>
<head>
    <title>{{ title }}</title>
</head>
<body>
    <h1>欢迎来到 {{ title }}</h1>
    {% if user %}
        <p>用户: {{ user }}</p>
    {% else %}
        <p>请登录</p>
    {% endif %}
</body>
</html>

在视图函数中渲染模板:

from flask import render_template

@app.route('/')
def index():
    return render_template('index.html', title='首页', user='admin')

4.2 模板继承

创建基础模板base.html

<!DOCTYPE html>
<html>
<head>
    <title>{% block title %}{% endblock %}</title>
</head>
<body>
    <nav>
        <a href="/">首页</a>
        <a href="/about">关于</a>
    </nav>
    <div class="content">
        {% block content %}{% endblock %}
    </div>
</body>
</html>

继承基础模板:

{% extends "base.html" %}

{% block title %}首页{% endblock %}

{% block content %}
    <h1>欢迎来到首页</h1>
{% endblock %}

4.3 静态文件

创建static目录存放CSS、JS、图片等静态资源:

<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
<script src="{{ url_for('static', filename='script.js') }}"></script>
<img src="{{ url_for('static', filename='logo.png') }}" alt="Logo">

5. 请求与响应

5.1 获取请求数据

from flask import request

@app.route('/submit', methods=['POST'])
def submit():
    # 获取表单数据
    username = request.form.get('username')
    email = request.form['email']
    
    # 获取URL参数
    page = request.args.get('page', 1)
    
    # 获取JSON数据
    data = request.get_json()
    
    # 获取请求头
    user_agent = request.headers.get('User-Agent')
    
    return f'用户名: {username}, 邮箱: {email}'

5.2 响应处理

from flask import make_response, jsonify, redirect

@app.route('/response')
def custom_response():
    # 创建自定义响应
    resp = make_response('自定义响应', 200)
    resp.headers['X-Custom'] = 'value'
    return resp

@app.route('/json')
def json_response():
    # 返回JSON响应
    return jsonify({'name': '张三', 'age': 25})

@app.route('/redirect')
def redirect_to_home():
    # 重定向
    return redirect(url_for('index'))

5.3 错误处理

@app.errorhandler(404)
def page_not_found(error):
    return render_template('404.html'), 404

@app.errorhandler(500)
def internal_error(error):
    return '服务器内部错误', 500

6. Session与Cookie

6.1 使用Session

from flask import session

app.secret_key = 'your_secret_key_here'  # 必须设置密钥

@app.route('/login', methods=['POST'])
def login():
    username = request.form['username']
    session['username'] = username  # 保存到session
    return '登录成功'

@app.route('/profile')
def profile():
    username = session.get('username')
    if username:
        return f'欢迎, {username}'
    return '请先登录'

@app.route('/logout')
def logout():
    session.pop('username', None)  # 删除session
    return '已退出登录'

6.2 设置Cookie

@app.route('/set_cookie')
def set_cookie():
    resp = make_response('Cookie已设置')
    resp.set_cookie('username', 'admin', max_age=3600)  # 有效期1小时
    return resp

@app.route('/get_cookie')
def get_cookie():
    username = request.cookies.get('username')
    return f'Cookie值: {username}'

7. 数据库操作

7.1 使用SQLite

import sqlite3
from flask import g

def get_db():
    if 'db' not in g:
        g.db = sqlite3.connect('example.db')
        g.db.row_factory = sqlite3.Row
    return g.db

@app.route('/init_db')
def init_db():
    db = get_db()
    db.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            email TEXT UNIQUE NOT NULL
        )
    ''')
    db.commit()
    return '数据库初始化完成'

@app.route('/add_user')
def add_user():
    db = get_db()
    db.execute('INSERT INTO users (username, email) VALUES (?, ?)',
               ('admin', 'admin@example.com'))
    db.commit()
    return '用户添加成功'

@app.route('/users')
def list_users():
    db = get_db()
    users = db.execute('SELECT * FROM users').fetchall()
    return '\n'.join([f"{user['id']}: {user['username']}" for user in users])

7.2 使用Flask-SQLAlchemy

安装扩展:

pip install flask-sqlalchemy

配置和使用:

from flask_sqlalchemy import SQLAlchemy

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///example.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def __repr__(self):
        return f'<User {self.username}>'

# 创建表
with app.app_context():
    db.create_all()

# 添加用户
@app.route('/add')
def add_user():
    user = User(username='admin', email='admin@example.com')
    db.session.add(user)
    db.session.commit()
    return '用户添加成功'

# 查询用户
@app.route('/users')
def list_users():
    users = User.query.all()
    return '\n'.join([f"{user.id}: {user.username}" for user in users])

8. 表单处理

8.1 使用Flask-WTF

安装扩展:

pip install flask-wtf

创建表单类:

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import DataRequired, Email, Length

class LoginForm(FlaskForm):
    username = StringField('用户名', validators=[DataRequired()])
    password = PasswordField('密码', validators=[DataRequired(), Length(min=6)])
    submit = SubmitField('登录')

class RegistrationForm(FlaskForm):
    username = StringField('用户名', validators=[DataRequired()])
    email = StringField('邮箱', validators=[DataRequired(), Email()])
    password = PasswordField('密码', validators=[DataRequired(), Length(min=6)])
    submit = SubmitField('注册')

在视图函数中使用:

@app.route('/login', methods=['GET', 'POST'])
def login():
    form = LoginForm()
    if form.validate_on_submit():
        username = form.username.data
        password = form.password.data
        # 验证逻辑...
        return f'登录成功: {username}'
    return render_template('login.html', form=form)

模板中渲染表单:

<form method="POST">
    {{ form.csrf_token }}
    {{ form.username.label }} {{ form.username() }}
    {% for error in form.username.errors %}
        <span>{{ error }}</span>
    {% endfor %}
    
    {{ form.password.label }} {{ form.password() }}
    {% for error in form.password.errors %}
        <span>{{ error }}</span>
    {% endfor %}
    
    {{ form.submit() }}
</form>

9. 用户认证

9.1 使用Flask-Login

安装扩展:

pip install flask-login

配置登录管理器:

from flask_login import LoginManager, UserMixin, login_user, login_required, logout_user, current_user

login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'login'  # 未登录时重定向的页面

class User(UserMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password_hash = db.Column(db.String(120), nullable=False)

@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))

@app.route('/login', methods=['GET', 'POST'])
def login():
    form = LoginForm()
    if form.validate_on_submit():
        user = User.query.filter_by(username=form.username.data).first()
        if user and check_password_hash(user.password_hash, form.password.data):
            login_user(user)
            return redirect(url_for('dashboard'))
    return render_template('login.html', form=form)

@app.route('/dashboard')
@login_required
def dashboard():
    return f'欢迎, {current_user.username}'

@app.route('/logout')
@login_required
def logout():
    logout_user()
    return redirect(url_for('index'))

10. 实战项目:博客系统

10.1 项目结构

myblog/
├── app.py              # 主应用文件
├── templates/          # 模板目录
│   ├── base.html       # 基础模板
│   ├── index.html      # 首页
│   ├── login.html      # 登录页
│   ├── register.html   # 注册页
│   └── post.html       # 文章详情页
├── static/             # 静态资源
│   └── style.css       # 样式文件
├── models.py           # 数据库模型
├── forms.py            # 表单定义
└── requirements.txt    # 依赖列表

10.2 核心代码

models.py

from flask_sqlalchemy import SQLAlchemy
from flask_login import UserMixin

db = SQLAlchemy()

class User(UserMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password_hash = db.Column(db.String(120), nullable=False)
    posts = db.relationship('Post', backref='author', lazy=True)

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    content = db.Column(db.Text, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

forms.py

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField, TextAreaField
from wtforms.validators import DataRequired, Email, Length

class LoginForm(FlaskForm):
    username = StringField('用户名', validators=[DataRequired()])
    password = PasswordField('密码', validators=[DataRequired()])
    submit = SubmitField('登录')

class RegisterForm(FlaskForm):
    username = StringField('用户名', validators=[DataRequired()])
    email = StringField('邮箱', validators=[DataRequired(), Email()])
    password = PasswordField('密码', validators=[DataRequired(), Length(min=6)])
    submit = SubmitField('注册')

class PostForm(FlaskForm):
    title = StringField('标题', validators=[DataRequired()])
    content = TextAreaField('内容', validators=[DataRequired()])
    submit = SubmitField('发布')

app.py

from flask import Flask, render_template, redirect, url_for, flash
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, login_user, login_required, logout_user, current_user
from werkzeug.security import generate_password_hash, check_password_hash
from models import db, User, Post
from forms import LoginForm, RegisterForm, PostForm
import datetime

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your_secret_key'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///blog.db'

db.init_app(app)

login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'login'

@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))

@app.route('/')
def index():
    posts = Post.query.order_by(Post.created_at.desc()).all()
    return render_template('index.html', posts=posts)

@app.route('/login', methods=['GET', 'POST'])
def login():
    if current_user.is_authenticated:
        return redirect(url_for('index'))
    form = LoginForm()
    if form.validate_on_submit():
        user = User.query.filter_by(username=form.username.data).first()
        if user and check_password_hash(user.password_hash, form.password.data):
            login_user(user)
            return redirect(url_for('index'))
        flash('登录失败,请检查用户名和密码')
    return render_template('login.html', form=form)

@app.route('/register', methods=['GET', 'POST'])
def register():
    if current_user.is_authenticated:
        return redirect(url_for('index'))
    form = RegisterForm()
    if form.validate_on_submit():
        hashed_password = generate_password_hash(form.password.data)
        user = User(username=form.username.data, email=form.email.data, password_hash=hashed_password)
        db.session.add(user)
        db.session.commit()
        flash('注册成功,请登录')
        return redirect(url_for('login'))
    return render_template('register.html', form=form)

@app.route('/logout')
@login_required
def logout():
    logout_user()
    return redirect(url_for('index'))

@app.route('/post/new', methods=['GET', 'POST'])
@login_required
def new_post():
    form = PostForm()
    if form.validate_on_submit():
        post = Post(title=form.title.data, content=form.content.data, author=current_user)
        db.session.add(post)
        db.session.commit()
        flash('文章发布成功')
        return redirect(url_for('index'))
    return render_template('post.html', form=form, title='新文章')

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
    app.run(debug=True)

10.3 运行项目

创建requirements.txt

flask
flask-sqlalchemy
flask-wtf
flask-login
werkzeug

安装依赖:

pip install -r requirements.txt

运行项目:

python app.py

访问http://localhost:5000即可查看博客系统。

11. 部署Flask应用

11.1 使用Gunicorn

安装Gunicorn:

pip install gunicorn

启动应用:

gunicorn --bind 0.0.0.0:5000 app:app

11.2 使用Nginx反向代理

配置Nginx:

server {
    listen 80;
    server_name your_domain.com;

    location / {
        proxy_pass http://127.0.0.1:5000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }
}

11.3 使用systemd管理进程

创建/etc/systemd/system/myblog.service

[Unit]
Description=Flask Blog App
After=network.target

[Service]
User=www-data
WorkingDirectory=/var/www/myblog
ExecStart=/usr/bin/gunicorn --workers=4 --bind=0.0.0.0:5000 app:app
Restart=always

[Install]
WantedBy=multi-user.target

启动服务:

sudo systemctl daemon-reload
sudo systemctl start myblog
sudo systemctl enable myblog
END .

相关系列文章

×