How to Use an SQLite Database in a Flask Application
Flask applications often need persistent data storage, and SQLite serves as an excellent lightweight database solution that requires zero configuration while providing all the features of a SQL database. Understanding how to properly integrate SQLite with Flask is crucial for building scalable web applications, especially when you need local data persistence without the overhead of a full database server. This guide will walk you through setting up SQLite with Flask, implementing database operations, handling connections efficiently, and avoiding common pitfalls that can impact your application’s performance and reliability.
How SQLite and Flask Work Together
SQLite is a self-contained, serverless database engine that stores data in a single file on your filesystem. Unlike traditional database servers like PostgreSQL or MySQL, SQLite doesn’t require a separate server process, making it perfect for development, testing, and smaller production applications. Flask, being a lightweight web framework, pairs naturally with SQLite through Python’s built-in sqlite3 module.
The integration works through database connections that are created, used, and closed during HTTP request cycles. Flask’s application context and request context systems help manage these connections efficiently, ensuring that database operations are properly isolated between requests while maintaining good performance.
Step-by-Step Implementation Guide
Let’s build a complete Flask application with SQLite integration from scratch. First, create your project structure:
mkdir flask-sqlite-app
cd flask-sqlite-app
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
pip install flask
Create the main application file app.py:
import sqlite3
import os
from flask import Flask, request, session, g, redirect, url_for, abort, render_template, flash
app = Flask(__name__)
app.config.from_object(__name__)
# Configuration
app.config.update(dict(
DATABASE=os.path.join(app.root_path, 'app.db'),
SECRET_KEY='your-secret-key-here',
USERNAME='admin',
PASSWORD='default'
))
def connect_db():
"""Connects to the specific database."""
rv = sqlite3.connect(app.config['DATABASE'])
rv.row_factory = sqlite3.Row
return rv
def init_db():
"""Initializes the database with our schema."""
db = get_db()
with app.open_resource('schema.sql', mode='r') as f:
db.cursor().executescript(f.read())
db.commit()
def get_db():
"""Opens a new database connection if there is none yet for the current application context."""
if not hasattr(g, 'sqlite_db'):
g.sqlite_db = connect_db()
return g.sqlite_db
@app.teardown_appcontext
def close_db(error):
"""Closes the database again at the end of the request."""
if hasattr(g, 'sqlite_db'):
g.sqlite_db.close()
Create the database schema file schema.sql:
DROP TABLE IF EXISTS entries;
CREATE TABLE entries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
content TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
DROP TABLE IF EXISTS users;
CREATE TABLE users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
email TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
Add routes for database operations:
@app.route('/')
def show_entries():
db = get_db()
cur = db.execute('SELECT title, content, created_at FROM entries ORDER BY id DESC')
entries = cur.fetchall()
return render_template('show_entries.html', entries=entries)
@app.route('/add', methods=['POST'])
def add_entry():
if not session.get('logged_in'):
abort(401)
db = get_db()
db.execute('INSERT INTO entries (title, content) VALUES (?, ?)',
[request.form['title'], request.form['text']])
db.commit()
flash('New entry was successfully posted')
return redirect(url_for('show_entries'))
@app.route('/login', methods=['GET', 'POST'])
def login():
error = None
if request.method == 'POST':
if request.form['username'] != app.config['USERNAME']:
error = 'Invalid username'
elif request.form['password'] != app.config['PASSWORD']:
error = 'Invalid password'
else:
session['logged_in'] = True
flash('You were logged in')
return redirect(url_for('show_entries'))
return render_template('login.html', error=error)
@app.route('/logout')
def logout():
session.pop('logged_in', None)
flash('You were logged out')
return redirect(url_for('show_entries'))
if __name__ == '__main__':
init_db()
app.run(debug=True)
Advanced Database Operations and Connection Management
For production applications, you’ll need more sophisticated database handling. Here’s an improved approach using connection pooling and proper error handling:
import sqlite3
import threading
from contextlib import contextmanager
from flask import Flask, g
class DatabaseManager:
def __init__(self, db_path, check_same_thread=False):
self.db_path = db_path
self.check_same_thread = check_same_thread
self._local = threading.local()
def get_connection(self):
if not hasattr(self._local, 'connection'):
self._local.connection = sqlite3.connect(
self.db_path,
check_same_thread=self.check_same_thread
)
self._local.connection.row_factory = sqlite3.Row
# Enable foreign key constraints
self._local.connection.execute('PRAGMA foreign_keys = ON')
return self._local.connection
@contextmanager
def get_cursor(self, commit=False):
conn = self.get_connection()
cursor = conn.cursor()
try:
yield cursor
if commit:
conn.commit()
except Exception as e:
conn.rollback()
raise e
finally:
cursor.close()
def close_connection(self):
if hasattr(self._local, 'connection'):
self._local.connection.close()
delattr(self._local, 'connection')
# Initialize database manager
db_manager = DatabaseManager('app.db')
# Helper functions for common operations
def execute_query(query, params=None, fetch_one=False, fetch_all=False, commit=False):
with db_manager.get_cursor(commit=commit) as cursor:
if params:
cursor.execute(query, params)
else:
cursor.execute(query)
if fetch_one:
return cursor.fetchone()
elif fetch_all:
return cursor.fetchall()
return cursor.rowcount
def execute_many(query, params_list):
with db_manager.get_cursor(commit=True) as cursor:
cursor.executemany(query, params_list)
return cursor.rowcount
Implement transaction handling for complex operations:
@contextmanager
def transaction():
conn = db_manager.get_connection()
try:
yield conn
conn.commit()
except Exception as e:
conn.rollback()
raise e
# Usage example
def transfer_data(from_id, to_id, amount):
try:
with transaction():
# Deduct from source
execute_query(
'UPDATE accounts SET balance = balance - ? WHERE id = ?',
(amount, from_id),
commit=False
)
# Add to destination
execute_query(
'UPDATE accounts SET balance = balance + ? WHERE id = ?',
(amount, to_id),
commit=False
)
# Log transaction
execute_query(
'INSERT INTO transactions (from_id, to_id, amount) VALUES (?, ?, ?)',
(from_id, to_id, amount),
commit=False
)
except sqlite3.Error as e:
flash(f'Transaction failed: {e}')
return False
return True
Real-World Examples and Use Cases
SQLite with Flask excels in several scenarios. Here are practical implementations for common use cases:
User Authentication System:
import hashlib
import secrets
from datetime import datetime, timedelta
class UserManager:
@staticmethod
def create_user(username, email, password):
salt = secrets.token_hex(16)
password_hash = hashlib.pbkdf2_hmac('sha256',
password.encode('utf-8'),
salt.encode('utf-8'),
100000)
try:
execute_query(
'INSERT INTO users (username, email, password_hash, salt) VALUES (?, ?, ?, ?)',
(username, email, password_hash.hex(), salt),
commit=True
)
return True
except sqlite3.IntegrityError:
return False
@staticmethod
def authenticate_user(username, password):
user = execute_query(
'SELECT id, password_hash, salt FROM users WHERE username = ?',
(username,),
fetch_one=True
)
if user:
stored_hash = bytes.fromhex(user['password_hash'])
password_hash = hashlib.pbkdf2_hmac('sha256',
password.encode('utf-8'),
user['salt'].encode('utf-8'),
100000)
return stored_hash == password_hash
return False
Caching System:
class SQLiteCache:
def __init__(self, table_name='cache_entries'):
self.table_name = table_name
self._create_table()
def _create_table(self):
execute_query(f'''
CREATE TABLE IF NOT EXISTS {self.table_name} (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
expires_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''', commit=True)
def set(self, key, value, ttl=3600):
expires_at = datetime.now() + timedelta(seconds=ttl)
execute_query(f'''
INSERT OR REPLACE INTO {self.table_name}
(key, value, expires_at) VALUES (?, ?, ?)
''', (key, str(value), expires_at), commit=True)
def get(self, key):
result = execute_query(f'''
SELECT value FROM {self.table_name}
WHERE key = ? AND (expires_at IS NULL OR expires_at > ?)
''', (key, datetime.now()), fetch_one=True)
return result['value'] if result else None
def cleanup_expired(self):
execute_query(f'''
DELETE FROM {self.table_name} WHERE expires_at < ?
''', (datetime.now(),), commit=True)
Performance Comparison and Optimization
SQLite performance in Flask applications depends heavily on proper configuration and usage patterns. Here's a comparison of different approaches:
| Method | Requests/Second | Memory Usage | Connection Overhead | Best For |
|---|---|---|---|---|
| Per-request connections | ~200-500 | Low | High | Simple apps |
| Connection pooling | ~800-1200 | Medium | Low | Production apps |
| Thread-local connections | ~600-900 | Medium | Medium | Multi-threaded apps |
| WAL mode + connection pool | ~1000-1500 | Medium-High | Low | High-concurrency apps |
Optimize SQLite performance with these configuration tweaks:
def optimize_sqlite_connection(conn):
"""Apply performance optimizations to SQLite connection."""
optimizations = [
'PRAGMA journal_mode = WAL', # Write-Ahead Logging
'PRAGMA synchronous = NORMAL', # Faster but still safe
'PRAGMA cache_size = 1000', # Increase cache size
'PRAGMA temp_store = MEMORY', # Store temp tables in RAM
'PRAGMA mmap_size = 268435456', # 256MB memory-mapped I/O
]
for pragma in optimizations:
conn.execute(pragma)
return conn
# Apply optimizations when connecting
def connect_db_optimized():
conn = sqlite3.connect(app.config['DATABASE'])
conn.row_factory = sqlite3.Row
return optimize_sqlite_connection(conn)
Common Pitfalls and Troubleshooting
Several issues frequently plague SQLite-Flask integrations. Here's how to identify and fix them:
Database Lock Issues:
import time
import random
def execute_with_retry(query, params=None, max_retries=5):
"""Execute query with exponential backoff retry on database locks."""
for attempt in range(max_retries):
try:
return execute_query(query, params, commit=True)
except sqlite3.OperationalError as e:
if "database is locked" in str(e) and attempt < max_retries - 1:
wait_time = (2 ** attempt) + random.uniform(0, 1)
time.sleep(wait_time)
continue
raise e
raise sqlite3.OperationalError("Max retries exceeded")
Connection Leak Detection:
import atexit
import weakref
class ConnectionTracker:
def __init__(self):
self.connections = weakref.WeakSet()
def track_connection(self, conn):
self.connections.add(conn)
return conn
def close_all(self):
for conn in self.connections:
try:
conn.close()
except:
pass
print(f"Closed {len(self.connections)} connections")
tracker = ConnectionTracker()
atexit.register(tracker.close_all)
# Modified connection function
def connect_db_tracked():
conn = sqlite3.connect(app.config['DATABASE'])
return tracker.track_connection(conn)
Migration System:
class MigrationManager:
def __init__(self, db_path):
self.db_path = db_path
self._ensure_migration_table()
def _ensure_migration_table(self):
execute_query('''
CREATE TABLE IF NOT EXISTS migrations (
id INTEGER PRIMARY KEY,
filename TEXT UNIQUE NOT NULL,
applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''', commit=True)
def apply_migration(self, filename, sql_content):
try:
with transaction():
execute_query(sql_content, commit=False)
execute_query(
'INSERT INTO migrations (filename) VALUES (?)',
(filename,),
commit=False
)
print(f"Applied migration: {filename}")
return True
except Exception as e:
print(f"Migration failed: {filename} - {e}")
return False
def get_applied_migrations(self):
return execute_query(
'SELECT filename FROM migrations ORDER BY id',
fetch_all=True
)
Best Practices and Security Considerations
Follow these guidelines for robust SQLite-Flask applications:
- Always use parameterized queries to prevent SQL injection attacks
- Implement proper connection lifecycle management with Flask's teardown handlers
- Enable WAL mode for better concurrency in production environments
- Set up regular database backups using SQLite's backup API
- Monitor database file size and implement log rotation for WAL files
- Use database migrations for schema changes instead of manual alterations
- Implement connection timeouts to prevent hanging requests
- Consider using SQLite's built-in encryption for sensitive data
For applications expecting high traffic or requiring guaranteed uptime, consider deploying on VPS hosting or dedicated server infrastructure where you can optimize system-level database performance and implement proper backup strategies.
SQLite with Flask provides an excellent foundation for web applications that need reliable data persistence without the complexity of traditional database servers. The lightweight nature of SQLite makes it perfect for development, testing, and many production scenarios, while Flask's flexibility allows for sophisticated database integration patterns that can scale with your application's needs.
This article incorporates information and material from various online sources. We acknowledge and appreciate the work of all original authors, publishers, and websites. While every effort has been made to appropriately credit the source material, any unintentional oversight or omission does not constitute a copyright infringement. All trademarks, logos, and images mentioned are the property of their respective owners. If you believe that any content used in this article infringes upon your copyright, please contact us immediately for review and prompt action.
This article is intended for informational and educational purposes only and does not infringe on the rights of the copyright owners. If any copyrighted material has been used without proper credit or in violation of copyright laws, it is unintentional and we will rectify it promptly upon notification. Please note that the republishing, redistribution, or reproduction of part or all of the contents in any form is prohibited without express written permission from the author and website owner. For permissions or further inquiries, please contact us.
More stories
