Tutorial (ASGI)¶

In this tutorial we’ll walk through building an API for a simple image sharing service. Along the way, we’ll discuss the basic anatomy of an asynchronous Falcon application: responders, routing, middleware, executing synchronous functions in an executor, and more!

Note

This tutorial covers the asynchronous flavor of Falcon using the ASGI protocol.

Synchronous (WSGI) Falcon application development is covered in our WSGI tutorial.

New Falcon users may also want to choose the WSGI flavor to familiarize themselves with Falcon’s basic concepts.

First Steps¶

Let’s start by creating a fresh environment and the corresponding project directory structure, along the lines of First Steps from the WSGI tutorial:

asgilook
├── .venv
└── asgilook
    ├── __init__.py
    └── app.py

We’ll create a virtualenv using the venv module from the standard library (Falcon requires Python 3.6+ for ASGI):

$ mkdir asgilook
$ python3 -m venv asgilook/.venv
$ source asgilook/.venv/bin/activate

Note

If your Python distribution does not happen to include the venv module, you can always install and use virtualenv instead.

Tip

Some of us find it convenient to manage virtualenvs with virtualenvwrapper or pipenv, particularly when it comes to hopping between several environments.

Next, install Falcon into your virtualenv. ASGI support requires version 3.0 or higher:

$ pip install "falcon>=3.*"

You can then create a basic Falcon ASGI application by adding an asgilook/app.py module with the following contents:

import falcon.asgi

app = falcon.asgi.App()

As in the WSGI tutorial’s introduction, let’s not forget to mark asgilook as a Python package:

$ touch asgilook/__init__.py

Hosting Our App¶

For running our async application, we’ll need an ASGI application server. Popular choices include:

For a simple tutorial application like ours, any of the above should do. Let’s pick the popular uvicorn for now:

$ pip install uvicorn

Configuration¶

Next, let’s make our app configurable by allowing the user to modify the file system path where images are stored. We’ll also allow the UUID generator to be customized.

As Falcon does not prescribe a specific configuration library or strategy, we are free to choose our own adventure (see also a related question in our FAQ: What is the recommended approach for app configuration?).

In this tutorial, we’ll just pass around a Config instance to resource initializers for easier testing (coming later in this tutorial). Create a new module, config.py next to app.py, and add the following code to it:

import os
import pathlib
import uuid


class Config:
    DEFAULT_CONFIG_PATH = '/tmp/asgilook'
    DEFAULT_UUID_GENERATOR = uuid.uuid4

    def __init__(self):
        self.storage_path = pathlib.Path(
            os.environ.get('ASGI_LOOK_STORAGE_PATH', self.DEFAULT_CONFIG_PATH))
        self.storage_path.mkdir(parents=True, exist_ok=True)

        self.uuid_generator = Config.DEFAULT_UUID_GENERATOR

Image Store¶

Since we are going to read and write image files, care must be taken to avoid blocking the app during I/O. We’ll give aiofiles a try:

pip install aiofiles

In addition, let’s twist the original WSGI “Look” design a bit, and convert all uploaded images to JPEG with the popular Pillow library:

pip install Pillow

We can now implement a basic async image store. Save the following code as store.py next to app.py and config.py:

import asyncio
import datetime
import io

import aiofiles
import falcon
import PIL.Image


class Image:

    def __init__(self, config, image_id, size):
        self._config = config

        self.image_id = image_id
        self.size = size
        self.modified = datetime.datetime.utcnow()

    @property
    def path(self):
        return self._config.storage_path / self.image_id

    @property
    def uri(self):
        return f'/images/{self.image_id}.jpeg'

    def serialize(self):
        return {
            'id': self.image_id,
            'image': self.uri,
            'modified': falcon.dt_to_http(self.modified),
            'size': self.size,
        }


class Store:

    def __init__(self, config):
        self._config = config
        self._images = {}

    def _load_from_bytes(self, data):
        return PIL.Image.open(io.BytesIO(data))

    def _convert(self, image):
        rgb_image = image.convert('RGB')

        converted = io.BytesIO()
        rgb_image.save(converted, 'JPEG')
        return converted.getvalue()

    def get(self, image_id):
        return self._images.get(image_id)

    def list_images(self):
        return sorted(self._images.values(), key=lambda item: item.modified)

    async def save(self, image_id, data):
        loop = asyncio.get_running_loop()
        image = await loop.run_in_executor(None, self._load_from_bytes, data)
        converted = await loop.run_in_executor(None, self._convert, image)

        path = self._config.storage_path / image_id
        async with aiofiles.open(path, 'wb') as output:
            await output.write(converted)

        stored = Image(self._config, image_id, image.size)
        self._images[image_id] = stored
        return stored

