flask_command_example.py

import boto3
from boto3.dynamodb.conditions import Key
from pydantic import BaseModel, ValidationError
from pydantic import EmailStr

from flask import Flask, jsonify, request, Response

app = Flask(__name__)


class NotFound(Exception):
    pass


class AlreadyExists(Exception):
    pass


class User(BaseModel):
    username: EmailStr
    name: str
    is_active: bool

    @classmethod
    def get_by_username(cls, username: str):
        table = boto3.resource('dynamodb').Table('users')
        response = table.query(
            KeyConditionExpression=Key('PK').eq(username) & Key('SK').eq(username)
        )
        try:
            user = cls(**response['Items'][0])
        except IndexError:
            raise NotFound

        return user

    def save(self):
        table = boto3.resource('dynamodb').Table('users')
        table.put_item(
            Item={
                'PK': self.username,
                'SK': self.username,
                **self.dict()
            }
        )


class AddUserCommand(BaseModel):
    username: EmailStr
    name: str

    def execute(self) -> dict:
        try:
            User.get_by_username(self.username)
            raise AlreadyExists
        except NotFound:
            pass

        user = User(
            username=self.username,
            name=self.name,
            is_active=True
        )
        user.save()

        return user.dict()


@app.errorhandler(ValidationError)
def handle_bad_request(exc):
    return Response(exc.json(), mimetype='application/json', status=400)


@app.route('/add-user/', methods=['POST'])
def add_user():
    cmd = AddUserCommand(
        **request.json
    )
    return jsonify(cmd.execute())


if __name__ == '__main__':
    app.run()