Last active
September 27, 2021 14:48
-
-
Save caseydm/4fc0edfb05cadd49f2784cd6359d231c to your computer and use it in GitHub Desktop.
Flask API Pagination
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from flask import Flask | |
| from flask_marshmallow import Marshmallow | |
| from flask_sqlalchemy import SQLAlchemy | |
| from models import Magazine | |
| from utils import build_link_header, validate_per_page | |
| app = Flask(__name__) | |
| db = SQLAlchemy(app) | |
| ma = Marshmallow(app) | |
| app.config["SECRET_KEY"] = os.getenv("SECRET_KEY") | |
| app.config["SQLALCHEMY_DATABASE_URI"] = os.getenv("DATABASE_URL") | |
| @app.route("/magazines") | |
| def magazines(): | |
| # process query parameters | |
| page = request.args.get("page", 1, type=int) | |
| per_page = validate_per_page(request.args.get("per-page", 100, type=int)) | |
| # query | |
| magazines = Magazine.query.paginate(page, per_page) | |
| # map with schema | |
| magazine_schema = MagazineSchema() | |
| magazines_dumped = magazine_schema.dump(magazines.items, many=True) | |
| # combined results with pagination | |
| results = { | |
| "results": magazines_dumped, | |
| "pagination": | |
| { | |
| "count": magazines.total, | |
| "page": page, | |
| "per_page": per_page, | |
| "pages": magazines.pages, | |
| }, | |
| } | |
| # paginated link headers | |
| base_url = "https://api.mysite.org/magazines" | |
| link_header = build_link_header( | |
| query=magazines, base_url=base_url, per_page=per_page | |
| ) | |
| return jsonify(results), 200, link_header | |
| @app.errorhandler(APIError) | |
| def handle_exception(err): | |
| """Return custom JSON when APIError or its children are raised""" | |
| # credit: https://medium.com/datasparq-technology/flask-api-exception-handling-with-custom-http-response-codes-c51a82a51a0f | |
| response = {"error": err.description, "message": ""} | |
| if len(err.args) > 0: | |
| response["message"] = err.args[0] | |
| # Add some logging so that we can monitor different types of errors | |
| app.logger.error("{}: {}".format(err.description, response["message"])) | |
| return jsonify(response), err.code |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| class APIError(Exception): | |
| """All custom API Exceptions""" | |
| pass | |
| class APIPaginationError(APIError): | |
| """Error when per-page parameter is out of bounds.""" | |
| code = 403 | |
| description = "pagination error" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from app import db | |
| class Magazine(db.Model): | |
| __tablename__ = "magazines" | |
| id = db.Column(db.Integer, primary_key=True) | |
| publisher = db.Column(db.String(100), nullable=False) | |
| title = db.Column(db.String(200), nullable=False) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from app import ma | |
| from models import Magazine | |
| class MagazineSchema(ma.SQLAlchemyAutoSchema): | |
| class Meta: | |
| model = Magazine | |
| fields = ("title", "publisher") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from exceptions import APIPaginationError | |
| def build_link_header(query, base_url, per_page): | |
| """ | |
| Adds pagination link headers to an API response. | |
| """ | |
| links = [ | |
| '<{0}?page=1&per-page={1}>; rel="first"'.format(base_url, per_page), | |
| '<{0}?page={1}&per-page={2}>; rel="last"'.format( | |
| base_url, query.pages, per_page | |
| ), | |
| ] | |
| if query.has_prev: | |
| links.append( | |
| '<{0}?page={1}&per-page={2}>; rel="prev"'.format( | |
| base_url, query.prev_num, per_page | |
| ) | |
| ) | |
| if query.has_next: | |
| links.append( | |
| '<{0}?page={1}&per-page={2}>; rel="next"'.format( | |
| base_url, query.next_num, per_page | |
| ) | |
| ) | |
| links = ",".join(links) | |
| return dict(Link=links) | |
| def validate_per_page(per_page): | |
| if per_page and per_page > 100 or per_page < 1: | |
| raise APIPaginationError("per-page parameter must be between 1 and 100") | |
| return per_page |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment