Flask RESTful API Validierung von Anfrageparametern mit Marshmallow-Schemas

Wenn Sie eine RESTful API erstellen, definieren Sie als erstes die Statuscodes und Fehlerantworten. RFC 7807 'Problem Details for HTTP APIs' gibt die Mindestparameter an, die Sie zurückgeben sollten. Wenn Sie sich damit noch nicht beschäftigt haben, sollten Sie das tun. Natürlich wollen Sie oft mehr Details darüber angeben, was schief gelaufen ist. APIs sind für Entwickler gedacht und wir wollen es ihnen leicht machen, zu verstehen, warum ein Aufruf fehlgeschlagen ist.

Wenn Sie eine API mit Flask bauen, ist es fast unmöglich, ein Serialisierungs- / Deserialisierungspaket wie Marshmallow zu ignorieren. Und zusammen mit dem Paket APISpec ist es einfach, Ihre API zu dokumentieren.

Wenn Sie Ihre API -Methoden aufrufen, ist das erste, was Sie tun müssen, die Validierung der Anfrageparameter. Im Folgenden zeige ich Ihnen, wie ich das mit Marshmallow gemacht habe.

Eingabevalidierung mit Marshmallow

Das Schreiben einer API ist anders als das Schreiben einer Webanwendung. Für eine Flask -Webanwendung mit Formularen verwenden wir das Paket WTForms , um die Anfrageparameter abzurufen und zu validieren. Für eine API verwenden wir Schemas. Schemas beschreiben, wie mit einem API zu kommunizieren ist und erlauben uns, unseren API z. B. mit Swagger zu dokumentieren.

Das Marshmallow-Paket ermöglicht uns nicht nur die Deserialisierung von Eingabeparametern, sondern enthält auch einen großen Satz von Validatoren.

Ein Beispiel für eine Modellklasse: City

In diesem Artikel werde ich die (SQLAlchemy) Modellklasse City verwenden.

class City(Base):
    __tablename__ = 'city'

    id = Column(Integer, primary_key=True)
    name = Column(String(100), server_default='')

Wir wollen die folgenden Operationen:

• Abrufen einer Liste von Städten, mit Paginierung und Filter
• Abrufen einer Stadt nach ID
• Erstellen einer Stadt
• Aktualisieren einer Stadt nach ID
• Löschen einer Stadt nach ID

Dann haben wir die folgenden Anfragen:

GET       /cities?page=1&per_page=10&search=an
GET       /cities/4
POST      /cities, city name in request body
PUT       /cities/5, city name in request body
DELETE   /cities/7

Und unsere Methoden sehen wie folgt aus:

@blueprint_cities.route('', methods=['GET'])
def cities_list():
    ...

@blueprint_cities.route('/<int:city_id>', methods=['GET'])
def cities_get(city_id):
    ...

@blueprint_cities.route('', methods=['POST'])
def cities_create():
    ...

@blueprint_cities.route('/<int:city_id>', methods=['PUT'])
def cities_update(city_id):
    ...

@blueprint_cities.route('/<int:city_id>', methods=['DELETE'])
def cities_delete(city_id):
    ...

Flask und Abfrageparameter

Es gibt drei Arten von Abfrageparametern:

• Pfadparameter: city_id
• Abfrageparameter: page, per_page, search
• Körper-Parameter: name

In Flask können wir auf diese Parameter über das Request-Objekt zugreifen:

• request.view_args
• request.args
• request.form

Übersetzung in Marshmallow-Schemata

Wir beginnen mit der Erstellung von Basisklassen, Schemas, für die drei Arten von Request-Parametern:

class RequestPathParamsSchema(Schema):
    pass

class RequestQueryParamsSchema(Schema):
    pass

class RequestBodyParamsSchema(Schema):
    pass

Dann erstellen wir Schemas für diese Parameter:

class CitiesRequestQueryParamsSchema(PaginationQueryParamsSchema, NameFilterQueryParamsSchema):
    pass
    
class CitiesRequestPathParamsSchema(RequestPathParamsSchema):
    city_id = fields.Int(
        description='The id of the city.',
        validate=validate.Range(min=1, max=9999),
        required=True)

