How to Serve Images Using aiohttp

In this tutorial, you will learn how to serve images using aiohttp in Python.

This tutorial will cover loading images from local storage, serving them directly from memory, optimizing image loading, fetching images from URLs, and implementing caching for image delivery.

 

 

Read and Serve Images

Load images from local storage

Use aiofiles to read images asynchronously from local storage.

import aiohttp
import aiofiles
from aiohttp import web
async def handle_local_image(request):
    file_path = 'images/sample.jpg'
    async with aiofiles.open(file_path, mode='rb') as f:
        image_data = await f.read()
    return web.Response(body=image_data, content_type='image/jpeg')
app = web.Application()
app.router.add_get('/image', handle_local_image)
web.run_app(app, host='127.0.0.1', port=8080)

When you access http://127.0.0.1:8080/image, the server responds with the image sample.jpg.

The server reads the image asynchronously from local storage and serves it.

Serve images directly from memory

Serve images directly from memory by storing them in a variable and returning them in the response.

import aiohttp
from aiohttp import web
async def handle_memory_image(request):
    image_data = b'\x89PNG\r\n\x1a\n...'  # Binary data of a PNG image
    return web.Response(body=image_data, content_type='image/png')
app = web.Application()
app.router.add_get('/memory-image', handle_memory_image)
web.run_app(app, host='127.0.0.1', port=8080)

When you access http://127.0.0.1:8080/memory-image, it returns the image stored in memory.

The image is served directly from memory.

You can convert the image to binary to use it in the above code like this:

def image_to_binary(file_path):
    with open(file_path, 'rb') as image_file:
        binary_data = image_file.read()
    return binary_data
image_data = image_to_binary('sample.jpg')

 

Autodetect Content Type

You can use mimetypes.guess_type to dynamically determine the content type for different image formats.

import aiohttp
import aiofiles
from aiohttp import web
import mimetypes
async def handle_image(request):
    file_path = 'images/sample.png'
    mime_type, _ = mimetypes.guess_type(file_path)
    async with aiofiles.open(file_path, mode='rb') as f:
        image_data = await f.read()
    return web.Response(body=image_data, content_type=mime_type)
app = web.Application()
app.router.add_get('/dynamic-image', handle_image)
web.run_app(app, host='127.0.0.1', port=8080)

The server responds with the image sample.png and sets the correct content type based on the file extension.

 

Read Large Images (Chunking)

Read large images in chunks to optimize memory usage.

import aiohttp
import aiofiles
from aiohttp import web
async def handle_large_image(request):
    file_path = 'images/large_sample.jpg'
    response = web.StreamResponse(status=200, reason='OK', headers={'Content-Type': 'image/jpeg'})
    await response.prepare(request)
    async with aiofiles.open(file_path, mode='rb') as f:
        chunk = await f.read(1024)
        while chunk:
            await response.write(chunk)
            chunk = await f.read(1024)
    return response
app = web.Application()
app.router.add_get('/large-image', handle_large_image)
web.run_app(app, host='127.0.0.1', port=8080)

The server streams the large image large_sample.jpg in chunks.

By reading and writing the image in chunks, the server handles large files without excessive memory consumption.

 

Serve Images from Remote URLs

You can fetch images from remote URLs and serve them directly to clients.

import aiohttp
from aiohttp import web
async def handle_remote_image(request):
    url = 'https://example.com/sample.jpg'
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            image_data = await resp.read()
            return web.Response(body=image_data, content_type='image/jpeg')
app = web.Application()
app.router.add_get('/remote-image', handle_remote_image)
web.run_app(app, host='127.0.0.1', port=8080)

The server fetches and serves the image from https://example.com/sample.jpg.

The server uses aiohttp.ClientSession to download images on the fly.

You can manipulate the image before serving it.

Using aiohttp.ClientSession

You can use aiohttp.ClientSession for image downloading and serving.

import aiohttp
from aiohttp import web
async def handle_dynamic_image(request):
    url = 'https://example.com/dynamic.jpg'
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            image_data = await resp.read()
            return web.Response(body=image_data, content_type='image/jpeg')
app = web.Application()
app.router.add_get('/dynamic-image', handle_dynamic_image)
web.run_app(app, host='127.0.0.1', port=8080)

The server downloads and serves the image from https://example.com/dynamic.jpg.

If you access http://127.0.0.1:8080/dynamic-image, you can see the downloaded image.

 

Caching Images

Implement in-memory caching

Implement caching to reduce load times for frequently accessed images.

import aiohttp
from aiohttp import web
cache = {}
async def handle_cached_image(request):
    url = 'https://example.com/cached.jpg'
    if url in cache:
        return web.Response(body=cache[url], content_type='image/jpeg')
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            image_data = await resp.read()
            cache[url] = image_data
            return web.Response(body=image_data, content_type='image/jpeg')
app = web.Application()
app.router.add_get('/cached-image', handle_cached_image)
web.run_app(app, host='127.0.0.1', port=8080)

The server caches the image from https://example.com/cached.jpg for faster subsequent access.

If you remove the source image or disconnect from the source, you can still serve the image from the cached memory.

Using aiohttp middlewares for caching

You can use aiohttp middlewares to implement caching logic across multiple routes.

import aiohttp
from aiohttp import web
cache = {}

@web.middleware
async def cache_middleware(request, handler):
    url = str(request.url)
    if url in cache:
        return web.Response(body=cache[url], content_type='image/jpeg')
    response = await handler(request)
    cache[url] = response.body
    return response
async def handle_middleware_image(request):
    url = 'https://example.com/middleware.jpg'
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            image_data = await resp.read()
            return web.Response(body=image_data, content_type='image/jpeg')
app = web.Application(middlewares=[cache_middleware])
app.router.add_get('/middleware-image', handle_middleware_image)
web.run_app(app, host='127.0.0.1', port=8080)

The server uses middleware to cache images to improve performance across multiple routes.

Middlewares provide a centralized way to implement caching.