WTForms with images in data-storage services

Mar 31, 2023 flask python sqlalchemy wtforms

In this article we will extend WTForms to create form with images stored in a data-storage service, such as AWS S3, GCP Storage or Azure Storage.

Prerequisites

In this example we will use Flask, Flask-SQLAlchemy and Flask-WTF, but the same thing can be done using other web frameworks and ORMs. We will need pillow as well in case we want to require images with specific width, height or aspect ratio, and a library to upload images to the storage service (e.g., boto3 for S3).

Define the models

For this example, create the tables images and monkeys:

class Image(Model):
    __tablename__ = 'images'

    id = db.Column(db.Integer, primary_key=True)
    url = db.Column(db.String(256), nullable=False, unique=True)


class Monkey(Model):
    __tablename__ = 'monkeys'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), nullable=False)
    image_id = db.Column(db.Integer, db.ForeignKey('images.id'))

    image = db.relationship('Image',
                            single_parent=True,
                            cascade='all,delete-orphan')

Extend WTForms objects

Create an image input and a validator to restrict the input type to images:

from flask_wtf.file import FileAllowed
from markupsafe import Markup
from werkzeug.datastructures import FileStorage
from wtforms.widgets import FileInput


class ImageInput(FileInput):

    def __call__(self, field, **kwargs):
        kwargs.setdefault('id', field.id)
        kwargs.setdefault('type', self.input_type)
        flags = getattr(field, 'flags', {})
        for k in dir(flags):
            if k in self.validation_attrs and k not in kwargs:
                kwargs[k] = getattr(flags, k)

        classes = ['image-input']
        loaded = field.data and not isinstance(field.data, FileStorage)
        if loaded:
            classes.append('loaded')
        classes_str = ' '.join(classes)
        html = f'<div class="{classes_str}">'
        html += '<input %s>' % self.html_params(name=field.name, **kwargs)
        if loaded:
            html += f'<img src="{field.data.url}">'
        else:
            html += '<img>'
        html += '</div>'
        return Markup(html)


class ImageAllowed(FileAllowed):

    def __init__(self, upload_set=('jpg', 'jpeg', 'png'), message=None):
        super().__init__(upload_set, message)
        mime_types = []
        for allowed_type in upload_set:
            if allowed_type.startswith('image/'):
                mime_type = allowed_type
            else:
                mime_type = f'image/{allowed_type}'
            mime_types.append(mime_type)
        self.field_flags = {'accept': ', '.join(mime_types)}

The input renders an HTML input element with “file” type, along with an img and wrapped by a div with class image-input:

<div class="image-input">
  <input accept="image/jpeg, image/jpg" id="image" name="image" type="file">
  <img>
</div>

With a bit of javascript and css we can make this input to display the image once the input changed, so we can see the image we are going to upload:

for (let element of document.querySelectorAll('.image-input input')) {
  element.onchange = function(ev) {
    let div = this.parentNode;
    let image = div.querySelector('img');
    let reader = new FileReader();
    reader.onload = () => {
      image.src = reader.result;
      div.classList.add('loaded');
    };
    reader.readAsDataURL(ev.target.files[0]);
  };
}

.image-input {
    display: block;
    width: 400px;
    height: 400px;
    border: solid 1px #dadce0;
    position: relative;
    overflow: clip;
}
.image-input.loaded {
    height: auto;
}
.image-input input[type="file"] {
    height: 100%;
    width: 100%;
    opacity: 0;
    position: absolute;
    top: 0;
    left: 0;
    z-index: 10;
    cursor: pointer;
}
.image-input img {
    width: 100%;
    display: block;
}

In case we want to add restrictions of image size and aspect ratio we can create another validator:

from PIL import Image
from werkzeug.datastructures import FileStorage
from wtforms.validators import ValidationError


