Flask Raw SQL

In this tutorial, you'll learn how to use raw SQL queries with Flask to interact with databases directly. While ORMs like SQLAlchemy offer abstraction and convenience, understanding raw SQL execution is valuable for performance optimization, complex queries, or when working with legacy systems.

Introduction to Raw SQL in Flask

Raw SQL refers to writing SQL queries directly instead of using an Object-Relational Mapper (ORM). Flask doesn't include a database interface by default, but it's designed to work seamlessly with various database libraries. The approach allows for:

Full control over the SQL being executed
Potentially better performance for complex queries
Direct access to database-specific features
Better understanding of what's happening under the hood

Setting Up Database Connections

Before executing raw SQL, you need to establish a database connection. Let's look at different options:

SQLite Connection

SQLite is a lightweight disk-based database that's perfect for smaller applications or development environments.

python
import sqlite3
from flask import Flask, g

app = Flask(__name__)

DATABASE = 'database.db'

def get_db():
    db = getattr(g, '_database', None)
    if db is None:
        db = g._database = sqlite3.connect(DATABASE)
        db.row_factory = sqlite3.Row  # This enables column access by name
    return db

@app.teardown_appcontext
def close_connection(exception):
    db = getattr(g, '_database', None)
    if db is not None:
        db.close()

MySQL Connection

For MySQL databases, you'll need to install the mysql-connector-python package:

bash
pip install mysql-connector-python

Here's how to set up a MySQL connection:

python
import mysql.connector
from flask import Flask, g

app = Flask(__name__)

def get_db():
    db = getattr(g, '_database', None)
    if db is None:
        db = g._database = mysql.connector.connect(
            host="localhost",
            user="yourusername",
            password="yourpassword",
            database="yourdatabase"
        )
    return db

@app.teardown_appcontext
def close_connection(exception):
    db = getattr(g, '_database', None)
    if db is not None:
        db.close()

PostgreSQL Connection

For PostgreSQL, install the psycopg2 package:

bash
pip install psycopg2-binary

And here's the connection setup:

python
import psycopg2
from psycopg2.extras import RealDictCursor
from flask import Flask, g

app = Flask(__name__)

def get_db():
    db = getattr(g, '_database', None)
    if db is None:
        db = g._database = psycopg2.connect(
            host="localhost",
            user="yourusername",
            password="yourpassword",
            dbname="yourdatabase",
            cursor_factory=RealDictCursor  # This enables column access by name
        )
    return db

@app.teardown_appcontext
def close_connection(exception):
    db = getattr(g, '_database', None)
    if db is not None:
        db.close()

Executing Raw SQL Queries

Now let's execute some raw SQL queries. We'll create helper functions to standardize our database operations:

python
def query_db(query, args=(), one=False):
    """Execute a SQL query and return the results."""
    cur = get_db().cursor()
    cur.execute(query, args)
    rv = cur.fetchall()
    cur.close()
    get_db().commit()  # Automatically commit changes
    return (rv[0] if rv else None) if one else rv

def execute_db(query, args=()):
    """Execute a SQL command that doesn't return results."""
    cur = get_db().cursor()
    cur.execute(query, args)
    get_db().commit()
    cur.close()
    return cur.rowcount  # Return how many rows were affected

Creating Tables

Let's create a simple users table:

