aiohttp file uploading

Blog post can have images, let’s support them!

Extend Post structure

Add image BLOB field.

Field	Description	Type
id	Post id	int
title	Title	str
text	Content	str
owner	Post creator	str
editor	Last editor	str
image	Image content	bytes

Note

Usually you serve images by Content Delivery Network or save them on BLOB storage like Amazon S3. Database keeps an image URL only.

But for sake of simplicity we use database for storing the image content.

Image web handler

@router.get("/{post}/image")
async def render_post_image(request: web.Request) -> web.Response:
    post_id = request.match_info["post"]
    db = request.config_dict["DB"]
    async with db.execute("SELECT image FROM posts WHERE id = ?", [post_id]) as cursor:
        row = await cursor.fetchone()
        if row is None or row["image"] is None:
            img = PIL.Image.new("RGB", (64, 64), color=0)
            fp = io.BytesIO()
            img.save(fp, format="JPEG")
            content = fp.getvalue()
        else:
            content = row["image"]
    return web.Response(body=content, content_type="image/jpeg")

Return a black image if nothing is present in DB, the image content with image/jpeg content type otherwise.

To render image we need to change view.html template:

{% block content %}
<h1>{{ post.title }}</h1>
...
<p><img src="/{{ post.id }}/image"></p>
...
{% endblock %}

Upload form

Add <input type="file" name="image" accept="image/png, image/jpeg"> HTML form field:

{% block content %}
<h1>Edit: {{ post.title }}</h1>
<form action="/{{ post.id }}/edit" method="POST"
      enctype="multipart/form-data">
  ...
  <input type="file" name="image" accept="image/png, image/jpeg">
  ...
</form>
{% endblock %}

Handle uploaded image

@router.post("/{post}/edit")
async def edit_post_apply(request: web.Request) -> web.Response:
    post_id = request.match_info["post"]
    db = request.config_dict["DB"]
    post = await request.post()
    image = post.get("image")
    await db.execute(
        f"UPDATE posts SET title = ?, text = ? WHERE id = ?",
        [post["title"], post["text"], post_id],
    )
    if image:
        img_content = image.file.read()  # type: ignore
        await apply_image(db, post_id, img_content)
    await db.commit()
    raise web.HTTPSeeOther(location=f"/{post_id}/edit")

If a new image was provided by HTML form – update DB:

async def apply_image(
    db: aiosqlite.Connection, post_id: int, img_content: bytes
) -> None:
    buf = io.BytesIO(img_content)
    loop = asyncio.get_event_loop()
    img = PIL.Image.open(buf)
    new_img = await loop.run_in_executor(None, img.resize, (64, 64), PIL.Image.LANCZOS)
    out_buf = io.BytesIO()
    new_img.save(out_buf, format="JPEG")
    await db.execute(
        "UPDATE posts SET image = ? WHERE id = ?", [out_buf.getvalue(), post_id]
    )

Full example for server with images support

Example for HTML version of blogs server with images: Full server example with templates and image uploading