Here we store data using aiofiles, and run Pillow image transformation functions in the default ThreadPoolExecutor, hoping that at least some of these image operations release the GIL during processing.

Note

The ProcessPoolExecutor is another alternative for long running tasks that do not release the GIL, such as CPU-bound pure Python code. Note, however, that ProcessPoolExecutor builds upon the multiprocessing module, and thus inherits its caveats: higher synchronization overhead, and the requirement for the task and its arguments to be picklable (which also implies that the task must be reachable from the global namespace, i.e., an anonymous lambda simply won’t work).

Images Resource(s)¶

In the ASGI flavor of Falcon, all responder methods, hooks and middleware methods must be awaitable coroutines. Let’s see how this works by implementing a resource to represent both a single image and a collection of images. Place the code below in a file named images.py:

import aiofiles
import falcon


class Images:

    def __init__(self, config, store):
        self._config = config
        self._store = store

    async def on_get(self, req, resp):
        resp.media = [image.serialize() for image in self._store.list_images()]

    async def on_get_image(self, req, resp, image_id):
        # NOTE: image_id: UUID is converted back to a string identifier.
        image = self._store.get(str(image_id))
        resp.stream = await aiofiles.open(image.path, 'rb')
        resp.content_type = falcon.MEDIA_JPEG

    async def on_post(self, req, resp):
        data = await req.stream.read()
        image_id = str(self._config.uuid_generator())
        image = await self._store.save(image_id, data)

        resp.location = image.uri
        resp.media = image.serialize()
        resp.status = falcon.HTTP_201

This module is an example of a Falcon “resource” class, as described in Routing. Falcon uses resource-based routing to encourage a RESTful architectural style. For each HTTP method that a resource supports, the target resource class simply implements a corresponding Python method with a name that starts with on_ and ends in the lowercased HTTP method name (e.g., on_get(), on_patch(), on_delete(), etc.)

Note

If a Python method is omitted for a given HTTP verb, the framework will automatically respond with 405 Method Not Allowed. Falcon also provides a default responder for OPTIONS requests that takes into account which methods are implemented for the target resource.

Here we opted to implement support for both a single image (which supports GET for downloading the image) and a collection of images (which supports GET for listing the collection, and POST for uploading a new image) in the same Falcon resource class. In order to make this work, the Falcon router needs a way to determine which methods to call for the collection vs. the item. This is done by using a suffixed route, as described in add_route() (see also: How do I implement both POSTing and GETing items for the same resource?).

Alternatively, we could have split the implementation to strictly represent one RESTful resource per class. In that case, there would have been no need to use suffixed responders. Depending on the application, using two classes instead of one may lead to a cleaner design. (See also: What is the recommended way to map related routes to resource classes?)

Note

In this example, we serve the image by simply assigning an open aiofiles file to resp.stream. This works because Falcon includes special handling for streaming async file-like objects.

Warning

In production deployment, serving files directly from the web server, rather than through the Falcon ASGI app, will likely be more efficient, and therefore should be preferred. See also: Can Falcon serve static files?

Also worth noting is that the on_get_image() responder will be receiving an image_id of type UUID. So what’s going on here? How will the image_id field, matched from a string path segment, become a UUID?

Falcon’s default router supports simple validation and transformation using field converters. In this example, we’ll use the UUIDConverter to validate the image_id input as UUID. Converters are specified for a route by including their shorthand identifiers in the URI template for the route; for instance, the route corresponding to on_get_image() will use the following template (see also the next chapter, as well as Routing):

/images/{image_id:uuid}.jpeg

Since our application is still internally centered on string identifiers, feel free to experiment with refactoring the image Store to use UUIDs natively!

(Alternatively, one could implement a custom field converter to use uuid only for validation, but return an unmodified string.)

Note

In contrast to asynchronous building blocks (responders, middleware, hooks etc.) of a Falcon ASGI application, field converters are simple synchronous data transformation functions that are not expected to perform any I/O.

Running Our Application¶

Now we’re ready to configure the routes for our app to map image paths in the request URL to an instance of our resource class.