class CitiesCreateRequestBodyParamsSchema(RequestBodyParamsSchema):
    name = fields.Str(
        description='The name of the city',
        validate=validate.Length(min=2, max=40),
        required=True)

class CitiesUpdateRequestBodyParamsSchema(RequestBodyParamsSchema):
    name = fields.Str(
        description='The name of the city',
        validate=validate.Length(min=2, max=40),
        required=True)

Das sieht sehr ähnlich aus wie bei der Verwendung von WTForms. Oben habe ich auf die beiden Klassen PaginationQueryParamsSchema und NameFilterQueryParamsSchema verwiesen. Hier sind sie:

class PaginationQueryParamsSchema(RequestQueryParamsSchema):
    page = fields.Int(
        missing=PAGINATION_PAGE_VALUE_DEFAULT,
        description='Pagination page number, first page is 1.',
        validate=validate.Range(min=PAGINATION_PAGE_VALUE_MIN, max=PAGINATION_PAGE_VALUE_MAX),
        required=False)
    per_page = fields.Int(
        missing=PAGINATION_PER_PAGE_VALUE_DEFAULT,
        description='Pagination items per page.',
        validate=validate.Range(min=PAGINATION_PER_PAGE_VALUE_MIN, max=PAGINATION_PER_PAGE_VALUE_MAX),
        required=False)

class NameFilterQueryParamsSchema(RequestQueryParamsSchema):
    name = fields.Str(
        description='The (part of the) name to search for',
        validate=validate.Length(min=2, max=40),
        required=False)

Für die Deserialisierung und Validierung von Objekten gibt uns Marshmallow die Methode load(). Diese Methode gibt ein Wörterbuch mit Feldnamen zurück, die auf Werte abgebildet sind. Wenn ein Validierungsfehler auftritt, wird ein ValidationError ausgelöst.

Um zum Beispiel den Request Url-Pfad-Parameter city_id in den Methoden GET, UPDATE und DELETE zu überprüfen, rufen wir load() wie folgt auf:

try:
    result = CitiesRequestPathParamsSchema().load(request.view_args)
except ValidationError as err:
    ...

Wenn keine Fehler gefunden werden, sieht das Ergebnis so aus:

{'city_id': 5}

Wir können das Gleiche für die Parameter der Anfrage-Url und des Anfrage-Bodys tun.

Eine einzige Funktion, die alle Schemas überprüft

Mit dem oben Gesagten müssen wir noch die load()-Methode für alle Schemas in einer Methode aufrufen. In unserem Beispiel hat nur die Methode request PUT zwei Schemas, aber in einem realistischeren API kann man mehr Schemas erwarten. Ich habe nach einer Möglichkeit gesucht, diese zu kombinieren, um Wiederholungen zu vermeiden und den Code zu reduzieren.

Da unsere Cities-Schemas von unseren Basisklassen RequestPathParamsSchema, RequestQueryParamsSchema und RequestBodyParamsSchema geerbt werden, können wir diese verwenden, um die Parameter auszuwählen, die an die load()-Methode übergeben werden. Hierfür wird die Funktion Python isinstance() verwendet. Sie gibt auch True zurück, wenn geprüft wird, ob ein Objekt von einer Basisklasse geerbt wurde.

Ich habe eine Hilfsklasse APIUtils mit einer Methode request_schemas_load() erstellt, in der ich ein oder mehrere Schemas übergeben kann, die validiert und geladen werden sollen.

