Using a Flask blueprint

Before defining the blueprint, we'll define some classes and functions that establish the problem domain. We'll define an enumeration of the Status values, with "Updated" and "Created" as the only two values. We'll define a set of dice using a dataclass named Dice. This class includes the current state of the dice, a unique identifier for this particular collection of dice, and an overall status to show whether it was initially created or has been updated. Separate from the class, it's helpful to have a make_dice() function to create a Dice instance.

For this example, here are the definitions:

from typing import Dict, Any, Tuple, List
from dataclasses import dataclass, asdict
import random
import secrets
from enum import Enum

class Status(str, Enum):
UPDATED = "Updated"
CREATED = "Created"

@dataclass
class Dice:
roll: List[int]
identifier: str
status: str

def reroll(self, keep_positions: List[int]) -> None:
for i in range(len(self.roll)):
if i not in keep_positions:
self.roll[i] = random.randint(1, 6)
self.status = Status.UPDATED

def make_dice(n_dice: int) -> Dice:
return Dice(
roll=[random.randint(1, 6) for _ in range(n_dice)],
identifier=secrets.token_urlsafe(8),
status=Status.CREATED
)

The Dice class represents a handful of six-sided dice. A particular roll of the dice also has an identifier; this is a surrogate key used to identify an initial toss of the dice. Selected dice can be rerolled, which is consistent with a number of dice games.

The status attribute is used as part of a RESTful response to show the current state of the object. The status attribute is either the "Created" or "Updated" string based on the Status class definition.

The make_dice() function creates a new Dice instance. It's defined outside the Dice class definition to emphasize its roll as creating a Dice instance. This could be defined as a method within the Dice class, and decorated with the @classmethod or @staticmethod decorator. 

While a very strict object-oriented design approach mandates that everything be part of a class, Python imposes no such restriction. It seems simpler to have a function outside a class, rather than a decorated function within the class. Both are defined in a common module, so the relationship between function and class is clear.

The reroll() method of a Dice instance updates the object's internal state. This is the important new feature in this section. 

The Open API Specification for this service has a skeleton definition as follows:

OPENAPI_SPEC = {
"openapi": "3.0.0",
"info": {
"title": "Chapter 13. Example 3",
"version": "2019.02",
"description": "Rolls dice",
},
"paths": {
"/rolls": {
"post": {
"description": "first roll",
"responses": {201: {"description": "Created"}},
},
"get": {
"description": "current state",
"responses": {200: {"description": "Current state"}},
},
"patch": {
"description": "subsequent roll",
"responses": {200: {"description": "Updated"}},
}
}
}
}

This specification provides one path, /rolls, which responds to a variety of method requests. A post request has a very terse description of "first roll"; only one of the possible responses is defined in this example specification. Similarly, the get and patch requests have the minimal definition required to pass a simple schema check.

This specification omits the lengthy definition of the parameters required for the post and patch requests. The post request requires a document in JSON Notation. Here's an example: {"dice": 5}. The patch request also requires a document in JSON Notation, the body of this document specifies which dice must be left alone, and which must be re-rolled. The content of the document only specifies which dice to keep; all of the others will be re-rolled. It will look like this: {"keep": [0, 1, 2]}

One of the tools for decomposing a complex Flask application is a blueprint. A blueprint is registered with a Flask application with a specific path prefix. This allows us to have multiple, related portions of a complex application by using several instances of a common blueprint.

For this example, we'll only register a single instance of a Blueprint object. The definition of a Blueprint object starts as follows:

from flask import Flask, jsonify, request, url_for, Blueprint, current_app, abort
from typing import Dict, Any, Tuple, List

SESSIONS: Dict[str, Dice] = {}

rolls = Blueprint("rolls", __name__)

@rolls.route("/openapi.json")
def openapi() -> Dict[str, Any]:
return jsonify(OPENAPI_SPEC)

@rolls.route("/rolls", methods=["POST"])
def make_roll() -> Tuple[Dict[str, Any], HTTPStatus, Dict[str, str]]:
body = request.get_json(force=True)
if set(body.keys()) != {"dice"}:
raise BadRequest(f"Extra fields in {body!r}")
try:
n_dice = int(body["dice"])
except ValueError as ex:
raise BadRequest(f"Bad 'dice' value in {body!r}")

dice = make_dice(n_dice)
SESSIONS[dice.identifier] = dice
current_app.logger.info(f"Rolled roll={dice!r}")

headers = {
"Location":
url_for("rolls.get_roll", identifier=dice.identifier)
}
return jsonify(asdict(dice)), HTTPStatus.CREATED, headers

The SESSIONS object uses an all-caps name to show that it is a global module. This can become our database. Currently, it's initialized as an empty dictionary object. In a more sophisticated application, it could be a shelve object. Pragmatically, this would be replaced with a proper database driver to handle locking and concurrent access.