class ImageSize:

    def __init__(self,
                 width=None,
                 width_tolerance=0,
                 height=None,
                 height_tolerance=0,
                 aspect_ratio=None,
                 aspect_ratio_tolerance=.01):
        assert width or height or aspect_ratio
        self.width = width
        self.width_tolerance = abs(width_tolerance)
        self.height = height
        self.height_tolerance = abs(height_tolerance)
        self.aspect_ratio = aspect_ratio
        self.aspect_ratio_tolerance = abs(aspect_ratio_tolerance)

    def __call__(self, form, field):
        if field.data and isinstance(field.data, FileStorage):
            image = Image.open(field.data)
            field.data.seek(0)
            if self.width:
                if not self._is_close(self.width,
                                      image.width,
                                      self.width_tolerance):
                    raise ValidationError('Invalid width')

            if self.height:
                if not self._is_close(self.height,
                                      image.height,
                                      self.height_tolerance):
                    raise ValidationError('Invalid height')

            if self.aspect_ratio and not (self.width and self.height):
                if not self._is_close(self.aspect_ratio,
                                   image.width/image.height,
                                   self.aspect_ratio_tolerance):
                    raise ValidationError('Invalid aspect ratio')

    def _is_close(self, a, b, tol):
        return abs(a-b) <= tol

Using ImageSize we can restrict the width to 1000px (ImageSize(width=1000)) or allow only squared images (ImageSize(aspect_ratio=1)). Note that we add tolerance variables to allow small errors, specially relevant when calculating aspect ratios since we need to divide integers and round down.

Then, we create the ImageField, the field we will use in our forms, with a default ImageAllowed validator:

from flask_wtf.file import FileField


class ImageField(FileField):
    widget = ImageInput()

    def __init__(self, label=None, validators=None, *args, **kwargs):
        if not validators:
            validators = [ImageAllowed()]
        elif not any(isinstance(i, ImageAllowed) for i in validators):
            validators += [ImageAllowed()]
        super().__init__(label, validators, *args, **kwargs)

Using the image field in WTForms

Now we can add an image field to our form. The upload process can be added to the form in populate_obj (this method runs after the form validation) or in the view (after validation, when calling form.populate_obj). Here is how to do it in the form:

from werkzeug.datastructures import FileStorage
from wtforms import StringField
from wtforms.validators import DataRequired

from app.models import Image as ImageModel  # In case PIL.Image is imported.
from app.utils import upload_image


class MonkeyForm(FlaskForm):
    name = StringField(validators=[DataRequired()])
    image = ImageField(validators=[ImageAllowed(['jpeg', 'jpg']),
                                   ImageSize(width=1920)])

    def populate_obj(self, obj):
        fs = None
        if isinstance(self.image.data, FileStorage):
            fs = self.image.data
            self.image.data = None
        super().populate_obj(obj)
        if fs:
            url = upload_image(fs, folder='monkeys')
            obj.image = ImageModel(url=url)

As you can see, a function upload_image is expected. This function uploads the FileStorage variable to the storage service and return its URL. Here’s an example for S3:

import os
from tempfile import mktemp

import boto3


AWS_ACCESS_KEY_ID = os.environ['AWS_ACCESS_KEY_ID']
AWS_SECRET_ACCESS_KEY = os.environ['AWS_SECRET_ACCESS_KEY']
AWS_DEFAULT_REGION = os.environ['AWS_DEFAULT_REGION']
S3_BUCKET_NAME = os.environ['S3_BUCKET_NAME']


def upload_image(fs, folder=None):
    path = mktemp()
    fs.save(path)
    s3_path = fs.filename
    if folder:
        s3_path = f'{folder}/{s3_path}'
    s3 = boto3.client('s3',
                      aws_access_key_id=AWS_ACCESS_KEY_ID,
                      aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
                      region_name=AWS_DEFAULT_REGION)
    s3.upload_file(Bucket=S3_BUCKET_NAME,
                   Filename=path,
                   Key=s3_path,
                   ExtraArgs={'ContentType': fs.mimetype})
    return f'https://{S3_BUCKET_NAME}.s3.{AWS_DEFAULT_REGION}' \
           f'.amazonaws.com/{s3_path}'