class  APIUtils:
    ...

    @classmethod
    def schema_check(cls, schema=None, json_data=None, title=None):
        # load and validate
        try:
            return schema.load(data=json_data, partial=True)

        except ValidationError as err:
            raise  APIError(
                status_code=400,
                title=title,
                messages=err.messages,
                data=err.data,
                valid_data=err.valid_data,
            )

    @classmethod
    def request_schemas_load(cls, schemas):
        if not isinstance(schemas, list):
            schemas = [schemas]

        result_path, result_query, result_body = {}, {}, {}
        for schema in schemas:
            if isinstance(schema, api_spec.RequestPathParamsSchema):
                # path params
                result_path.update(cls.schema_check(
                    schema=schema,
                    json_data=request.view_args,
                    title='One or more request url path parameters did not validate'))

            if isinstance(schema, api_spec.RequestQueryParamsSchema):
                # query params
                result_query.update(cls.schema_check(
                    schema=schema, 
                    json_data=request.args,
                    title='One or more request url query parameters did not validate'))

            if isinstance(schema, api_spec.RequestBodyParamsSchema):
                # body params
                result_body.update(cls.schema_check(
                    schema=schema, 
                    json_data=request.get_json(),
                    title='One or more request body parameters did not validate'))

        return {
            'path': result_path,
            'query': result_query,
            'body': result_body,
        }

    @classmethod
    def get_by_id_or_404(cls, res, res_id, res_name, res_id_name):
        obj = app_db.session.query(res).get(res_id)
        if obj is  None:
            raise  APIError(
                status_code=404,
                title='The requested resource could not be found',
                messages={
                    res_name: [
                        'Not found',
                    ]
                },
                data={
                    res_id_name: res_id,
                },
            )
        return obj

Im obigen Code iterieren wir durch die Schemas und laden und validieren sie eins nach dem anderen. Das Ergebnis ist ein Wörterbuch der Marshmallow schema.load() Ergebniswörterbücher. Die Annahme ist, dass wir innerhalb eines Anfragetyps (path, query, body) keine unterschiedlichen Parameter mit demselben Namen haben.

Titel ist der RFC 7807-Titel. Dies ist eine Meldung für alle Arten von Anfrageparameter-Fehlern.

Im Falle eines Fehlers wird eine Ausnahme ausgelöst. Die Daten der ValidationError-Meldungen sind auch in der Antwort
enthalten und enthalten Details zum Fehler:

• err.messages
• err.data
• err.valid_data

Ich füge auch die Methode get_by_id_or_404() hinzu, um die Antwort anzuzeigen, falls eine Ressource nicht gefunden wurde.

Code für die Methode City update

Mit APIUtils Helper-Methode request_schemas_load() können wir nun den Code für die City-Update-Methode schreiben:

@blueprint_cities.route('/<int:city_id>', methods=['PUT'])
def cities_update(city_id):
    result =  APIUtils.request_schemas_load([
        CitiesRequestPathParamsSchema(),
        CitiesUpdateRequestBodyParamsSchema()])

    city =  APIUtils.get_by_id_or_404(City, city_id, 'City', 'city_id')

    for k, v in result['body'].items():
        setattr(city, k, v)
    app_db.session.commit()

    return jsonify({
        'data': CitiesResponseSchema().dump(city)
    }), 200

Im obigen Beispiel verwende ich die Methode APIUtils.get_by_id_or_404(), um eine einzelne Stadt anhand ihrer ID zu ermitteln. Wenn die Stadt nicht existiert, wird ein APIError ausgelöst. Wenn es keine Fehler gibt und die Stadt gefunden wird, aktualisiere ich die Body-Parameter mit den Werten in der Anfrage.

Einige API -Fehlerantworten

Ich verwende Curl , um einige Ergebnisse zu zeigen. Ich habe die Datenbank bereits mit einigen Städten geladen.

Zuerst erhalten wir zwei Städte:

curl -i "http://127.0.0.1:5000/api/v1/cities?page=1&per_page=2"

Die Antwort:

HTTP/1.0 200 OK
Content-Type:  application/json
Content-Length: 247
Server: Werkzeug/1.0.1  Python/3.8.5
Date: Mon, 29 Mar 2021 15:51:39 GMT

{
  "data": [
    {
      "id": 1, 
      "name": "Beijing"
    }, 
    {
      "id": 2, 
      "name": "Berlin"
    }
  ], 
  "meta": {
    "count": 2, 
    "limit": 2, 
    "offset": 0, 
    "page": 1, 
    "per_page": 2, 
    "total": 11
  }
}

Nun wollen wir den Städtenamen Berlin auf Hamburg aktualisieren, allerdings mit Fehlern.

Beispiel 1: falscher Pfadparameter, city_id = 0

curl -i -X PUT  -H "Content-Type:  application/json" -d '{"name":"Hamburg"}'  http://127.0.0.1:5000/api/cities/0

Die Antwort:

HTTP/1.0 400 BAD REQUEST
Content-Type:  application/json
Content-Length: 247
Server: Werkzeug/1.0.1  Python/3.8.5
Date: Mon, 29 Mar 2021 16:13:22 GMT

