Endpoints and Payloads - 实现对API的增删改查 CRUD ()

文件结构：

example
|  models.py
|--flaskr
|   | __init__.py

上代码

# __init__.py
# 版权所有

import os
from flask import Flask, request, abort, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_cors import CORS
import random
from models import setup_db, Book

BOOKS_PER_SHELF = 8

def paginate_books(request, selection):
    page = request.args.get('page', 1, type=int)
    start = (page - 1) * BOOKS_PER_SHELF
    end = start   BOOKS_PER_SHELF

    books = [book.format() for book in selection]
    current_books = books[start:end]

    return current_books

def create_app(test_config=None):
    # create and configue the app
    app = Flask(__name__)
    setup_db(app)
    CORS(app)
    
    # CORS Headers
    @app.after_request
    def after_request(response):
        response.headers.add(
            "Access-Control-Allow-Headers", "Content-Type,Authorization,true"
        )
        response.headers.add(
            "Access-Control-Allow-Methods", "GET,PUT,POST,DELETE,OPTIONS"
        )
        return response
    
    # Retrivies all books, paginated
    @app.route("/books")
    def retrieve_books():
        selection = Book.query.order_by(Book.id).all()
        current_books = paginate_books(request, selection)

        if len(current_books) == 0:
            abort(404)
        
        return jsonify(
            {
                "success": True,
                "books": current_books,
                "total_books": len(Book.query.all())
            }
        )
    
    # Update a single book's rating
    @app.route("/books/<int:book_id>", methods=["PATCH"])
    def update_book(book_id):
        body = request.get_json()

        try:
            book = Book.query.filter(Book.id == book_id).one_or_none()
            if book is None:
                abort(404)
            
            if "rating" in body:
                book.rating = int(body.get("rating"))
            
            book.update()

            return jsonify(
                {
                    'success': True,
                    'id': book.id
                }
            )
        except:
            abort(400)
    
    # Delete a single book
    @app.route("/books/<int:book_id>", methods=["DELETE"])
    def delete_book(book_id):
        try:
            book = Book.query.filter(Book.id == book_id).one_or_none()

            if book is None:
                abort(404)
            
            book.delete()
            selection = Book.query.order_by(Book.id).all()
            current_books = paginate_books(request, selection)

            return jsonify(
                {
                    "success": True,
                    "deleted": book_id,
                    "books": current_books,
                    "total_books": len(Book.query.all())
                }
            )
        except:
            abort(422)
    
    # Create a new book
    @app.route("/books", methods=["POST"])
    def create_book():
        body = request.get_json()

        new_id = body.get("id", None)
        new_title = body.get("title", None)
        new_author = body.get("author", None)
        new_rating = body.get("rating", None)

        try:
            book = Book(id = new_id, title=new_title, author=new_author, rating=new_rating)
            book.insert()

            selection = Book.query.order_by(Book.id).all()
            current_books = paginate_books(request, selection)

            return jsonify(
                {
                    "success": True,
                    "created": book.id,
                    "books": current_books,
                    "total_books": len(Book.query.all()),
                }
            )

        except:
            abort(422)
    
    @app.errorhandler(404)
    def not_found(error):
        return jsonify({
            "success": False,
            "error": 404,
            "message": "resource not found"
        }), 404


    @app.errorhandler(422)
    def unprocessable(error):
        return jsonify({
            "success": False,
            "error": 422,
            "message": "unprocessable"
        }), 422

    @app.errorhandler(400)
    def bad_request(error):
        return jsonify({
            "success": False,
            "error": 400,
            "message": "bad request"
        }), 400

    @app.errorhandler(405)
    def method_not_allowed(error):
        return jsonify({
            "success": False,
            "error": 405,
            "message": "method not allowed"
        }), 405

    return app 

# models.py
# 版权所有

import os
from sqlalchemy import Column, String, Integer, create_engine
from flask_sqlalchemy import SQLAlchemy
import json

database_name = "bookshelf"
database_path = 'postgres://username@localhost:5432/bookshelf'

db = SQLAlchemy()

def setup_db(app, database_path=database_path):
    app.config["SQLALCHEMY_DATABASE_URI"] = database_path
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
    db.app = app
    db.init_app(app)
    db.create_all()

class Book(db.Model):
    __tablename__ = "books"

    id = Column(Integer, primary_key=True)
    title = Column(String)
    author = Column(String)
    rating = Column(Integer)

    def __init__(self, id, title, author, rating):
        self.id = id
        self.title = title
        self.author = author
        self.rating = rating

    def insert(self):
        db.session.add(self)
        db.session.commit()

    def update(self):
        db.session.commit()

    def delete(self):
        db.session.delete(self)
        db.session.commit()

    def format(self):
        return {
            "id": self.id,
            "title": self.title,
            "author": self.author,
            "rating": self.rating,
        }

运行程序步骤：

1. 如果要测试，就得在本地创建数据库，查看如何创建数据库

2. 数据库创建好之后，在Terminal

$ export FLASK_APP=flaskr
$ export FLASK_ENV=development
$ flask run

3. 新开一个Terminal窗口
   实现查

# 查 数据
$ curl http://127.0.0.1:5000/books

结果如下，只截取一部分

{
      "author": "Example", 
      "id": 14, 
      "rating": 4, 
      "title": "Example"
    }
  ], 
  "success": true, 
  "total_books": 9
}

实现改

# 实现 改，只改一个entry
$ curl http://127.0.0.1:5000/books/14 -X PATCH -H "Content-Type: application/json" -d '{"rating":"1"}'

结果如下：

{
  "id": 14, 
  "success": true
}

实现 删除 操作

# 删除 操作
$ curl -X DELETE http://127.0.0.1:5000/books/14

结果如下，只截取一部分

  "deleted": 14, 
  "success": true, 
  "total_books": 8
}

实现 增 操作

# 增 操作
$ curl -X POST -H "Content-Type: application/json" -d '{"id":"20", "title":"new_example", "author":"new example", "rating":"5"}' http://127.0.0.1:5000/books

结果如下，只截取一部分

  "created": 20, 
  "success": true, 
  "total_books": 9
}

如果要判断准确，可以去数据库查看

#  select * from books;

psql 常用命令

这篇好文章是转载于：学新通技术网