Flask Queries
In any web application, the ability to interact with a database is essential. This involves creating records, retrieving information, updating existing data, and removing records when necessary. In Flask applications, we typically use SQLAlchemy as an Object Relational Mapper (ORM) to simplify database operations. This tutorial will guide you through different types of database queries in Flask.
Introduction to Flask Database Queries
Database queries allow your web application to communicate with your database to perform Create, Read, Update, and Delete operations (commonly known as CRUD). When using Flask with SQLAlchemy, you can write queries in a Pythonic way without having to write raw SQL (although that option is still available when needed).
Before diving into queries, let's assume we have the following model setup for a simple blog application:
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
db = SQLAlchemy()
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)
posts = db.relationship('Post', backref='author', lazy=True)
def __repr__(self):
return f'<User {self.username}>'
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)
date_posted = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
def __repr__(self):
return f'<Post {self.title}>'
Basic CRUD Operations
Creating Records
To add new records to your database, you create instances of your model classes, add them to the database session, and commit the changes:
@app.route('/add_user', methods=['POST'])
def add_user():
# Create a new user
new_user = User(username='johndoe', email='[email protected]')
# Add to the session and commit
db.session.add(new_user)
db.session.commit()
return f'User created: {new_user.username}'
You can also add multiple records at once:
@app.route('/add_multiple_users', methods=['POST'])
def add_multiple_users():
users = [
User(username='alice', email='[email protected]'),
User(username='bob', email='[email protected]'),
User(username='charlie', email='[email protected]')
]
db.session.add_all(users)
db.session.commit()
return 'Multiple users added!'
Reading Records
The most common operation in web applications is reading data. SQLAlchemy provides several methods to query your database:
Retrieving All Records
@app.route('/users')
def get_all_users():
users = User.query.all()
result = ''
for user in users:
result += f'Username: {user.username}, Email: {user.email}<br>'
return result
Retrieving a Single Record
You can fetch a specific record by its primary key using get():
@app.route('/user/<int:user_id>')
def get_user(user_id):
user = User.query.get(user_id)
if user:
return f'User found: {user.username}, {user.email}'
else:
return 'User not found', 404
Filtering Records
To find records matching specific criteria, use filter() or filter_by():
@app.route('/user/find/<username>')
def find_user(username):
# Using filter_by (for equality conditions)
user = User.query.filter_by(username=username).first()
if user:
return f'User found: {user.username}, {user.email}'
else:
return 'User not found', 404
For more complex filters:
@app.route('/posts/search/<keyword>')
def search_posts(keyword):
# Using filter (for more complex conditions)
posts = Post.query.filter(Post.content.contains(keyword)).all()
result = f'Found {len(posts)} posts containing "{keyword}":<br>'
for post in posts:
result += f'- {post.title}<br>'
return result
Updating Records
To update existing records, you first query for them, modify their attributes, and then commit the changes:
@app.route('/user/update/<int:user_id>', methods=['POST'])
def update_user(user_id):
user = User.query.get(user_id)
if not user:
return 'User not found', 404
# Update the user's email
user.email = '[email protected]'
db.session.commit()
return f'User {user.username} updated with email: {user.email}'
Deleting Records
To delete records, you query for them and then use delete():
@app.route('/user/delete/<int:user_id>', methods=['POST'])
def delete_user(user_id):
user = User.query.get(user_id)
if not user:
return 'User not found', 404
db.session.delete(user)
db.session.commit()
return f'User {user.username} has been deleted'
Advanced Queries
Ordering Results
You can order your query results using order_by():
@app.route('/posts/newest')
def newest_posts():
# Get posts ordered by date (newest first)
posts = Post.query.order_by(Post.date_posted.desc()).all()
result = 'Newest posts:<br>'
for post in posts:
result += f'- {post.title} (Posted on: {post.date_posted})<br>'
return result
Limiting Results
To limit the number of results returned, use limit():
@app.route('/posts/recent')
def recent_posts():
# Get only the 5 most recent posts
posts = Post.query.order_by(Post.date_posted.desc()).limit(5).all()
result = '5 most recent posts:<br>'
for post in posts:
result += f'- {post.title} (Posted on: {post.date_posted})<br>'
return result
Pagination
For large datasets, you'll want to implement pagination:
@app.route('/posts/page/<int:page_num>')
def paginated_posts(page_num):
# Get posts with pagination (10 posts per page)
paginated = Post.query.order_by(Post.date_posted.desc()).paginate(
page=page_num, per_page=10, error_out=False
)
result = f'Page {page_num} of {paginated.pages}:<br>'
for post in paginated.items:
result += f'- {post.title}<br>'
if paginated.has_next:
result += f'<a href="/posts/page/{page_num + 1}">Next Page</a>'
if paginated.has_prev:
result += f'<a href="/posts/page/{page_num - 1}">Previous Page</a>'
return result
Joining Tables
To query across relationships, you can use joins:
@app.route('/users/with_posts')
def users_with_posts():
# Find users who have at least one post
users = User.query.join(Post).group_by(User.id).all()
result = 'Users who have written posts:<br>'
for user in users:
result += f'- {user.username} (Posts: {len(user.posts)})<br>'
return result
Aggregations and Counts
You can perform aggregations like counting:
from sqlalchemy import func
@app.route('/stats')
def stats():
# Count total users and posts
user_count = db.session.query(func.count(User.id)).scalar()
post_count = db.session.query(func.count(Post.id)).scalar()
# Calculate average posts per user
avg_posts = post_count / user_count if user_count > 0 else 0
return f'''
Statistics:
- Total users: {user_count}
- Total posts: {post_count}
- Average posts per user: {avg_posts:.2f}
'''
Real-World Application Example
Let's put it all together with a more complete example of a blog application with proper routes:
from flask import Flask, render_template, request, redirect, url_for, flash
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///blog.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.secret_key = 'some_secret_key'
db = SQLAlchemy(app)
# Models defined here (User and Post)
@app.route('/')
def home():
# Get the 5 most recent posts for the homepage
recent_posts = Post.query.order_by(Post.date_posted.desc()).limit(5).all()
return render_template('home.html', posts=recent_posts)
@app.route('/posts')
def all_posts():
page = request.args.get('page', 1, type=int)
posts = Post.query.order_by(Post.date_posted.desc()).paginate(
page=page, per_page=10
)
return render_template('all_posts.html', posts=posts)
@app.route('/post/<int:post_id>')
def view_post(post_id):
post = Post.query.get_or_404(post_id)
return render_template('post_detail.html', post=post)
@app.route('/create_post', methods=['GET', 'POST'])
def create_post():
if request.method == 'POST':
title = request.form['title']
content = request.form['content']
user_id = 1 # In a real app, this would be the logged-in user's ID
new_post = Post(title=title, content=content, user_id=user_id)
db.session.add(new_post)
db.session.commit()
flash('Your post has been created!', 'success')
return redirect(url_for('home'))
return render_template('create_post.html')
@app.route('/post/<int:post_id>/update', methods=['GET', 'POST'])
def update_post(post_id):
post = Post.query.get_or_404(post_id)
# In a real app, check if the current user is the author
if request.method == 'POST':
post.title = request.form['title']
post.content = request.form['content']
db.session.commit()
flash('Your post has been updated!', 'success')
return redirect(url_for('view_post', post_id=post.id))
return render_template('update_post.html', post=post)
@app.route('/post/<int:post_id>/delete', methods=['POST'])
def delete_post(post_id):
post = Post.query.get_or_404(post_id)
# In a real app, check if the current user is the author
db.session.delete(post)
db.session.commit()
flash('Your post has been deleted!', 'success')
return redirect(url_for('home'))
@app.route('/search')
def search():
query = request.args.get('q', '')
if query:
posts = Post.query.filter(
Post.title.contains(query) | Post.content.contains(query)
).all()
else:
posts = []
return render_template('search_results.html', posts=posts, query=query)
if __name__ == '__main__':
app.run(debug=True)
In this example, we've created a simple blog application with routes for viewing posts, creating posts, updating posts, deleting posts, and searching.
Error Handling and Transactions
When working with databases, it's important to handle errors and use transactions properly:
@app.route('/create_post_with_tags', methods=['POST'])
def create_post_with_tags():
try:
# Start a transaction
title = request.form['title']
content = request.form['content']
user_id = 1 # In a real app, this would be the logged-in user's ID
# Create the post
new_post = Post(title=title, content=content, user_id=user_id)
db.session.add(new_post)
# Create some tags (this would be more dynamic in a real app)
# This assumes you have a Tag model and a post_tags association table
tag1 = Tag.query.get(1)
tag2 = Tag.query.get(2)
new_post.tags.extend([tag1, tag2])
# Commit everything as one transaction
db.session.commit()
return 'Post created with tags!'
except Exception as e:
# If anything goes wrong, roll back the transaction
db.session.rollback()
return f'Error: {str(e)}', 500
Raw SQL Queries
While SQLAlchemy ORM is powerful, sometimes you might need to execute raw SQL queries:
@app.route('/raw_sql_example')
def raw_sql_example():
# Execute a raw SQL query
sql = "SELECT users.username, COUNT(posts.id) as post_count FROM users " \
"LEFT JOIN posts ON users.id = posts.user_id " \
"GROUP BY users.id ORDER BY post_count DESC"
result = db.engine.execute(sql)
output = 'Users by post count:<br>'
for row in result:
output += f'- {row.username}: {row.post_count} posts<br>'
return output
Summary
In this tutorial, we've covered:
- Basic CRUD operations with SQLAlchemy in Flask
- Advanced queries including filtering, ordering, and pagination
- Joining tables and performing aggregations
- A real-world example of a blog application
- Error handling and transactions
- Raw SQL queries when needed
Flask-SQLAlchemy makes database operations straightforward, allowing you to focus on building your application logic rather than dealing with low-level database details. By mastering these query techniques, you'll be able to build robust Flask applications that efficiently interact with your database.
Additional Resources and Exercises
Resources
Exercises
-
Exercise 1: Create a simple Flask application with a
Taskmodel that includes fields for title, description, creation date, and completion status. Implement routes to create, list, update, and delete tasks. -
Exercise 2: Extend the application to include a
Categorymodel related toTask. Implement filtering of tasks by category. -
Exercise 3: Add pagination to your task listing page, showing 10 tasks per page.
-
Exercise 4: Implement a search feature that allows users to find tasks by keywords in the title or description.
-
Exercise 5: Create a dashboard that shows statistics about tasks: total tasks, completed tasks percentage, tasks by category, etc.
By completing these exercises, you'll gain valuable hands-on experience with Flask database queries and deepen your understanding of SQLAlchemy's capabilities.
If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)