Let’s also refactor our app.py module to let us invoke create_app() wherever we need it. This will become useful later on when we start writing test cases.

Modify app.py to read as follows:

import falcon.asgi

from .config import Config
from .images import Images
from .store import Store


def create_app(config=None):
    config = config or Config()
    store = Store(config)
    images = Images(config, store)

    app = falcon.asgi.App()
    app.add_route('/images', images)
    app.add_route('/images/{image_id:uuid}.jpeg', images, suffix='image')

    return app

As mentioned earlier, we need to use a route suffix for the Images class to distinguish between a GET for a single image vs. the entire collection of images.

Here, we map the '/images/{image_id:uuid}.jpeg' URI template to a single image resource. By specifying an 'image' suffix, we cause the framework to look for responder methods that have names ending in '_image' (e.g, on_get_image()).

We also specify the uuid field converter as discussed in the previous section.

In order to bootstrap an ASGI app instance for uvicorn to reference, we’ll create a simple asgi.py module with the following contents:

from .app import create_app

app = create_app()

Running the application is not too dissimilar from the previous command line:

$ uvicorn asgilook.asgi:app

Provided uvicorn is started as per the above command line, let’s try uploading some images in a separate terminal:

$ http POST localhost:8000/images @/home/user/Pictures/test.png

HTTP/1.1 201 Created
content-length: 173
content-type: application/json
date: Tue, 24 Dec 2019 17:32:18 GMT
location: /images/5cfd9fb6-259a-4c72-b8b0-5f4c35edcd3c.jpeg
server: uvicorn

{
    "id": "5cfd9fb6-259a-4c72-b8b0-5f4c35edcd3c",
    "image": "/images/5cfd9fb6-259a-4c72-b8b0-5f4c35edcd3c.jpeg",
    "modified": "Tue, 24 Dec 2019 17:32:19 GMT",
    "size": [
        462,
        462
    ]
}

Next, try retrieving the uploaded image:

$ http localhost:8000/images/5cfd9fb6-259a-4c72-b8b0-5f4c35edcd3c.jpeg

HTTP/1.1 200 OK
content-type: image/jpeg
date: Tue, 24 Dec 2019 17:34:53 GMT
server: uvicorn
transfer-encoding: chunked

+-----------------------------------------+
| NOTE: binary data not shown in terminal |
+-----------------------------------------+

We could also open the link in a web browser or pipe it to an image viewer to verify that the image was successfully converted to a JPEG.

Let’s check the image collection now:

$ http localhost:8000/images

HTTP/1.1 200 OK
content-length: 175
content-type: application/json
date: Tue, 24 Dec 2019 17:36:31 GMT
server: uvicorn

[
    {
        "id": "5cfd9fb6-259a-4c72-b8b0-5f4c35edcd3c",
        "image": "/images/5cfd9fb6-259a-4c72-b8b0-5f4c35edcd3c.jpeg",
        "modified": "Tue, 24 Dec 2019 17:32:19 GMT",
        "size": [
            462,
            462
        ]
    }
]

The application file layout should now look like this:

asgilook
├── .venv
└── asgilook
    ├── __init__.py
    ├── app.py
    ├── asgi.py
    ├── config.py
    ├── images.py
    └── store.py

Dynamic Thumbnails¶

Let’s pretend our image service customers want to render images in multiple resolutions, for instance, as srcset for responsive HTML images or other purposes.

Let’s add a new method Store.make_thumbnail() to perform scaling on the fly:

async def make_thumbnail(self, image, size):
    async with aiofiles.open(image.path, 'rb') as img_file:
        data = await img_file.read()

    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, self._resize, data, size)

We’ll also add an internal helper to run the Pillow thumbnail operation that is offloaded to a threadpool executor, again, in hoping that Pillow can release the GIL for some operations:

def _resize(self, data, size):
    image = PIL.Image.open(io.BytesIO(data))
    image.thumbnail(size)

    resized = io.BytesIO()
    image.save(resized, 'JPEG')
    return resized.getvalue()

The store.Image class can be extended to also return URIs to thumbnails:

