Add SQLAlchemy support to your Flask application.
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Enhanced query classes with Flask-specific convenience methods for common web patterns like 404 error handling and result processing.
Enhanced SQLAlchemy Query with Flask convenience methods for web applications.
class Query(sa_orm.Query):
"""
SQLAlchemy Query subclass with Flask-specific convenience methods.
Provides common web application patterns like automatic 404 error
handling for missing results. This is the default query class for
Model.query properties.
Note: The query interface is considered legacy in SQLAlchemy.
Modern code should prefer session.execute(select()) patterns.
"""
def get_or_404(self, ident: Any, description: str | None = None) -> Any:
"""
Get record by primary key or abort with 404 Not Found.
Parameters:
- ident: Primary key value to lookup
- description: Custom error message for 404 page
Returns:
Model instance matching the primary key
Raises:
404 error if no record found with given primary key
"""
def first_or_404(self, description: str | None = None) -> Any:
"""
Get first result from query or abort with 404 Not Found.
Parameters:
- description: Custom error message for 404 page
Returns:
First model instance from query results
Raises:
404 error if query returns no results
"""
def one_or_404(self, description: str | None = None) -> Any:
"""
Get exactly one result from query or abort with 404 Not Found.
Parameters:
- description: Custom error message for 404 page
Returns:
Single model instance from query results
Raises:
404 error if query returns no results or multiple results
"""
def paginate(
self,
*,
page: int | None = None,
per_page: int | None = None,
max_per_page: int | None = None,
error_out: bool = True,
count: bool = True,
) -> Pagination:
"""
Apply pagination to the query results.
Parameters:
- page: Current page number (from request args if None)
- per_page: Items per page (from request args if None)
- max_per_page: Maximum allowed per_page value
- error_out: Abort with 404 on invalid page parameters
- count: Calculate total item count for pagination metadata
Returns:
Pagination object containing page results and metadata
"""The _QueryProperty class that provides the query attribute on model classes.
class _QueryProperty:
"""
Class property that creates query objects for model classes.
Internal implementation detail that provides Model.query property.
"""
def __get__(self, obj: Model | None, cls: type[Model]) -> Query:
"""
Return configured query instance for the model class.
Parameters:
- obj: Model instance (unused)
- cls: Model class to create query for
Returns:
Query instance configured for the model class
"""# Model with automatic query property
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True)
email = db.Column(db.String(120))
# Basic querying
users = User.query.all()
user = User.query.get(1)
active_users = User.query.filter_by(active=True).all()@app.route('/user/<int:user_id>')
def show_user(user_id):
# Automatically returns 404 if user doesn't exist
user = User.query.get_or_404(user_id)
return f"User: {user.username}"
@app.route('/user/first-admin')
def first_admin():
# Get first admin user or 404 if none exist
admin = User.query.filter_by(role='admin').first_or_404()
return f"First admin: {admin.username}"
@app.route('/user/email/<email>')
def user_by_email(email):
# Expect exactly one user with this email or 404
user = User.query.filter_by(email=email).one_or_404()
return f"User: {user.username}"@app.route('/user/<int:user_id>')
def show_user(user_id):
user = User.query.get_or_404(
user_id,
description=f"User {user_id} not found in our system"
)
return f"User: {user.username}"@app.route('/users')
def list_users():
page = request.args.get('page', 1, type=int)
# Paginate query results
pagination = User.query.order_by(User.username).paginate(
page=page,
per_page=20,
max_per_page=100,
error_out=False # Don't 404 on invalid page, use page 1
)
return {
'users': [u.username for u in pagination.items],
'page': pagination.page,
'pages': pagination.pages,
'has_next': pagination.has_next,
'has_prev': pagination.has_prev
}# Filtering and ordering
recent_posts = Post.query.filter(
Post.created_at > datetime.utcnow() - timedelta(days=7)
).order_by(Post.created_at.desc()).all()
# Joins and relationships
posts_with_authors = Post.query.join(User).filter(
User.active == True
).order_by(Post.title).all()
# Aggregation
user_post_counts = db.session.query(
User.username,
db.func.count(Post.id).label('post_count')
).join(Post).group_by(User.id).all()# Custom query class
class CustomQuery(Query):
def active(self):
return self.filter_by(active=True)
def by_username(self, username):
return self.filter_by(username=username)
# Use custom query class
db = SQLAlchemy(app, query_class=CustomQuery)
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80))
active = db.Column(db.Boolean, default=True)
# Now all queries have custom methods
active_users = User.query.active().all()
user = User.query.by_username('john').first_or_404()While the Query interface is available, SQLAlchemy 2.x style is preferred:
# Legacy Query style (still supported)
users = User.query.filter_by(active=True).all()
# Modern select() style (recommended)
users = db.session.execute(
db.select(User).where(User.active == True)
).scalars().all()
# Both can use Flask conveniences
user = db.get_or_404(User, user_id) # Modern
user = User.query.get_or_404(user_id) # Legacyclass User(db.Model):
id = db.Column(db.Integer, primary_key=True)
posts = db.relationship('Post', backref='author', lazy='dynamic')
class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
# Query through relationships
user = User.query.get(1)
user_posts = user.posts.filter_by(published=True).all()
# Backref queries
post = Post.query.get(1)
post_author = post.author # Automatically loadeddb.session.execute(db.select()) for new codejoinedload() or selectinload() to avoid N+1 queriesInstall with Tessl CLI
npx tessl i tessl/pypi-flask-sqlalchemy