flask_sqlalchemy - dwilson2547/wiki_demo GitHub Wiki
Flask-SQLAlchemy is an extension for Flask that adds SQLAlchemy support to your application. It simplifies database interactions by providing a high-level ORM (Object-Relational Mapping) interface, allowing you to define database models as Python classes and interact with your database using Python objects instead of writing raw SQL.
- 1. What is Flask-SQLAlchemy?
- 2. Key Features
- 3. Setting Up Flask-SQLAlchemy
- 4. CRUD Operations with Flask-SQLAlchemy
- 5. Relationships in Flask-SQLAlchemy
- 6. Advanced Querying
- 7. Flask-SQLAlchemy with Flask-RESTful
- 8. Performance Tips
- 9. Error Handling
- 10. Flask-SQLAlchemy vs. Other ORMs
- 11. When to Use Flask-SQLAlchemy
- 12. Example: Full CRUD API with Flask-SQLAlchemy
- 13. Learning Resources
- 14. Summary
Flask-SQLAlchemy is a wrapper around SQLAlchemy that integrates seamlessly with Flask. It provides:
- Declarative base class for defining models.
- Session management for database transactions.
- Connection pooling and engine configuration.
- Migrations support (via Flask-Migrate).
- Define database tables as Python classes.
- Automatic table creation based on model definitions.
- Query interface for retrieving and manipulating data.
- Base class for defining models.
-
Example:
from flask_sqlalchemy import SQLAlchemy db = SQLAlchemy(app) 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}>'
- Automatic session handling for database transactions.
-
Example:
user = User(username='alice', email='[email protected]') db.session.add(user) db.session.commit()
- Define one-to-many, many-to-one, and many-to-many relationships.
-
Example (One-to-Many):
class Post(db.Model): id = db.Column(db.Integer, primary_key=True) title = db.Column(db.String(80), nullable=False) body = db.Column(db.Text, nullable=False) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) user = db.relationship('User', backref=db.backref('posts', lazy=True)) # User has many posts (accessible via user.posts)
- Powerful querying with SQLAlchemy’s query API.
-
Example:
# Get all users users = User.query.all() # Filter users by username alice = User.query.filter_by(username='alice').first() # Complex queries from sqlalchemy import or_ users = User.query.filter(or_(User.username == 'alice', User.username == 'bob')).all()
- Database schema migrations using Alembic.
-
Example:
pip install flask-migrate
from flask_migrate import Migrate migrate = Migrate(app, db)
flask db init # Initialize migrations flask db migrate # Create a migration flask db upgrade # Apply the migration
pip install flask-sqlalchemyfrom flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db' # SQLite database file
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False # Disable modification tracking
db = SQLAlchemy(app)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)
class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(80), nullable=False)
body = db.Column(db.Text, nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)flask shell
>>> db.create_all()# Create a new user
new_user = User(username='alice', email='[email protected]')
db.session.add(new_user)
db.session.commit()
# Create a new post
new_post = Post(title='First Post', body='Hello, world!', user_id=1)
db.session.add(new_post)
db.session.commit()# Get all users
users = User.query.all()
# Get a user by ID
user = User.query.get(1)
# Filter users
alice = User.query.filter_by(username='alice').first()
# Join queries
posts = Post.query.join(User).filter(User.username == 'alice').all()user = User.query.get(1)
user.email = '[email protected]'
db.session.commit()user = User.query.get(1)
db.session.delete(user)
db.session.commit()class User(db.Model):
# ... (previous fields)
posts = db.relationship('Post', backref='author', lazy=True)
class Post(db.Model):
# ... (previous fields)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)Usage:
# Get all posts by a user
user = User.query.get(1)
posts = user.posts # Access via the relationship# Association table
post_tags = db.Table('post_tags',
db.Column('post_id', db.Integer, db.ForeignKey('post.id'), primary_key=True),
db.Column('tag_id', db.Integer, db.ForeignKey('tag.id'), primary_key=True)
)
class Post(db.Model):
# ... (previous fields)
tags = db.relationship('Tag', secondary=post_tags, lazy='subquery',
backref=db.backref('posts', lazy=True))
class Tag(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), unique=True, nullable=False)Usage:
# Add a tag to a post
post = Post.query.get(1)
tag = Tag(name='python')
post.tags.append(tag)
db.session.commit()from sqlalchemy import or_, and_, not_
# OR condition
users = User.query.filter(or_(User.username == 'alice', User.username == 'bob')).all()
# AND condition
users = User.query.filter(and_(User.username != 'alice', User.email.like('%@example.com'))).all()
# NOT condition
users = User.query.filter(not_(User.username == 'alice')).all()# Order by username
users = User.query.order_by(User.username).all()
# Limit results
users = User.query.limit(5).all()
# Offset and limit (pagination)
users = User.query.offset(10).limit(5).all()from sqlalchemy import func
# Count users
user_count = db.session.query(func.count(User.id)).scalar()
# Group by
from sqlalchemy import desc
user_counts = db.session.query(User.username, func.count(Post.id)).join(Post).group_by(User.username).order_by(desc(func.count(Post.id))).all()pip install flask-restfulfrom flask_restful import Api, Resource
api = Api(app)
class UserResource(Resource):
def get(self, user_id):
user = User.query.get_or_404(user_id)
return {'username': user.username, 'email': user.email}
def delete(self, user_id):
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return {'message': 'User deleted'}, 200
api.add_resource(UserResource, '/users/<int:user_id>')-
Lazy Loading (Default): Loads related data only when accessed.
user = User.query.get(1) posts = user.posts # Separate query to load posts
-
Eager Loading: Loads related data in the same query.
user = User.query.options(db.joinedload(User.posts)).get(1) posts = user.posts # No additional query
-
Add indexes to frequently queried columns.
class User(db.Model): username = db.Column(db.String(80), unique=True, nullable=False, index=True)
- Configure SQLAlchemy connection pooling for better performance.
app.config['SQLALCHEMY_POOL_SIZE'] = 20 app.config['SQLALCHEMY_MAX_OVERFLOW'] = 10 app.config['SQLALCHEMY_POOL_TIMEOUT'] = 30 app.config['SQLALCHEMY_POOL_RECYCLE'] = 3600
| Error | Cause | Solution |
|---|---|---|
OperationalError |
Database connection issues | Check database URI and server status |
IntegrityError |
Violation of database constraints | Handle unique/foreign key violations |
DataError |
Invalid data types | Validate input data |
InterfaceError |
Database interface issues | Check database driver and connection |
from sqlalchemy.exc import SQLAlchemyError
from flask import jsonify
@app.errorhandler(SQLAlchemyError)
def handle_db_error(e):
return jsonify({'error': 'Database error', 'message': str(e)}), 500| Feature | Flask-SQLAlchemy | Django ORM | SQLAlchemy Core |
|---|---|---|---|
| Integration | Flask-specific | Django-specific | Framework-agnostic |
| Ease of Use | High (Flask-friendly) | High (Django-friendly) | Moderate (more manual setup) |
| Flexibility | High (full SQLAlchemy power) | Moderate (Django conventions) | Highest (low-level control) |
| Migrations | Flask-Migrate (Alembic) | Built-in | Alembic |
| Performance | Good | Good | Best (optimized queries) |
| Use Cases | Flask apps, small to medium projects | Django apps | Complex queries, performance-critical apps |
✅ Flask Applications: When building web apps or APIs with Flask. ✅ Rapid Prototyping: Quickly define models and interact with the database. ✅ Flexibility Needed: When you need both high-level ORM and low-level SQL control. ✅ Small to Medium Projects: Ideal for projects that don’t require Django’s "batteries-included" approach. ✅ Learning SQLAlchemy: A great introduction to SQLAlchemy’s ORM features.
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
class Task(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(100), nullable=False)
description = db.Column(db.Text)
done = db.Column(db.Boolean, default=False)
def __repr__(self):
return f'<Task {self.title}>'
with app.app_context():
db.create_all()# Create a task
@app.route('/tasks', methods=['POST'])
def create_task():
data = request.get_json()
task = Task(title=data['title'], description=data.get('description', ''))
db.session.add(task)
db.session.commit()
return jsonify({'id': task.id, 'title': task.title}), 201
# Get all tasks
@app.route('/tasks', methods=['GET'])
def get_tasks():
tasks = Task.query.all()
return jsonify([{'id': task.id, 'title': task.title, 'done': task.done} for task in tasks])
# Get a single task
@app.route('/tasks/<int:task_id>', methods=['GET'])
def get_task(task_id):
task = Task.query.get_or_404(task_id)
return jsonify({'id': task.id, 'title': task.title, 'description': task.description, 'done': task.done})
# Update a task
@app.route('/tasks/<int:task_id>', methods=['PUT'])
def update_task(task_id):
task = Task.query.get_or_404(task_id)
data = request.get_json()
task.title = data.get('title', task.title)
task.description = data.get('description', task.description)
task.done = data.get('done', task.done)
db.session.commit()
return jsonify({'id': task.id, 'title': task.title, 'done': task.done})
# Delete a task
@app.route('/tasks/<int:task_id>', methods=['DELETE'])
def delete_task(task_id):
task = Task.query.get_or_404(task_id)
db.session.delete(task)
db.session.commit()
return jsonify({'message': 'Task deleted'}), 200flask run-
Create a Task:
curl -X POST -H "Content-Type: application/json" -d '{"title": "Learn Flask", "description": "Study Flask-SQLAlchemy"}' http://127.0.0.1:5000/tasks
-
Get All Tasks:
curl http://127.0.0.1:5000/tasks
-
Update a Task:
curl -X PUT -H "Content-Type: application/json" -d '{"done": true}' http://127.0.0.1:5000/tasks/1
- Official Documentation: flask-sqlalchemy.palletsprojects.com
- Tutorials:
-
Books:
- Essential SQLAlchemy by Rick Copeland
- Flask Web Development by Miguel Grinberg
- Courses:
- Flask-SQLAlchemy is a powerful, flexible ORM extension for Flask applications.
- Key Features: ORM, declarative models, relationships, querying, and migrations.
- Strengths: Easy to use, integrates seamlessly with Flask, and provides full SQLAlchemy power.
- Weaknesses: Requires understanding of SQLAlchemy’s concepts (e.g., sessions, relationships).
- Use Cases: Ideal for Flask applications needing database interactions, from small projects to medium-sized APIs.
Flask-SQLAlchemy is the go-to choice for Flask developers who want a robust, Pythonic way to interact with databases without writing raw SQL. It strikes a balance between simplicity and power, making it suitable for both beginners and experienced developers.