「学习笔记」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 .