python
@app.route('/setup')
def setup():
    execute_db('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT NOT NULL UNIQUE,
            email TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    return 'Database setup completed!'

For MySQL or PostgreSQL, the syntax would be slightly different:

python
# For MySQL
execute_db('''
    CREATE TABLE IF NOT EXISTS users (
        id INT AUTO_INCREMENT PRIMARY KEY,
        username VARCHAR(100) NOT NULL UNIQUE,
        email VARCHAR(100) NOT NULL,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
''')

# For PostgreSQL
execute_db('''
    CREATE TABLE IF NOT EXISTS users (
        id SERIAL PRIMARY KEY,
        username VARCHAR(100) NOT NULL UNIQUE,
        email VARCHAR(100) NOT NULL,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
''')

Inserting Data

Let's create a route to add a new user:

python
from flask import request, jsonify

@app.route('/users/add', methods=['POST'])
def add_user():
    data = request.get_json()
    
    if not data or 'username' not in data or 'email' not in data:
        return jsonify({'error': 'Invalid data'}), 400
    
    try:
        execute_db(
            'INSERT INTO users (username, email) VALUES (?, ?)',
            (data['username'], data['email'])
        )
        return jsonify({'message': 'User added successfully'}), 201
    except Exception as e:
        return jsonify({'error': str(e)}), 500

For MySQL or PostgreSQL, you would use different placeholders:

python
# For MySQL
execute_db(
    'INSERT INTO users (username, email) VALUES (%s, %s)',
    (data['username'], data['email'])
)

# For PostgreSQL
execute_db(
    'INSERT INTO users (username, email) VALUES (%s, %s)',
    (data['username'], data['email'])
)

Querying Data

Now let's retrieve users from the database:

python
@app.route('/users')
def get_users():
    users = query_db('SELECT * FROM users')
    
    # Convert SQLite Row objects to dictionaries for JSON serialization
    result = [dict(user) for user in users]
    
    return jsonify(result)

Retrieving a Single User

python
@app.route('/users/<username>')
def get_user(username):
    user = query_db('SELECT * FROM users WHERE username = ?', [username], one=True)
    
    if user is None:
        return jsonify({'error': 'User not found'}), 404
    
    return jsonify(dict(user))

Updating Data

python
@app.route('/users/<username>', methods=['PUT'])
def update_user(username):
    data = request.get_json()
    
    if not data or 'email' not in data:
        return jsonify({'error': 'Invalid data'}), 400
    
    try:
        affected_rows = execute_db(
            'UPDATE users SET email = ? WHERE username = ?',
            (data['email'], username)
        )
        
        if affected_rows == 0:
            return jsonify({'error': 'User not found'}), 404
            
        return jsonify({'message': 'User updated successfully'})
    except Exception as e:
        return jsonify({'error': str(e)}), 500

Deleting Data

python
@app.route('/users/<username>', methods=['DELETE'])
def delete_user(username):
    try:
        affected_rows = execute_db('DELETE FROM users WHERE username = ?', [username])
        
        if affected_rows == 0:
            return jsonify({'error': 'User not found'}), 404
            
        return jsonify({'message': 'User deleted successfully'})
    except Exception as e:
        return jsonify({'error': str(e)}), 500

Preventing SQL Injection

When using raw SQL, it's crucial to protect against SQL injection attacks. Always use parameterized queries (as shown in our examples) rather than string concatenation:

python
# NEVER DO THIS (vulnerable to SQL injection):
username = request.args.get('username')
execute_db(f"SELECT * FROM users WHERE username = '{username}'")

# ALWAYS DO THIS (safe):
username = request.args.get('username')
query_db("SELECT * FROM users WHERE username = ?", [username])

Transactions in Raw SQL

When you need to ensure multiple operations either all succeed or all fail, use transactions:

python
@app.route('/transfer', methods=['POST'])
def transfer_money():
    data = request.get_json()
    from_account = data['from']
    to_account = data['to']
    amount = data['amount']
    
    conn = get_db()
    try:
        # Start a transaction
        conn.execute('BEGIN TRANSACTION')
        
        # Deduct from the first account
        conn.execute(
            'UPDATE accounts SET balance = balance - ? WHERE id = ?',
            (amount, from_account)
        )
        
        # Add to the second account
        conn.execute(
            'UPDATE accounts SET balance = balance + ? WHERE id = ?',
            (amount, to_account)
        )
        
        # Commit the transaction
        conn.commit()
        return jsonify({'status': 'success'})
    except Exception as e:
        # Roll back any changes if something goes wrong
        conn.rollback()
        return jsonify({'status': 'error', 'message': str(e)}), 500

Practical Example: A Simple Blog

Let's create a more complete example of a blog application using raw SQL:

Database Setup

python
def init_db():
    with app.app_context():
        db = get_db()
        
        # Create tables
        db.execute('''
            CREATE TABLE IF NOT EXISTS posts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                content TEXT NOT NULL,
                author TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        
        db.execute('''
            CREATE TABLE IF NOT EXISTS comments (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                post_id INTEGER NOT NULL,
                author TEXT NOT NULL,
                content TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (post_id) REFERENCES posts (id)
            )
        ''')
        
        db.commit()

# Initialize the database when the app starts
with app.app_context():
    init_db()

Blog Routes

python
@app.route('/posts', methods=['GET', 'POST'])
def posts():
    if request.method == 'POST':
        data = request.get_json()
        
        if not all(key in data for key in ['title', 'content', 'author']):
            return jsonify({'error': 'Missing required fields'}), 400
            
        execute_db(
            'INSERT INTO posts (title, content, author) VALUES (?, ?, ?)',
            (data['title'], data['content'], data['author'])
        )
        
        return jsonify({'message': 'Post created successfully'}), 201
    else:
        posts = query_db('SELECT * FROM posts ORDER BY created_at DESC')
        return jsonify([dict(post) for post in posts])

@app.route('/posts/<int:post_id>', methods=['GET'])
def get_post(post_id):
    # Get the post
    post = query_db('SELECT * FROM posts WHERE id = ?', [post_id], one=True)
    
    if post is None:
        return jsonify({'error': 'Post not found'}), 404
    
    # Get the comments for this post
    comments = query_db('SELECT * FROM comments WHERE post_id = ? ORDER BY created_at', [post_id])
    
    result = dict(post)
    result['comments'] = [dict(comment) for comment in comments]
    
    return jsonify(result)

@app.route('/posts/<int:post_id>/comments', methods=['POST'])
def add_comment(post_id):
    data = request.get_json()
    
    if not all(key in data for key in ['author', 'content']):
        return jsonify({'error': 'Missing required fields'}), 400
        
    # Check if the post exists
    post = query_db('SELECT id FROM posts WHERE id = ?', [post_id], one=True)
    if post is None:
        return jsonify({'error': 'Post not found'}), 404
        
    execute_db(
        'INSERT INTO comments (post_id, author, content) VALUES (?, ?, ?)',
        (post_id, data['author'], data['content'])
    )
    
    return jsonify({'message': 'Comment added successfully'}), 201

Advanced SQL Features

Raw SQL lets you use advanced database features that might not be easily accessible through ORMs:

Full Text Search in SQLite

python
@app.route('/search')
def search():
    query = request.args.get('q', '')
    
    if not query:
        return jsonify([])
    
    # Simple text search
    posts = query_db(
        "SELECT * FROM posts WHERE title LIKE ? OR content LIKE ?", 
        [f"%{query}%", f"%{query}%"]
    )
    
    return jsonify([dict(post) for post in posts])

Using Subqueries

python
@app.route('/popular-authors')
def popular_authors():
    # Find authors with the most comments on their posts
    authors = query_db('''
        SELECT p.author, COUNT(c.id) as comment_count
        FROM posts p
        LEFT JOIN comments c ON p.id = c.post_id
        GROUP BY p.author
        ORDER BY comment_count DESC
        LIMIT 5
    ''')
    
    return jsonify([dict(author) for author in authors])

Summary

In this tutorial, you've learned how to:

Set up database connections in Flask for different database types
Create helper functions for executing raw SQL queries
Perform CRUD operations using raw SQL
Prevent SQL injection attacks
Use transactions for data integrity
Build a practical blog example using raw SQL queries
Implement advanced SQL features like text search and complex joins

Raw SQL in Flask gives you the flexibility to fully control your database interactions, which can be essential for optimizing performance or implementing complex queries. However, it requires a good understanding of SQL and careful attention to security concerns like SQL injection.

Additional Resources

Exercises

Create a simple task management application that allows users to create, read, update, and delete tasks using raw SQL.
Modify the blog example to include categories for posts and implement filtering by category.
Implement pagination for the blog posts list using SQL's LIMIT and OFFSET.
Add user authentication to the blog example and only allow users to edit or delete their own posts.
Implement a search feature that can search across multiple tables (posts and comments) and highlight the matching text.

If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)

Introduction to Raw SQL in Flask​

Setting Up Database Connections​

SQLite Connection​

MySQL Connection​

PostgreSQL Connection​

Executing Raw SQL Queries​

Creating Tables​

Inserting Data​

Querying Data​

Retrieving a Single User​

Updating Data​

Deleting Data​

Preventing SQL Injection​

Transactions in Raw SQL​

Practical Example: A Simple Blog​

Database Setup​

Blog Routes​

Advanced SQL Features​

Full Text Search in SQLite​

Using Subqueries​

Summary​

Additional Resources​

Exercises​