Flask Database Integration
Most web applications need persistent data storage.
This chapter uses Python's built-in SQLite database to show how to integrate a database into Flask without installing additional dependencies.
SQLite Quick Start
SQLite is a lightweight database that stores data in a single file, and is supported by the Python standard library.
It is the best choice for learning and prototyping—no need to install a database server or configure anything.
Example
# A simple SQLite example (without using Flask)
import sqlite3
# Connect to the database (example.db is automatically created if it does not exist)
conn = sqlite3.connect("example.db")
# row_factory makes query results accessible by field names like a dictionary
conn.row_factory = sqlite3.Row
# Create table
conn.execute("CREATE TABLE IF NOT EXISTS user (id INTEGER PRIMARY KEY, username TEXT UNIQUE, password TEXT)")
# Insert data
conn.execute("INSERT INTO user (username, password) VALUES (?, ?)", ("example", "pass123"))
conn.commit()
# Query data
rows = conn.execute("SELECT * FROM user").fetchall()
for row in rows:
print(dict(row)) # Output: {'id': 1, 'username': 'example', 'password': 'pass123'}
conn.close()
import sqlite3
# Connect to the database (example.db is automatically created if it does not exist)
conn = sqlite3.connect("example.db")
# row_factory makes query results accessible by field names like a dictionary
conn.row_factory = sqlite3.Row
# Create table
conn.execute("CREATE TABLE IF NOT EXISTS user (id INTEGER PRIMARY KEY, username TEXT UNIQUE, password TEXT)")
# Insert data
conn.execute("INSERT INTO user (username, password) VALUES (?, ?)", ("example", "pass123"))
conn.commit()
# Query data
rows = conn.execute("SELECT * FROM user").fetchall()
for row in rows:
print(dict(row)) # Output: {'id': 1, 'username': 'example', 'password': 'pass123'}
conn.close()
Using the g object in Flask to manage database connections
gIt is a special object whose lifecycle is a single request.
Store the database connectiongIt ensures that multiple database operations within the same request share the same connection, which is automatically cleaned up after the request ends.
Example
# File path: db.py (database management module)
import sqlite3
from flask import g, current_app
def get_db():
"""Get a database connection, reused within the same request"""
if "db" not in g:
# Read database path from configuration
db_path = current_app.config.get("DATABASE", "example.db")
g.db = sqlite3.connect(db_path)
g.db.row_factory = sqlite3.Row # Allow results to be accessed by field name
return g.db
def close_db(error=None):
"""Close the database connection (automatically called at the end of the request)"""
db = g.pop("db", None)
if db is not None:
db.close()
def init_db():
"""Initialize database table structure"""
db = get_db()
db.execute("""
CREATE TABLE IF NOT EXISTS user (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS post (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
body TEXT NOT NULL,
author_id INTEGER NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (author_id) REFERENCES user (id)
)
""")
db.commit()
def init_app(app):
"""Register database-related functions in the Flask application"""
# Automatically close the database connection at the end of the request
app.teardown_appcontext(close_db)
# Register CLI command: flask init-db
app.cli.command("init-db")(init_db)
import sqlite3
from flask import g, current_app
def get_db():
"""Get a database connection, reused within the same request"""
if "db" not in g:
# Read database path from configuration
db_path = current_app.config.get("DATABASE", "example.db")
g.db = sqlite3.connect(db_path)
g.db.row_factory = sqlite3.Row # Allow results to be accessed by field name
return g.db
def close_db(error=None):
"""Close the database connection (automatically called at the end of the request)"""
db = g.pop("db", None)
if db is not None:
db.close()
def init_db():
"""Initialize database table structure"""
db = get_db()
db.execute("""
CREATE TABLE IF NOT EXISTS user (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS post (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
body TEXT NOT NULL,
author_id INTEGER NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (author_id) REFERENCES user (id)
)
""")
db.commit()
def init_app(app):
"""Register database-related functions in the Flask application"""
# Automatically close the database connection at the end of the request
app.teardown_appcontext(close_db)
# Register CLI command: flask init-db
app.cli.command("init-db")(init_db)
Factory pattern integration
Use the factory pattern to integrate the database with other modules:
Example
# File path: app.py
from flask import Flask
def create_app():
app = Flask(__name__)
app.secret_key = "dev-secret-key"
# Configure database path (stored in the instance folder)
import os
app.config["DATABASE"] = os.path.join(app.instance_path, "example.db")
# Ensure the instance folder exists
os.makedirs(app.instance_path, exist_ok=True)
# Initialize the database
from db import init_app as init_db
init_db(app)
# Register blueprint
from auth import bp as auth_bp
from blog import bp as blog_bp
app.register_blueprint(auth_bp, url_prefix="/auth")
app.register_blueprint(blog_bp)
return app
from flask import Flask
def create_app():
app = Flask(__name__)
app.secret_key = "dev-secret-key"
# Configure database path (stored in the instance folder)
import os
app.config["DATABASE"] = os.path.join(app.instance_path, "example.db")
# Ensure the instance folder exists
os.makedirs(app.instance_path, exist_ok=True)
# Initialize the database
from db import init_app as init_db
init_db(app)
# Register blueprint
from auth import bp as auth_bp
from blog import bp as blog_bp
app.register_blueprint(auth_bp, url_prefix="/auth")
app.register_blueprint(blog_bp)
return app
Write a complete blog API
Combine the database to implement CRUD (Create, Read, Update, Delete) operations for articles:
Example
# File path: blog.py
from flask import Blueprint, request, jsonify, abort, g
from db import get_db
bp = Blueprint("blog", __name__)
@bp.get("/api/posts")
def list_posts():
"""Get the list of all articles"""
db = get_db()
posts = db.execute(
"SELECT p.id, p.title, p.body, p.created_at, u.username"
" FROM post p JOIN user u ON p.author_id = u.id"
" ORDER BY p.created_at DESC"
).fetchall()
# Convert a list of Row objects to a list of dicts
return jsonify([dict(post) for post in posts])
@bp.get("/api/posts/<int:post_id>")
def get_post(post_id):
"""Get the details of a single article"""
db = get_db()
post = db.execute(
"SELECT id, title, body, created_at FROM post WHERE id = ?",
(post_id,)
).fetchone()
if post is None:
abort(404, description=f"Article {post_id} does not exist")
return jsonify(dict(post))
@bp.post("/api/posts")
def create_post():
"""Create a new article"""
data = request.json
title = data.get("title", "").strip()
body = data.get("body", "").strip()
author_id = data.get("author_id", 1) # Default author ID
if not title:
return jsonify({"error": "Title cannot be empty"}), 400
db = get_db()
cursor = db.execute(
"INSERT INTO post (title, body, author_id) VALUES (?, ?, ?)",
(title, body, author_id)
)
db.commit()
return jsonify({"id": cursor.lastrowid, "title": title, "message": "Article created successfully"}), 201
@bp.put("/api/posts/<int:post_id>")
def update_post(post_id):
"""Update an article"""
data = request.json
title = data.get("title", "").strip()
body = data.get("body", "").strip()
if not title:
return jsonify({"error": "Title cannot be empty"}), 400
db = get_db()
db.execute(
"UPDATE post SET title = ?, body = ? WHERE id = ?",
(title, body, post_id)
)
db.commit()
return jsonify({"id": post_id, "message": "Article updated successfully"})
@bp.delete("/api/posts/<int:post_id>")
def delete_post(post_id):
"""Delete an article"""
db = get_db()
db.execute("DELETE FROM post WHERE id = ?", (post_id,))
db.commit()
return jsonify({"message": "Article deleted"})
from flask import Blueprint, request, jsonify, abort, g
from db import get_db
bp = Blueprint("blog", __name__)
@bp.get("/api/posts")
def list_posts():
"""Get the list of all articles"""
db = get_db()
posts = db.execute(
"SELECT p.id, p.title, p.body, p.created_at, u.username"
" FROM post p JOIN user u ON p.author_id = u.id"
" ORDER BY p.created_at DESC"
).fetchall()
# Convert a list of Row objects to a list of dicts
return jsonify([dict(post) for post in posts])
@bp.get("/api/posts/<int:post_id>")
def get_post(post_id):
"""Get the details of a single article"""
db = get_db()
post = db.execute(
"SELECT id, title, body, created_at FROM post WHERE id = ?",
(post_id,)
).fetchone()
if post is None:
abort(404, description=f"Article {post_id} does not exist")
return jsonify(dict(post))
@bp.post("/api/posts")
def create_post():
"""Create a new article"""
data = request.json
title = data.get("title", "").strip()
body = data.get("body", "").strip()
author_id = data.get("author_id", 1) # Default author ID
if not title:
return jsonify({"error": "Title cannot be empty"}), 400
db = get_db()
cursor = db.execute(
"INSERT INTO post (title, body, author_id) VALUES (?, ?, ?)",
(title, body, author_id)
)
db.commit()
return jsonify({"id": cursor.lastrowid, "title": title, "message": "Article created successfully"}), 201
@bp.put("/api/posts/<int:post_id>")
def update_post(post_id):
"""Update an article"""
data = request.json
title = data.get("title", "").strip()
body = data.get("body", "").strip()
if not title:
return jsonify({"error": "Title cannot be empty"}), 400
db = get_db()
db.execute(
"UPDATE post SET title = ?, body = ? WHERE id = ?",
(title, body, post_id)
)
db.commit()
return jsonify({"id": post_id, "message": "Article updated successfully"})
@bp.delete("/api/posts/<int:post_id>")
def delete_post(post_id):
"""Delete an article"""
db = get_db()
db.execute("DELETE FROM post WHERE id = ?", (post_id,))
db.commit()
return jsonify({"message": "Article deleted"})
Test API:
# 获取文章列表
$ curl http://127.0.0.1:5000/api/posts
# 创建文章
$ curl -X POST http://127.0.0.1:5000/api/posts \
-H "Content-Type: application/json" \
-d '{"title":"EXAMPLE 教程","body":"Flask 入门教程内容","author_id":1}'
# 更新文章
$ curl -X PUT http://127.0.0.1:5000/api/posts/1 \
-H "Content-Type: application/json" \
-d '{"title":"EXAMPLE 教程（修订版）","body":"更新后的内容"}'
# 删除文章
$ curl -X DELETE http://127.0.0.1:5000/api/posts/1
Best Practices: When the API returns JSON data, use the correct HTTP status code—for creation use201, use200Use for client errors400Use for non-existent resources404This makes it easier for frontend code to determine the request result.
Database initialization command
Through the Flask CLI, you can create custom commands to initialize the database:
# 初始化数据库（创建表结构） (.venv) $ flask init-db # 验证表已创建 (.venv) $ sqlite3 instance/example.db ".tables" post user
SQL injection protection
Use parameterized queries (?Placeholders) are key to preventing SQL injection:
Example
# Correct approach: Use ? placeholders (parameterized queries)
# The SQLite driver automatically escapes parameters to prevent malicious input from being executed as SQL
db.execute("SELECT * FROM user WHERE username = ?", (username,))
# Wrong approach: Use f-string to concatenate SQL (extremely dangerous!)
# db.execute(f"SELECT * FROM user WHERE username = '{username}'")
# Malicious input '; DROP TABLE user; -- would delete the entire table!
# The SQLite driver automatically escapes parameters to prevent malicious input from being executed as SQL
db.execute("SELECT * FROM user WHERE username = ?", (username,))
# Wrong approach: Use f-string to concatenate SQL (extremely dangerous!)
# db.execute(f"SELECT * FROM user WHERE username = '{username}'")
# Malicious input '; DROP TABLE user; -- would delete the entire table!
Using Flask-SQLAlchemy (Advanced Optional Reading)
For large projects, manually writing SQL gradually becomes tedious.
Flask-SQLAlchemyIt is a widely popular extension that provides ORM (Object-Relational Mapping) capability:
Example
# Install: pip install flask-sqlalchemy
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///example.db"
db = SQLAlchemy(app)
# Define data model (table structure represented by a Python class)
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}>"
# Create table
with app.app_context():
db.create_all()
# Create a user
user = User(username="example", email="test@example.com")
db.session.add(user)
db.session.commit()
# Query a user
users = User.query.all()
example_user = User.query.filter_by(username="example").first()
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///example.db"
db = SQLAlchemy(app)
# Define data model (table structure represented by a Python class)
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}>"
# Create table
with app.app_context():
db.create_all()
# Create a user
user = User(username="example", email="test@example.com")
db.session.add(user)
db.session.commit()
# Query a user
users = User.query.all()
example_user = User.query.filter_by(username="example").first()
other extensionsThis chapter uses native SQLite to avoid introducing additional dependencies, making it suitable for learning and small projects. For real production projects, we recommend using SQLAlchemy or a similar ORM, which can significantly reduce repetitive code and improve security.