Build a RESTful API With Flask, MongoDB, and Python

1
pip install -e .

1
export MONGO_URI="mongodb+srv://USERNAME:PASSW0RD@cluster0-abcde.azure.mongodb.net/cocktails?retryWrites=true&w=majority"

1
mongoimport --uri "$MONGO_URI" --file ./recipes.json

1
FLASK_DEBUG=true FLASK_APP=cocktailapi flask run

1
{
2
    "_links": {
3
        "last": {
4
            "href": "http://localhost:5000/cocktails/?page=5"
5
        }, 
6
        "next": {
7
            "href": "http://localhost:5000/cocktails/?page=5"
8
        }, 
9
        "prev": {
10
            "href": "http://localhost:5000/cocktails/?page=3"
11
        }, 
12
        "self": {
13
            "href": "http://localhost:5000/cocktails/?page=4"
14
        }
15
    }, 
16
    "recipes": [
17
        {
18
            "_id": "5f7daa198ec9dfb536781b0d", 
19
            "date_added": null, 
20
            "date_updated": null, 
21
            "ingredients": [
22
            {
23
                "name": "Light rum", 
24
                "quantity": {
25
                "unit": "oz", 
26
                }
27
            }, 
28
            {
29
                "name": "Grapefruit juice", 
30
                "quantity": {
31
                "unit": "oz", 
32
                }
33
            }, 
34
            {
35
                "name": "Bitters", 
36
                "quantity": {
37
                "unit": "dash", 
38
                }
39
            }
40
            ], 
41
            "instructions": [
42
            "Pour all of the ingredients into an old-fashioned glass almost filled with ice cubes", 
43
            "Stir well."
44
            ], 
45
            "name": "Monkey Wrench", 
46
            "slug": "monkey-wrench"
47
        },
48
    ]
49
    ...

1
# model.py
2
3
class Cocktail(BaseModel):
4
    id: Optional[PydanticObjectId] = Field(None, alias="_id")
5
    slug: str
6
    name: str
7
    ingredients: List[Ingredient]
8
    instructions: List[str]
9
    date_added: Optional[datetime]
10
    date_updated: Optional[datetime]
11
12
    def to_json(self):
13
        return jsonable_encoder(self, exclude_none=True)
14
15
    def to_bson(self):
16
        data = self.dict(by_alias=True, exclude_none=True)
17
        if data["_id"] is None:
18
            data.pop("_id")
19
        return data

1
# objectid.py
2
3
class PydanticObjectId(ObjectId):
4
    """
5
    ObjectId field. Compatible with Pydantic.
6
    """
7
8
    @classmethod
9
    def __get_validators__(cls):
10
        yield cls.validate
11
12
    @classmethod
13
    def validate(cls, v):
14
        return PydanticObjectId(v)
15
16
    @classmethod
17
    def __modify_schema__(cls, field_schema: dict):
18
        field_schema.update(
19
            type="string",
20
            examples=["5eb7cf5a86d9755df3a6c593", "5eb7cfb05e32e07750a1756a"],
21
        )
22
23
ENCODERS_BY_TYPE[PydanticObjectId] = str

1
@app.route("/cocktails/", methods=["POST"])
2
def new_cocktail():
3
    raw_cocktail = request.get_json()
4
    raw_cocktail["date_added"] = datetime.utcnow()
5
6
    cocktail = Cocktail(**raw_cocktail)
7
    insert_result = recipes.insert_one(cocktail.to_bson())
8
    cocktail.id = PydanticObjectId(str(insert_result.inserted_id))
9
    print(cocktail)
10
11
    return cocktail.to_json()

1
@app.route("/cocktails/<string:slug>", methods=["GET"])
2
def get_cocktail(slug):
3
    recipe = recipes.find_one_or_404({"slug": slug})
4
    return Cocktail(**recipe).to_json()

1
"_links": {
2
    "last": {
3
        "href": "http://localhost:5000/cocktails/?page=5"
4
    }, 
5
    "next": {
6
        "href": "http://localhost:5000/cocktails/?page=5"
7
    }, 
8
    "prev": {
9
        "href": "http://localhost:5000/cocktails/?page=3"
10
    }, 
11
    "self": {
12
        "href": "http://localhost:5000/cocktails/?page=4"
13
    }
14
}, 

1
@app.route("/cocktails/")
2
def list_cocktails():
3
    """
4
    GET a list of cocktail recipes.
5
6
    The results are paginated using the `page` parameter.
7
    """
8
9
    page = int(request.args.get("page", 1))
10
    per_page = 10  # A const value.
11
12
    # For pagination, it's necessary to sort by name,
13
    # then skip the number of docs that earlier pages would have displayed,
14
    # and then to limit to the fixed page size, ``per_page``.
15
    cursor = recipes.find().sort("name").skip(per_page * (page - 1)).limit(per_page)
16
17
    cocktail_count = recipes.count_documents({})
18
19
    links = {
20
        "self": {"href": url_for(".list_cocktails", page=page, _external=True)},
21
        "last": {
22
            "href": url_for(
23
                ".list_cocktails", page=(cocktail_count // per_page) + 1, _external=True
24
            )
25
        },
26
    }
27
    # Add a 'prev' link if it's not on the first page:
28
    if page > 1:
29
        links["prev"] = {
30
            "href": url_for(".list_cocktails", page=page - 1, _external=True)
31
        }
32
    # Add a 'next' link if it's not on the last page:
33
    if page - 1 < cocktail_count // per_page:
34
        links["next"] = {
35
            "href": url_for(".list_cocktails", page=page + 1, _external=True)
36
        }
37
38
    return {
39
        "recipes": [Cocktail(**doc).to_json() for doc in cursor],
40
        "_links": links,
41
    }

1
@app.errorhandler(404)
2
def resource_not_found(e):
3
    """
4
    An error-handler to ensure that 404 errors are returned as JSON.
5
    """
6
    return jsonify(error=str(e)), 404
7
8
9
@app.errorhandler(DuplicateKeyError)
10
def resource_not_found(e):
11
    """
12
    An error-handler to ensure that MongoDB duplicate key errors are returned as JSON.
13
    """
14
    return jsonify(error=f"Duplicate key error."), 400