The Blueprint object, like the Flask object, is assigned a name. In many cases, the module name can be used. For this example, the module name will be used by the overall Flask application, so the name "rolls" was used.

In this example, we've assigned the Blueprint object to a rolls variable. It's also common to see a name such as bp used for this. A shorter name makes it slightly easier to type the route decorator name.

The preceding example defines two routes:

  • One route is for the Open API Specification document.
  • The second route is for the /rolls path, when used with the POST method.

This combination will lead to the typical four-part process of handling a RESTful request:

  •  Parsing: The Flask request object is interrogated to get the body of the request. The request.get_json(force=True) will use a JSON parser on the body of the request, irrespective of Content-Type header. While a well-written client application should provide appropriate headers, this code will tolerate a missing or incorrect header. A few validity checks are applied to the body of the request. In this example, a specialized exception is raised if the request doesn't include the 'dice' key, and if the value for the 'dice' key isn't a valid integer. More checks for valid inputs might include being sure the number of dice was between 1 and 100; this can help to avoid the problem of rolling millions or billions of dice. In the following example, we'll see how this exception is mapped to a proper HTTP status code of 400 Bad Request.
  • Evaluating: The make_dice() function creates a Dice instance. This is saved into the global module SESSIONS database. Because each Dice instance is given a unique, randomized key, the SESSIONS database retains a history of all of the Dice objects. This is a simple in-memory database, and the information only lasts as long as the server is running. A more complete application would use a separate persistence layer.
  • Logging. The Flask current_app object is used to get access to the Flask logger and write a log message on the response.
  • Responding. The Flask jsonify() function is used to serialize the response into JSON Notation. This includes an additional Location header. This additional header is offered to a client to provide the correct, canonical URI for locating the object that was created. We've used the Flask url_for() function to create the proper URL. This function uses the name of the function, qualified by the "rolls.get_roll" blueprint name; the URI path is reverse-engineered from the function and the argument value.

The function for retrieving a roll uses a GET request, and is as follows:

@rolls.route("/rolls/<identifier>", methods=["GET"])
def get_roll(identifier) -> Tuple[Dict[str, Any], HTTPStatus]:
if identifier not in SESSIONS:
abort(HTTPStatus.NOT_FOUND)

return jsonify(asdict(SESSIONS[identifier])), HTTPStatus.OK

This function represents a minimalist approach. If the identifier is unknown, a 404 Not Found message is returned. There's no additional evaluation or logging, since no state change occurred. This function serializes the current state of the dice.

The response to a PATCH request is as follows:

@rolls.route("/rolls/<identifier>", methods=["PATCH"])
def patch_roll(identifier) -> Tuple[Dict[str, Any], HTTPStatus]:
if identifier not in SESSIONS:
abort(HTTPStatus.NOT_FOUND)
body = request.get_json(force=True)
if set(body.keys()) != {"keep"}:
raise BadRequest(f"Extra fields in {body!r}")
try:
keep_positions = [int(d) for d in body["keep"]]
except ValueError as ex:
raise BadRequest(f"Bad 'keep' value in {body!r}")

dice = SESSIONS[identifier]
dice.reroll(keep_positions)

return jsonify(asdict(dice)), HTTPStatus.OK

This has two kinds of validation to perform:

  • The identifier must be valid.
  • The document provided in the body of the request must also be valid.

This example performs a minimal validation to develop a list of positions assigned to the keep_positions variable.

The evaluation is performed by the reroll() method of the Dice instance. This fits the idealized notion of separating the Dice model from the RESTful application that exposes the model. There's relatively little substantial state-change processing in the Flask application. 

Consider a desire for additional validation to confirm that the position values are between zero and the number of dice in SESSIONS[identifier].roll. Is this a part of the Flask application? Or, is this the responsibility of the Dice class definition?

There are two kinds of validations involved here:

  • Serialization syntax: This validation includes a JSON syntax, a data type representation, and the isolated constraints expressed in the Open API Specification. This validation is limited to simple constraints on the values that can be serialized in JSON. We can formalize the {"dice": d} document schema to ensure that k is a positive integer. Similarly, we can formalize the {"keep": [k, k, k, ...]} document schema to require the values for k to be positive integers. However, there's no way to express a relationship between the values for n and k. There's also no way to express the requirement that the values for k be unique. 
  • Problem domain data: This validation goes beyond what can be specified in the Open API specification. It is tied to the problem domain and requires the JSON serialization be valid. This can include actions such as validating relationships among objects, confirming that state changes are permitted, or checking more nuanced rules such as string formats for email addresses.

This leads to multiple tiers of RESTful request parsing and validation. Some features of the request are tied directly to serialization and simple data type questions. Other features of a request are part of the problem domain, and must be deferred to the classes that support the RESTful API.

..................Content has been hidden....................

You can't read the all page of ebook, please click here login for view all page.
Reset