Flask RESTful API validación de parámetros de solicitud con esquemas Marshmallow
Cree esquemas separados para los parámetros de la ruta, la consulta y el cuerpo de la solicitud y valídelos con una sola función.

Cuando se construye un RESTful API lo primero que se hace es definir los códigos de estado y las respuestas de error. El RFC 7807 'Problem Details for HTTP APIs' especifica los parámetros mínimos que debe devolver. Si no has mirado esto, te sugiero que lo hagas. Por supuesto, a menudo querrá incluir más detalles sobre lo que salió mal. APIs son para los desarrolladores y queremos facilitarles la comprensión de por qué falló una llamada.
Cuando construyes un API con Flask es casi imposible ignorar un paquete de serialización / deserialización como Marshmallow. Y junto con el paquete APISpec es fácil documentar su API.
Cuando llamas a tus métodos API lo primero que hay que hacer es validar los parámetros de la petición. A continuación te muestro cómo he hecho esto con Marshmallow.
Validación de entradas con Marshmallow
Escribir un API es diferente a escribir una aplicación web. Para una aplicación web Flask con formularios, utilizamos el paquete WTForms para recuperar y validar los parámetros de la solicitud. Para una API, utilizamos esquemas. Los esquemas describen cómo comunicarse con un API y nos permiten documentar nuestro API con Swagger, por ejemplo.
El paquete Marshmallow no sólo nos permite deserializar los parámetros de entrada, sino que también contiene un amplio conjunto de validadores.
Una clase modelo de ejemplo: Ciudad
En este artículo voy a utilizar la clase modelo City (SQLAlchemy).
class City(Base):
__tablename__ = 'city'
id = Column(Integer, primary_key=True)
name = Column(String(100), server_default='')
Queremos las siguientes operaciones:
- Obtener una lista de ciudades, con paginación y filtro
- Obtener una ciudad por id
- Crear una ciudad
- Actualizar una ciudad por id
- Borrar una ciudad por id
Entonces tenemos las siguientes peticiones:
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
Y nuestros métodos se ven así:
@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 y parámetros de petición
Hay tres tipos de parámetros de petición:
- Parámetros de ruta: city_id
- Parámetros de consulta: page, per_page, search
- Parámetros del cuerpo: nombre
En Flask podemos acceder a estos parámetros utilizando el objeto request:
- request.view_args
- request.args
- request.form
Traducción a esquemas Marshmallow
Comenzamos creando clases base, esquemas, para los tres tipos de parámetros de la petición:
class RequestPathParamsSchema(Schema):
pass
class RequestQueryParamsSchema(Schema):
pass
class RequestBodyParamsSchema(Schema):
pass
Luego creamos esquemas para estos parámetros:
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)
Esto se parece mucho a cuando se utiliza WTForms. Arriba hice referencia a dos clases PaginationQueryParamsSchema y NameFilterQueryParamsSchema. Aquí están:
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)
Para la deserialización y validación de los objetos, Marshmallow nos da el método load(). Este método devuelve un diccionario de nombres de campos mapeados a valores. Si se produce un error de validación, se levanta un ValidationError.
Por ejemplo, para comprobar el parámetro city_id de la ruta Url de la petición en los métodos GET, UPDATE y DELETE, llamamos a load() de la siguiente manera:
try:
result = CitiesRequestPathParamsSchema().load(request.view_args)
except ValidationError as err:
...
Cuando no se encuentran errores, el resultado es el siguiente:
{'city_id': 5}
Podemos hacer lo mismo para los parámetros de consulta de la Url de la petición y los parámetros del cuerpo de la petición.
Una única función que comprueba todos los esquemas
Con lo anterior todavía tenemos que llamar al método load() para todos los esquemas de un método. En nuestro ejemplo sólo el método PUT tiene dos esquemas, pero en un método más realista API se pueden esperar más esquemas. Estaba buscando una manera de combinar esto para evitar la repetición y reducir el código.
Debido a que nuestros esquemas de Ciudades son heredados de nuestras clases base RequestPathParamsSchema, RequestQueryParamsSchema y RequestBodyParamsSchema, podemos usarlos para seleccionar los parámetros a pasar al método load(). Para ello se utiliza la función Python isinstance(). También devuelve True al comprobar si un objeto es heredado de una clase base.
He creado una clase de ayuda APIUtils con un método request_schemas_load() donde puedo pasar uno o más esquemas para ser validados y cargados.
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
En el código anterior, iteramos a través de los esquemas y los cargamos y validamos uno por uno. El resultado es un diccionario de los resultados de Marshmallow schema.load(). La suposición es que no tenemos diferentes parámetros con el mismo nombre dentro de un tipo de petición (path, query, body).
Title es el título del RFC 7807. Este es un mensaje para todo tipo de errores de parámetros de petición.
En caso de error, se lanza una excepción. Los datos de los mensajes ValidationError también se incluyen en la respuesta
e incluyen detalles del error:
- err.messages
- err.datos
- err.datos_validos
También añado el método get_by_id_or_404() para mostrar la respuesta en caso de que no se encuentre un recurso.
Código del método de actualización de la ciudad
Con el método de ayuda APIUtils request_schemas_load() ya podemos escribir el código del método de actualización de la Ciudad:
@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
En lo anterior utilizo el método APIUtils.get_by_id_or_404() para obtener una única Ciudad por id. Si la ciudad no existe, se produce un APIError. Cuando no hay errores y se encuentra la ciudad, actualizo los parámetros del cuerpo con los valores de la petición.
Algunas respuestas de error API
Utilizo Curl para mostrar algunos resultados. Ya he cargado la base de datos con algunas ciudades.
Primero obtenemos dos ciudades:
curl -i "http://127.0.0.1:5000/api/v1/cities?page=1&per_page=2"
La respuesta:
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
}
}
Ahora vamos a actualizar el nombre de la ciudad Berlín a Hamburgo, pero con errores.
Ejemplo 1: parámetro de ruta malo, city_id = 0
curl -i -X PUT -H "Content-Type: application/json" -d '{"name":"Hamburg"}' http://127.0.0.1:5000/api/cities/0
La respuesta:
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
}
}
Ejemplo 2: parámetro de cuerpo malo, nombre = H
curl -i -X PUT -H "Content-Type: application/json" -d '{"name":"H"}' http://127.0.0.1:5000/api/v1/cities/2
La respuesta:
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"
}
}
Ejemplo 3: añadir parámetro desconocido al cuerpo: algo = nada
curl -i -X PUT -H "Content-Type: application/json" -d '{"name":"Hamburg", "something": "nothing"}' http://127.0.0.1:5000/api/v1/cities/2
La respuesta:
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"
}
}
Ejemplo 4: seleccionar ciudad desconocida, para mostrar 404
curl -i -X PUT -H "Content-Type: application/json" -d '{"name":"Hamburg"}' http://127.0.0.1:5000/api/v1/cities/20
La respuesta:
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
}
}
Uso del preprocesamiento y postprocesamiento del esquema
Puede encontrarse con problemas al utilizar valores por defecto (opcionales). Afortunadamente, Marshmallow es lo suficientemente flexible como para permitirle hacer su propio procesamiento. Puedes extender tu esquema con los métodos de preprocesamiento y postprocesamiento pre_load() y post_load().
Por ejemplo, extendí el PaginationQueryParamsSchema para manejar adecuadamente los valores predeterminados de paginación añadiendo el método pre_load():
class PaginationQueryParamsSchema(RequestQueryParamsSchema):
def pre_load_(self, data, many, **kwargs):
# your processing here
...
return data
page = fields.Int(
...
per_page = fields.Int(
...
Resumen
Marshmallow no sólo facilita el uso de esquemas al volcar objetos, sino que también incluye un amplio conjunto de funciones de validación de esquemas que podemos utilizar para comprobar los parámetros de las peticiones. En este post he mostrado una forma de procesar los parámetros de petición de un API RESTful. Por supuesto, hay otras maneras. Usando el objeto de petición Flask podemos pasar request.view_args para los parámetros de la ruta Url, request.args para los parámetros de la consulta Url y request.form para los parámetros del cuerpo, al método schema.load().
Enlaces / créditos
marshmallow: simplified object serialization
https://marshmallow.readthedocs.io/en/stable/
Problem Details for HTTP APIs
https://tools.ietf.org/html/rfc7807
Deje un comentario
Comente de forma anónima o inicie sesión para comentar.
Comentarios (1)
Deje una respuesta.
Responda de forma anónima o inicie sesión para responder.

yo please change your font colors. its horrible to read the text/code.
Recientes
- Recoger y bloquear IP addresses con ipset y Python
- Cómo cancelar tareas con Python Asynchronous IO (AsyncIO)
- Ejecutar un comando Docker dentro de un contenedor Cron Docker
- Creación de un Captcha con Flask, WTForms, SQLAlchemy, SQLite
- Multiprocessing, bloqueo de archivos, SQLite y pruebas
- Envío de mensajes a Slack mediante chat_postMessage
Más vistos
- Flask RESTful API validación de parámetros de solicitud con esquemas Marshmallow
- Usando UUIDs en lugar de Integer Autoincrement Primary Keys con SQLAlchemy y MariaDb
- Usando Python's pyOpenSSL para verificar los certificados SSL descargados de un host
- Conectarse a un servicio en un host Docker desde un contenedor Docker
- Usando PyInstaller y Cython para crear un ejecutable de Python
- SQLAlchemy: Uso de Cascade Deletes para eliminar objetos relacionados