def thumbnails(self):
    def reductions(size, min_size):
        width, height = size
        factor = 2
        while width // factor >= min_size and height // factor >= min_size:
            yield (width // factor, height // factor)
            factor *= 2

    return [
        f'/thumbnails/{self.image_id}/{width}x{height}.jpeg'
        for width, height in reductions(
            self.size, self._config.min_thumb_size)]

Here, we only generate URIs for a series of downsized resolutions. The actual scaling will happen on the fly upon requesting these resources.

Each thumbnail in the series is approximately half the size (one quarter area-wise) of the previous one, similar to how mipmapping works in computer graphics. You may wish to experiment with this resolution distribution.

Furthermore, it is practical to impose a minimum resolution, as any potential benefit from switching between very small thumbnails (a few kilobytes each) is likely to be overshadowed by the request overhead. As you may have noticed in the above snippet, we are referencing this lower size limit as self._config.min_thumb_size. The app configuration will need to be updated to add the min_thumb_size option (by default initialized to 64 pixels) as follows:

import os
import pathlib
import uuid


class Config:
    DEFAULT_CONFIG_PATH = '/tmp/asgilook'
    DEFAULT_MIN_THUMB_SIZE = 64
    DEFAULT_UUID_GENERATOR = uuid.uuid4

    def __init__(self):
        self.storage_path = pathlib.Path(
            os.environ.get('ASGI_LOOK_STORAGE_PATH', self.DEFAULT_CONFIG_PATH))
        self.storage_path.mkdir(parents=True, exist_ok=True)

        self.uuid_generator = Config.DEFAULT_UUID_GENERATOR
        self.min_thumb_size = self.DEFAULT_MIN_THUMB_SIZE

After updating store.py, the module should now look like this:

import asyncio
import datetime
import io

import aiofiles
import falcon
import PIL.Image


class Image:

    def __init__(self, config, image_id, size):
        self._config = config

        self.image_id = image_id
        self.size = size
        self.modified = datetime.datetime.utcnow()

    @property
    def path(self):
        return self._config.storage_path / self.image_id

    @property
    def uri(self):
        return f'/images/{self.image_id}.jpeg'

    def serialize(self):
        return {
            'id': self.image_id,
            'image': self.uri,
            'modified': falcon.dt_to_http(self.modified),
            'size': self.size,
            'thumbnails': self.thumbnails(),
        }

    def thumbnails(self):
        def reductions(size, min_size):
            width, height = size
            factor = 2
            while width // factor >= min_size and height // factor >= min_size:
                yield (width // factor, height // factor)
                factor *= 2

        return [
            f'/thumbnails/{self.image_id}/{width}x{height}.jpeg'
            for width, height in reductions(
                self.size, self._config.min_thumb_size)]


class Store:

    def __init__(self, config):
        self._config = config
        self._images = {}

    def _load_from_bytes(self, data):
        return PIL.Image.open(io.BytesIO(data))

    def _convert(self, image):
        rgb_image = image.convert('RGB')

        converted = io.BytesIO()
        rgb_image.save(converted, 'JPEG')
        return converted.getvalue()

    def _resize(self, data, size):
        image = PIL.Image.open(io.BytesIO(data))
        image.thumbnail(size)

        resized = io.BytesIO()
        image.save(resized, 'JPEG')
        return resized.getvalue()

    def get(self, image_id):
        return self._images.get(image_id)

    def list_images(self):
        return sorted(self._images.values(), key=lambda item: item.modified)

    async def make_thumbnail(self, image, size):
        async with aiofiles.open(image.path, 'rb') as img_file:
            data = await img_file.read()

        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(None, self._resize, data, size)

    async def save(self, image_id, data):
        loop = asyncio.get_running_loop()
        image = await loop.run_in_executor(None, self._load_from_bytes, data)
        converted = await loop.run_in_executor(None, self._convert, image)

        path = self._config.storage_path / image_id
        async with aiofiles.open(path, 'wb') as output:
            await output.write(converted)

        stored = Image(self._config, image_id, image.size)
        self._images[image_id] = stored
        return stored

Let’s also add a Thumbnails resource to expose the new functionality. The final version of images.py reads:

import aiofiles
import falcon


class Images:

    def __init__(self, config, store):
        self._config = config
        self._store = store

    async def on_get(self, req, resp):
        resp.media = [image.serialize() for image in self._store.list_images()]

    async def on_get_image(self, req, resp, image_id):
        # NOTE: image_id: UUID is converted back to a string identifier.
        image = self._store.get(str(image_id))
        if not image:
            raise falcon.HTTPNotFound

        resp.stream = await aiofiles.open(image.path, 'rb')
        resp.content_type = falcon.MEDIA_JPEG

    async def on_post(self, req, resp):
        data = await req.stream.read()
        image_id = str(self._config.uuid_generator())
        image = await self._store.save(image_id, data)

        resp.location = image.uri
        resp.media = image.serialize()
        resp.status = falcon.HTTP_201


class Thumbnails:

    def __init__(self, store):
        self._store = store

    async def on_get(self, req, resp, image_id, width, height):
        image = self._store.get(str(image_id))
        if not image:
            raise falcon.HTTPNotFound
        if req.path not in image.thumbnails():
            raise falcon.HTTPNotFound

        resp.content_type = falcon.MEDIA_JPEG
        resp.data = await self._store.make_thumbnail(image, (width, height))

Note

Even though we are only building a sample application, it is a good idea to cultivate a habit of making your code secure by design and secure by default.

In this case, we see that generating thumbnails on the fly, based on arbitrary dimensions embedded in the URI, could easily be abused to create a denial-of-service attack.

This particular attack is mitigated by validating the input (in this case, the requested path) against a list of allowed values.

Finally, a new thumbnail route needs to be added in app.py. This step is left as an exercise for the reader.

Tip

Draw inspiration from the thumbnail URI formatting string:

f'/thumbnails/{self.image_id}/{width}x{height}.jpeg'

The actual URI template for the thumbnails route should look quite similar to the above.

Remember that we want to use the uuid converter for the image_id field, and image dimensions (width and height) should ideally be converted to ints.

(If you get stuck, see the final version of app.py later in this tutorial.)

Note

If you try to request a non-existent resource (e.g., due to a missing route , or simply a typo in the URI), the framework will automatically render an HTTP 404 Not Found response by raising an instance of HTTPNotFound (unless that exception is intercepted by a custom error handler, or if the path matches a sink prefix).

Conversely, if a route is matched to a resource, but there is no responder for the HTTP method in question, Falcon will render HTTP 405 Method Not Allowed via HTTPMethodNotAllowed.

The new thumbnails end-point should now render thumbnails on the fly:

$ http POST localhost:8000/images @/home/user/Pictures/test.png

HTTP/1.1 201 Created
content-length: 319
content-type: application/json
date: Tue, 24 Dec 2019 18:58:20 GMT
location: /images/f2375273-8049-4b10-b17e-8851db9ac7af.jpeg
server: uvicorn

{
    "id": "f2375273-8049-4b10-b17e-8851db9ac7af",
    "image": "/images/f2375273-8049-4b10-b17e-8851db9ac7af.jpeg",
    "modified": "Tue, 24 Dec 2019 18:58:21 GMT",
    "size": [
        462,
        462
    ],
    "thumbnails": [
        "/thumbnails/f2375273-8049-4b10-b17e-8851db9ac7af/231x231.jpeg",
        "/thumbnails/f2375273-8049-4b10-b17e-8851db9ac7af/115x115.jpeg"
    ]
}


$ http localhost:8000/thumbnails/f2375273-8049-4b10-b17e-8851db9ac7af/115x115.jpeg

HTTP/1.1 200 OK
content-length: 2985
content-type: image/jpeg
date: Tue, 24 Dec 2019 19:00:14 GMT
server: uvicorn

+-----------------------------------------+
| NOTE: binary data not shown in terminal |
+-----------------------------------------+

Again, we could also verify thumbnail URIs in the browser or image viewer that supports HTTP input.

Caching Responses¶

Although scaling thumbnails on-the-fly sounds cool, and we also avoid many pesky small files littering our storage, it consumes CPU resources, and we would soon find our application crumbling under load.

Let’s mitigate this problem with response caching. We’ll use Redis, taking advantage of aioredis for async support:

pip install aioredis

We will also need to serialize response data (the Content-Type header and the body in the first version); msgpack should do:

pip install msgpack

Our application will obviously need access to a Redis server. Apart from just installing Redis server on your machine, one could also:

Spin up Redis in Docker, eg:
```
docker run -p 6379:6379 redis
```
Assuming Redis is installed on the machine, one could also try pifpaf for spinning up Redis just temporarily for uvicorn:
```
pifpaf run redis -- uvicorn asgilook.asgi:app
```

We will perform caching with a Falcon Middleware component. Again, note that all middleware callbacks must be asynchronous. Even initializing the Redis connection with aioredis.create_redis_pool() must be awaited. But how can we await coroutines from within our synchronous create_app() function?

ASGI application lifespan events come to the rescue. An ASGI application server emits these events upon application startup and shutdown.

Let’s implement the process_startup() handler in our middleware to execute code upon our application startup:

async def process_startup(self, scope, event):
    self.redis = await self._config.create_redis_pool(
        self._config.redis_host)

Warning

The Lifespan Protocol is an optional extension; please check if your ASGI server of choice implements it.

uvicorn (that we picked for this tutorial) supports Lifespan.

At minimum, our middleware will need to know the Redis host(s) to use. Let’s also make our Redis connection factory configurable to afford injecting different Redis client implementations for production and testing.

Note

Rather than requiring the caller to pass the host to the connection factory, a wrapper method could be used to implicitly reference self.redis_host. Such a design might prove helpful for apps that need to create client connections in more than one place.

Assuming we call our new configuration items redis_host and create_redis_pool(), respectively, the final version of config.py now reads:

import os
import pathlib
import uuid

import aioredis


class Config:
    DEFAULT_CONFIG_PATH = '/tmp/asgilook'
    DEFAULT_MIN_THUMB_SIZE = 64
    DEFAULT_REDIS_HOST = 'redis://localhost'
    DEFAULT_REDIS_POOL = aioredis.create_redis_pool
    DEFAULT_UUID_GENERATOR = uuid.uuid4

    def __init__(self):
        self.storage_path = pathlib.Path(
            os.environ.get('ASGI_LOOK_STORAGE_PATH', self.DEFAULT_CONFIG_PATH))
        self.storage_path.mkdir(parents=True, exist_ok=True)

        self.create_redis_pool = Config.DEFAULT_REDIS_POOL
        self.min_thumb_size = self.DEFAULT_MIN_THUMB_SIZE
        self.redis_host = self.DEFAULT_REDIS_HOST
        self.uuid_generator = Config.DEFAULT_UUID_GENERATOR

Let’s complete the Redis cache component by implementing two more middleware methods, in addition to process_startup(). Create a cache.py module containing the following code.

import msgpack


class RedisCache:
    PREFIX = 'asgilook:'
    INVALIDATE_ON = frozenset({'DELETE', 'POST', 'PUT'})
    CACHE_HEADER = 'X-ASGILook-Cache'
    TTL = 3600

    def __init__(self, config):
        self._config = config

        # NOTE(vytas): To be initialized upon application startup (see the
        #   method below).
        self._redis = None

    async def _serialize_response(self, resp):
        data = await resp.render_body()
        return msgpack.packb([resp.content_type, data], use_bin_type=True)

    def _deserialize_response(self, resp, data):
        resp.content_type, resp.data = msgpack.unpackb(data, raw=False)
        resp.complete = True
        resp.context.cached = True

    async def process_startup(self, scope, event):
        if self._redis is None:
            self._redis = await self._config.create_redis_pool(
                self._config.redis_host)

    async def process_request(self, req, resp):
        resp.context.cached = False

        if req.method in self.INVALIDATE_ON:
            return

        key = f'{self.PREFIX}/{req.path}'
        data = await self._redis.get(key)
        if data is not None:
            self._deserialize_response(resp, data)
            resp.set_header(self.CACHE_HEADER, 'Hit')
        else:
            resp.set_header(self.CACHE_HEADER, 'Miss')

    async def process_response(self, req, resp, resource, req_succeeded):
        if not req_succeeded:
            return

        key = f'{self.PREFIX}/{req.path}'

        if req.method in self.INVALIDATE_ON:
            await self._redis.delete(key)
        elif not resp.context.cached:
            data = await self._serialize_response(resp)
            await self._redis.set(key, data, expire=self.TTL)

For caching to take effect, we also need to modify app.py to add the RedisCache component to our application’s middleware list. The final version of app.py should look something like this:

import falcon.asgi

from .cache import RedisCache
from .config import Config
from .images import Images, Thumbnails
from .store import Store


def create_app(config=None):
    config = config or Config()
    cache = RedisCache(config)
    store = Store(config)
    images = Images(config, store)
    thumbnails = Thumbnails(store)

    app = falcon.asgi.App(middleware=[cache])
    app.add_route('/images', images)
    app.add_route('/images/{image_id:uuid}.jpeg', images, suffix='image')
    app.add_route('/thumbnails/{image_id:uuid}/{width:int}x{height:int}.jpeg',
                  thumbnails)

    return app

Now, subsequent access to /thumbnails should be cached, as indicated by the x-asgilook-cache header:

$ http localhost:8000/thumbnails/167308e4-e444-4ad9-88b2-c8751a4e37d4/115x115.jpeg

HTTP/1.1 200 OK
content-length: 2985
content-type: image/jpeg
date: Tue, 24 Dec 2019 19:46:51 GMT
server: uvicorn
x-asgilook-cache: Hit

+-----------------------------------------+
| NOTE: binary data not shown in terminal |
+-----------------------------------------+

Note

Left as another exercise for the reader: individual images are streamed directly from aiofiles instances, and caching therefore does not work for them at the moment.

The project’s structure should now look like this:

asgilook
├── .venv
└── asgilook
    ├── __init__.py
    ├── app.py
    ├── asgi.py
    ├── cache.py
    ├── config.py
    ├── images.py
    └── store.py

Testing Our Application¶

So far, so good? We have only tested our application by sending a handful of requests manually. Have we tested all code paths? Have we covered typical user inputs to the application?

Creating a comprehensive test suite is vital not only for verifying that the application is behaving correctly at the moment, but also for limiting the impact of any regressions introduced into the codebase over time.

In order to implement a test suite, we’ll need to revise our dependencies and decide which abstraction level we are after:

Will we run a real Redis server?
Will we store “real” files on a filesystem or just provide a fixture for aiofiles?
Will we inject real dependencies, or use mocks and monkey patching?

There is no right and wrong here, as different testing strategies (or a combination thereof) have their own advantages in terms of test running time, how easy it is to implement new tests, how similar the test environment is to production, etc.

Another thing to choose is a testing framework. Just as in the WSGI tutorial, let’s use pytest. This is a matter of taste; if you prefer xUnit/JUnit-style layout, you’ll feel at home with the stdlib’s unittest.

In order to more quickly deliver a working solution, we’ll allow our tests to access the real filesystem. For our convenience, pytest offers several temporary directory utilities out of the box. Let’s wrap its tmpdir_factory to create a simple storage_path fixture that we’ll share among all tests in the suite (in the pytest parlance, a “session”-scoped fixture).

Tip

The pytest website includes in-depth documentation on the use of fixtures. Please visit pytest fixtures: explicit, modular, scalable to learn more.

As mentioned in the previous section, there are many ways to spin up a temporary or permanent Redis server; or mock it altogether. For our tests, we’ll try fakeredis, a pure Python implementation tailored specifically for writing unit tests.

pytest and fakeredis can be installed as:

$ pip install fakeredis pytest

We’ll also create a directory for our tests and make it a Python package to avoid any problems with importing local utility modules or checking code coverage:

$ mkdir -p tests
$ touch tests/__init__.py

Next, let’s implement fixtures to replace uuid and aioredis, and inject them into our tests via conftest.py (place your code in the newly created tests directory):

import io
import random
import uuid

import fakeredis.aioredis
import falcon.asgi
import falcon.testing
import PIL.Image
import PIL.ImageDraw
import pytest

from asgilook.app import create_app
from asgilook.config import Config


@pytest.fixture()
def predictable_uuid():
    fixtures = (
        uuid.UUID('36562622-48e5-4a61-be67-e426b11821ed'),
        uuid.UUID('3bc731ac-8cd8-4f39-b6fe-1a195d3b4e74'),
        uuid.UUID('ba1c4951-73bc-45a4-a1f6-aa2b958dafa4'),
    )

    def uuid_func():
        try:
            return next(fixtures_it)
        except StopIteration:
            return uuid.uuid4()

    fixtures_it = iter(fixtures)
    return uuid_func


@pytest.fixture(scope='session')
def storage_path(tmpdir_factory):
    return tmpdir_factory.mktemp('asgilook')


@pytest.fixture
def client(predictable_uuid, storage_path):
    config = Config()
    config.create_redis_pool = fakeredis.aioredis.create_redis_pool
    config.redis_host = None
    config.storage_path = storage_path
    config.uuid_generator = predictable_uuid

    app = create_app(config)
    return falcon.testing.TestClient(app)


@pytest.fixture(scope='session')
def png_image():
    image = PIL.Image.new('RGBA', (640, 360), color='black')

    draw = PIL.ImageDraw.Draw(image)
    for _ in range(32):
        x0 = random.randint(20, 620)
        y0 = random.randint(20, 340)
        x1 = random.randint(20, 620)
        y1 = random.randint(20, 340)
        if x0 > x1:
            x0, x1 = x1, x0
        if y0 > y1:
            y0, y1 = y1, y0
        draw.ellipse([(x0, y0), (x1, y1)], fill='yellow', outline='red')

    output = io.BytesIO()
    image.save(output, 'PNG')
    return output.getvalue()


@pytest.fixture(scope='session')
def image_size():
    def report_size(data):
        image = PIL.Image.open(io.BytesIO(data))
        return image.size

    return report_size

Note

In the png_image fixture above, we are drawing random images that will look different every time the tests are run.

If your testing flow affords that, it is often a great idea to introduce some unpredictability in your test inputs. This will provide more confidence that your application can handle a broader range of inputs than just 2-3 test cases crafted specifically for that sole purpose.

On the other hand, random inputs can make assertions less stringent and harder to formulate, so judge according to what is the most important for your application. You can also try to combine the best of both worlds by using a healthy mix of rigid fixtures and fuzz testing.

Note

More information on conftest.py's anatomy and pytest configuration can be found in the latter’s documentation: conftest.py: local per-directory plugins.

With the groundwork in place, we can start with a simple test that will attempt to GET the /images resource. Place the following code in a new tests/test_images.py module:

def test_list_images(client):
    resp = client.simulate_get('/images')

    assert resp.status_code == 200
    assert resp.json == []

Let’s give it a try:

$ pytest tests/test_images.py

========================= test session starts ==========================
platform linux -- Python 3.8.0, pytest-6.2.1, py-1.10.0, pluggy-0.13.1
rootdir: /falcon/tutorials/asgilook
collected 1 item

tests/test_images.py .                                           [100%]

========================== 1 passed in 0.01s ===========================

Success! 🎉

At this point, our project structure, containing the asgilook and test packages, should look like this:

asgilook
├── .venv
├── asgilook
│   ├── __init__.py
│   ├── app.py
│   ├── asgi.py
│   ├── cache.py
│   ├── config.py
│   ├── images.py
│   └── store.py
└── tests
    ├── __init__.py
    ├── conftest.py
    └── test_images.py

Now, we need more tests! Try adding a few more test cases to tests/test_images.py, using the WSGI Testing Tutorial as your guide (the interface for Falcon’s testing framework is mostly the same for ASGI vs. WSGI). Additional examples are available under examples/asgilook/tests in the Falcon repository.

Tip

For more advanced test cases, the falcon.testing.ASGIConductor class is worth a look.

Code Coverage¶

How much of our asgilook code is covered by these tests?

An easy way to get a coverage report is by using the pytest-cov plugin (available on PyPi).

After installing pytest-cov we can generate a coverage report as follows:

$ pytest --cov=asgilook --cov-report=term-missing tests/

Oh, wow! We do happen to have full line coverage, except for asgilook/asgi.py. If desired, we can instruct coverage to omit this module by listing it in the omit section of a .coveragerc file.

What is more, we could turn the current coverage into a requirement by adding --cov-fail-under=100 (or any other percent threshold) to our pytest command.

Note

The pytest-cov plugin is quite simplistic; more advanced testing strategies such as blending different types of tests and/or running the same tests in multiple environments would most probably involve running coverage directly, and combining results.

What Now?¶

Congratulations, you have successfully completed the Falcon ASGI tutorial!

Needless to say, our sample ASGI application could still be improved in numerous ways:

Make the image store persistent and reusable across worker processes. Maybe by using a database?
Improve error handling for malformed images.
Check how and when Pillow releases the GIL, and tune what is offloaded to a threadpool executor.
Test Pillow-SIMD to boost performance.
Publish image upload events via SSE or WebSockets.
…And much more (patches welcome, as they say)!

Compared to the sync version, asynchronous code can at times be harder to design and reason about. Should you run into any issues, our friendly community is available to answer your questions and help you work through any sticky problems (see also: Getting Help).

Tutorial (ASGI)¶

First Steps¶

Hosting Our App¶

Configuration¶

Image Store¶

Images Resource(s)¶

Running Our Application¶

Dynamic Thumbnails¶

Caching Responses¶

Testing Our Application¶

Code Coverage¶

What Now?¶

Patrons

Falcon

Navigation

Related Topics