Flask Database with SQLAlchemy
Setting up Flask-SQLAlchemy, defining a model, and a full create/read/update/delete example.
Setting up Flask-SQLAlchemy
Flask doesn't include an ORM — the standard choice is Flask-SQLAlchemy, which wraps SQLAlchemy (Python's most widely used ORM) and integrates it with Flask's application and configuration system.
pip install flask-sqlalchemy
from flask import Flask
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)
SQLALCHEMY_DATABASE_URI points at the database — sqlite:///app.db is a local file, perfect for development; a real deployment typically points this at PostgreSQL or MySQL instead (e.g. postgresql://user:password@localhost/mydb) without changing a single line of model or query code.
Defining a model
A model is a Python class that maps to a database table — each class attribute maps to a column:
class Task(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(120), nullable=False)
done = db.Column(db.Boolean, default=False, nullable=False)
def __repr__(self):
return f"<Task {self.id} {self.title!r}>"
Create the actual table(s) in the database (typically once, or via a migration tool like Flask-Migrate in a real project):
with app.app_context():
db.create_all()
Create
from flask import jsonify, request
@app.route("/tasks", methods=["POST"])
def create_task():
title = request.json.get("title")
task = Task(title=title)
db.session.add(task)
db.session.commit() # writes the INSERT and assigns task.id
return jsonify({"id": task.id, "title": task.title, "done": task.done}), 201
db.session represents the current unit of work — add() stages the new object, and nothing actually reaches the database until commit().
Read
@app.route("/tasks", methods=["GET"])
def list_tasks():
tasks = Task.query.all()
return jsonify([
{"id": t.id, "title": t.title, "done": t.done} for t in tasks
])
@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, "done": task.done})
Task.query is Flask-SQLAlchemy's query interface — .all() fetches every row, .filter_by(done=False) narrows by column value, and get_or_404() looks up a single row by primary key, automatically returning a 404 response if no matching row exists (instead of you writing that None check by hand every time).
Update
@app.route("/tasks/<int:task_id>", methods=["PUT"])
def update_task(task_id):
task = Task.query.get_or_404(task_id)
data = request.json
task.title = data.get("title", task.title)
task.done = data.get("done", task.done)
db.session.commit() # writes the UPDATE
return jsonify({"id": task.id, "title": task.title, "done": task.done})
Because task is an object already tracked by the session (it was loaded through db.session/Task.query), simply changing its attributes and calling commit() is enough — there's no separate "save" call needed.
Delete
@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() # writes the DELETE
return "", 204
A 204 No Content response is the conventional response for a successful delete — the body is intentionally empty.
Common mistakes
- Forgetting
db.session.commit()afteradd()ordelete()— the change is staged in the session but never actually written to the database, and silently disappears when the request ends. - Manually checking
if task is None: return 404everywhere instead of usingget_or_404(), which does the same thing in one call. - Hardcoding
sqlite:///app.dbin the source and forgetting to configure a differentSQLALCHEMY_DATABASE_URIfor production via an environment variable.
Interview questions
Q: What does db.session represent, and why do you have to call .commit()?
db.session is the SQLAlchemy unit of work — a staging area that tracks pending inserts, updates, and deletes. Nothing is sent to the database until commit(), which wraps the pending changes in a transaction and writes them — this lets you make several related changes and have them succeed or fail together, rather than each individual change hitting the database immediately.
Q: How does get_or_404() improve on a plain Task.query.get(id)?
Task.query.get(id) returns None if no row matches, which means every call site needs its own if task is None: return some_404_response check. get_or_404() does that lookup and immediately aborts the request with a proper 404 response if nothing is found, collapsing a repeated pattern into a single expressive call.