Flask Database Integration |

Most web applications require persistent data storage.

This chapter uses Python’s built-in SQLite database to demonstrate how to integrate a database into Flask, without requiring any additional dependencies.

---

SQLite Quick Start

SQLite is a lightweight database where data is stored in a single file, and it is supported out of the box by Python’s standard library.

It is the best choice for learning and prototyping—no database server installation or configuration is needed.

Example

# A simple SQLite example (without Flask)

import sqlite3

# Connect to the database (automatically creates tutorial.db if it doesn’t exist)

conn = sqlite3.connect("tutorial.db")

# row_factory allows accessing query results 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 (?, ?)", ("tutorial", "pass123"))

conn.commit()

# Query data

rows = conn.execute("SELECT * FROM user").fetchall()

for row in rows:

    print(dict(row))  # Output: {'id': 1, 'username': 'tutorial', 'password': 'pass123'}

conn.close()

---

Using the g Object to Manage Database Connections in Flask

g is a special object whose lifecycle is tied to a single request.

Storing the database connection in g ensures that multiple database operations within the same request share the same connection, and the connection 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 database connection, reused within the same request"""

    if "db" not in g:

        # Read database path from config

        db_path = current_app.config.get("DATABASE", "tutorial.db")

        g.db = sqlite3.connect(db_path)

        g.db.row_factory = sqlite3.Row  # Enable field-name access for results

    return g.db

def close_db(error=None):

    """Close database connection (automatically called at request end)"""

    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 functionality in the Flask app"""

    # Automatically close database connection at request end

    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 instance folder)

    import os

    app.config = os.path.join(app.instance_path, "tutorial.db")

    # Ensure instance folder exists

    os.makedirs(app.instance_path, exist_ok=True)

    # Initialize database

    from db import init_app as init_db

    init_db(app)

    # Register blueprints

    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

---

Building a Complete Blog API

Combine the database to implement CRUD (Create, Read, Update, Delete) operations for posts:

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 all posts"""
    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 list of Row objects to list of dicts
    return jsonify([dict(post) for post in posts])

@bp.get("/api/posts/<int:post_id>")
def get_post(post_id):
    """Get a single post by ID"""
    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"Post {post_id} does not exist")
    return jsonify(dict(post))

@bp.post("/api/posts")
def create_post():
    """Create a new post"""
    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": "Post created successfully"}), 201

@bp.put("/api/posts/<int:post_id>")
def update_post(post_id):
    """Update a post"""
    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": "Post updated successfully"})

@bp.delete("/api/posts/<int:post_id>")
def delete_post(post_id):
    """Delete a post"""
    db = get_db()
    db.execute("DELETE FROM post WHERE id = ?", (post_id,))
    db.commit()
    return jsonify({"message": "Post deleted"})

Test the API:

# Get list of posts
$ curl http://127.0.0.1:5000/api/posts

# Create a post
$ curl -X POST http://127.0.0.1:5000/api/posts 
  -H "Content-Type: application/json" 
  -d '{"title":"Online Tutorial","body":"Flask Getting Started Content","author_id":1}'

# Update a post
$ curl -X PUT http://127.0.0.1:5000/api/posts/1 
  -H "Content-Type: application/json" 
  -d '{"title":"Online Tutorial (Revised)","body":"Updated content"}'

# Delete a post
$ curl -X DELETE http://127.0.0.1:5000/api/posts/1

Best Practice: When returning JSON data from an API, use appropriate HTTP status codes—201 for creation, 200 for success, 400 for client errors, and 404 for resource not found. This makes it easier for frontend code to determine the request outcome.

---

Database Initialization Command

Use Flask CLI to create a custom command for initializing the database:

# Initialize database (create tables)
(.venv) $ flask init-db

# Verify tables are created
(.venv) $ sqlite3 instance/tutorial.db ".tables"
post  user

---

SQL Injection Prevention

Using parameterized queries (? placeholders) is key to preventing SQL injection:

Example

# Correct: Use ? placeholder (parameterized query)
# SQLite driver automatically escapes parameters to prevent malicious input from being executed as SQL
db.execute("SELECT * FROM user WHERE username = ?", (username,))

# Incorrect: Concatenate SQL with f-string (extremely dangerous!)
# db.execute(f"SELECT * FROM user WHERE username = '{username}'")

# Malicious input like '; DROP TABLE user; -- would delete the entire table!

---

Using Flask-SQLAlchemy (Advanced Optional Reading)

For large-scale projects, writing raw SQL manually can become tedious.

Flask-SQLAlchemy is a popular extension that provides ORM (Object-Relational Mapping) capabilities:

Example

# Install: pip install flask-sqlalchemy

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config = "sqlite:///tutorial.db"

db = SQLAlchemy(app)

# Define data models (table structure using Python classes)
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 tables
with app.app_context():
    db.create_all()

# Create a user
user = User(username="tutorial", email="test@")
db.session.add(user)
db.session.commit()

# Query users
users = User.query.all()
tutorial_user = User.query.filter_by(username="tutorial").first()

In this chapter, we use native SQLite to avoid introducing extra dependencies, making it ideal for learning and small projects. For real-world production projects, it is recommended to use SQLAlchemy or a similar ORM, which significantly reduces boilerplate code and improves security.