{
  "status": 400, 
  "title": "One or more request url path parameters did not validate", 
  "messages": {
    "city_id": [
      "Must be greater than or equal to 1 and less than or equal to 9999."
    ]
  }, 
  "data": {
    "city_id": 0
  }
}

Beispiel 2: fehlerhafter Body-Parameter, name = H

curl -i -X PUT  -H "Content-Type:  application/json" -d '{"name":"H"}'  http://127.0.0.1:5000/api/v1/cities/2

Die Antwort:

HTTP/1.0 400 BAD REQUEST
Content-Type:  application/json
Content-Length: 205
Server: Werkzeug/1.0.1  Python/3.8.5
Date: Mon, 29 Mar 2021 16:15:46 GMT

{
  "status": 400, 
  "title": "One or more request body parameters did not validate", 
  "messages": {
    "name": [
      "Length must be between 2 and 40."
    ]
  }, 
  "data": {
    "name": "H"
  }
}

Beispiel 3: Unbekannten Parameter zum body hinzufügen: something = nothing

curl -i -X PUT  -H "Content-Type:  application/json" -d '{"name":"Hamburg", "something": "nothing"}'  http://127.0.0.1:5000/api/v1/cities/2

Die Antwort:

HTTP/1.0 400 BAD REQUEST
Content-Type:  application/json
Content-Length: 273
Server: Werkzeug/1.0.1  Python/3.8.5
Date: Mon, 29 Mar 2021 16:21:40 GMT

{
  "status": 400,
  "title": "One or more request body parameters did not validate", 
  "messages": {
    "something": [
      "Unknown field."
    ]
  }, 
  "data": {
    "name": "Hamburg", 
    "something": "nothing"
  },
  "valid_data": {
    "name": "Hamburg"
  }
}

Beispiel 4: unbekannte Stadt auswählen, um 404 anzuzeigen

curl -i -X PUT  -H "Content-Type:  application/json" -d '{"name":"Hamburg"}'  http://127.0.0.1:5000/api/v1/cities/20

Die Antwort:

HTTP/1.0 404 NOT FOUND
Content-Type:  application/json
Content-Length: 173
Server: Werkzeug/1.0.1  Python/3.8.5
Date: Tue, 30 Mar 2021 16:26:38 GMT

{
  "status": 404, 
  "title": "The requested resource could not be found", 
  "messages": {
    "City": [
      "Not found"
    ]
  }, 
  "data": {
    "city_id": 20
  }
}

Schema-Vor- und Nachbearbeitung verwenden

Sie können auf Probleme stoßen, wenn Sie Standardwerte (optionale Werte) verwenden. Glücklicherweise ist Marshmallow flexibel genug, dass Sie Ihre eigene Verarbeitung durchführen können. Sie können Ihr Schema mit den Vorverarbeitungs- und Nachverarbeitungsmethoden pre_load() und post_load() erweitern.

Zum Beispiel habe ich das PaginationQueryParamsSchema erweitert, um die Paginierungsvorgaben richtig zu behandeln, indem ich die pre_load() Methode hinzugefügt habe:

class PaginationQueryParamsSchema(RequestQueryParamsSchema):
    def pre_load_(self, data, many, **kwargs):
		# your processing here
		...
        return data

    page = fields.Int(
		...

    per_page = fields.Int(
		...

Zusammenfassung

Marshmallow macht es nicht nur einfach, Schemas beim Dumpen von Objekten zu verwenden, sondern enthält auch einen umfangreichen Satz von Schema-Validierungsfunktionen, die wir zur Überprüfung von Anfrageparametern verwenden können. In diesem Beitrag habe ich eine Möglichkeit gezeigt, die Request-Parameter eines RESTful API zu verarbeiten. Natürlich gibt es auch andere Wege. Mit dem Flask -Request-Objekt können wir request.view_args für die Url-Pfadparameter, request.args für die Url-Abfrageparameter und request.form für die Body-Parameter an die Methode schema.load() übergeben.

Links / Impressum

marshmallow: simplified object serialization
https://marshmallow.readthedocs.io/en/stable/

Problem Details for HTTP APIs
https://tools.ietf.org/html/rfc7807