Quotes API (Flask + SQLAlchemy)¶
Below is a full example of a REST API for a quotes app using Flask and SQLAlchemy with marshmallow. It demonstrates a number of features, including:
Custom validation
Nesting fields
Using
dump_only=True
to specify read-only fieldsOutput filtering using the
only
parameterUsing
@pre_load
to preprocess input data.
# /// script
# requires-python = ">=3.9"
# dependencies = [
# "flask",
# "flask-sqlalchemy>=3.1.1",
# "marshmallow",
# "sqlalchemy>2.0",
# ]
# ///
from __future__ import annotations
import datetime
from flask import Flask, request
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.exc import NoResultFound
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from marshmallow import Schema, ValidationError, fields, pre_load
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:////tmp/quotes.db"
class Base(DeclarativeBase):
pass
db = SQLAlchemy(app, model_class=Base)
##### MODELS #####
class Author(db.Model): # type: ignore[name-defined]
id: Mapped[int] = mapped_column(primary_key=True)
first: Mapped[str]
last: Mapped[str]
class Quote(db.Model): # type: ignore[name-defined]
id: Mapped[int] = mapped_column(primary_key=True)
content: Mapped[str] = mapped_column(nullable=False)
author_id: Mapped[int] = mapped_column(db.ForeignKey(Author.id))
author: Mapped[Author] = relationship(backref=db.backref("quotes", lazy="dynamic"))
posted_at: Mapped[datetime.datetime]
##### SCHEMAS #####
class AuthorSchema(Schema):
id = fields.Int(dump_only=True)
first = fields.Str()
last = fields.Str()
formatted_name = fields.Method("format_name", dump_only=True)
def format_name(self, author):
return f"{author.last}, {author.first}"
# Custom validator
def must_not_be_blank(data):
if not data:
raise ValidationError("Data not provided.")
class QuoteSchema(Schema):
id = fields.Int(dump_only=True)
author = fields.Nested(AuthorSchema, validate=must_not_be_blank)
content = fields.Str(required=True, validate=must_not_be_blank)
posted_at = fields.DateTime(dump_only=True)
# Allow client to pass author's full name in request body
# e.g. {"author': 'Tim Peters"} rather than {"first": "Tim", "last": "Peters"}
@pre_load
def process_author(self, data, **kwargs):
author_name = data.get("author")
if author_name:
first, last = author_name.split(" ")
author_dict = {"first": first, "last": last}
else:
author_dict = {}
data["author"] = author_dict
return data
author_schema = AuthorSchema()
authors_schema = AuthorSchema(many=True)
quote_schema = QuoteSchema()
quotes_schema = QuoteSchema(many=True, only=("id", "content"))
##### API #####
@app.route("/authors")
def get_authors():
authors = Author.query.all()
# Serialize the queryset
result = authors_schema.dump(authors)
return {"authors": result}
@app.route("/authors/<int:pk>")
def get_author(pk):
try:
author = Author.query.filter(Author.id == pk).one()
except NoResultFound:
return {"message": "Author could not be found."}, 400
author_result = author_schema.dump(author)
quotes_result = quotes_schema.dump(author.quotes.all())
return {"author": author_result, "quotes": quotes_result}
@app.route("/quotes/", methods=["GET"])
def get_quotes():
quotes = Quote.query.all()
result = quotes_schema.dump(quotes, many=True)
return {"quotes": result}
@app.route("/quotes/<int:pk>")
def get_quote(pk):
try:
quote = Quote.query.filter(Quote.id == pk).one()
except NoResultFound:
return {"message": "Quote could not be found."}, 400
result = quote_schema.dump(quote)
return {"quote": result}
@app.route("/quotes/", methods=["POST"])
def new_quote():
json_data = request.get_json()
if not json_data:
return {"message": "No input data provided"}, 400
# Validate and deserialize input
try:
data = quote_schema.load(json_data)
except ValidationError as err:
return err.messages, 422
first, last = data["author"]["first"], data["author"]["last"]
author = Author.query.filter_by(first=first, last=last).first()
if author is None:
# Create a new author
author = Author(first=first, last=last)
db.session.add(author)
# Create new quote
quote = Quote(
content=data["content"],
author=author,
posted_at=datetime.datetime.now(datetime.UTC),
)
db.session.add(quote)
db.session.commit()
result = quote_schema.dump(Quote.query.get(quote.id))
return {"message": "Created new quote.", "quote": result}
if __name__ == "__main__":
with app.app_context():
db.create_all()
app.run(debug=True, port=5000)
Using The API
Run the app.
$ uv run examples/flask_example.py
We’ll use the httpie cli to send requests
Install it with uv
.
$ uv tool install httpie
First we’ll POST some quotes.
$ http POST :5000/quotes/ author="Tim Peters" content="Beautiful is better than ugly."
$ http POST :5000/quotes/ author="Tim Peters" content="Now is better than never."
$ http POST :5000/quotes/ author="Peter Hintjens" content="Simplicity is always better than functionality."
If we provide invalid input data, we get 400 error response. Let’s omit “author” from the input data.
$ http POST :5000/quotes/ content="I have no author"
{
"author": [
"Data not provided."
]
}
Now we can GET a list of all the quotes.
$ http :5000/quotes/
{
"quotes": [
{
"content": "Beautiful is better than ugly.",
"id": 1
},
{
"content": "Now is better than never.",
"id": 2
},
{
"content": "Simplicity is always better than functionality.",
"id": 3
}
]
}
We can also GET the quotes for a single author.
$ http :5000/authors/1
{
"author": {
"first": "Tim",
"formatted_name": "Peters, Tim",
"id": 1,
"last": "Peters"
},
"quotes": [
{
"content": "Beautiful is better than ugly.",
"id": 1
},
{
"content": "Now is better than never.",
"id": 2
}
]
}