app.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 CommandException(Exception):
    code = 500


class NotFound(CommandException):
    code = 404


class AlreadyExists(CommandException):
    code = 400


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_validation_error(exc):
    return Response(exc.json(), mimetype='application/json', status=400)


@app.errorhandler(CommandException)
def handle_command_exception(exc):
    return Response(
        {'details': exc.__class__.__name__},
        mimetype='application/json',
        status=exc.code
    )


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


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

test_flask_command_example.py

import json
import pathlib

import boto3
import pytest
from jsonschema import validate

from .app import AddUserCommand, User, app, AlreadyExists


@pytest.fixture
def database_table():
    # it's created, too long creation
    yield
    # clear all records
    table = boto3.resource('dynamodb').Table('users')
    last_key = None

    while True:
        query_kwargs = {}
        if last_key is not None:
            query_kwargs['ExclusiveStartKey'] = last_key
        response = table.scan(**query_kwargs)
        with table.batch_writer() as batch:
            for item in response['Items']:
                batch.delete_item(
                    Key={
                        'PK': item['PK'],
                        'SK': item['SK']
                    }
                )

        last_key = response.get('LstEvaluatedKey')

        if last_key is None:
            break


@pytest.fixture
def client():
    app.config['TESTING'] = True

    with app.test_client() as client:
        yield client


def test_add_user_command(database_table):
    cmd = AddUserCommand(
        username='john@doe.com',
        name='John Doe'
    )

    cmd.execute()

    user = User.get_by_username('john@doe.com')

    assert user.username == 'john@doe.com'
    assert user.name == 'John Doe'


def test_add_user_command_already_exists(database_table):
    User(name='John Doe', username='john@doe.com', is_active=True).save()
    cmd = AddUserCommand(
        username='john@doe.com',
        name='John Doe'
    )

    with pytest.raises(AlreadyExists):
        cmd.execute()


def test_add_user(client, database_table):
    response_schema = json.loads(
        pathlib.Path(
            f'{pathlib.Path(__file__).parent.absolute()}/schemas/User.json'
        ).read_bytes()
    )
    response = client.post(
        '/add-user/',
        data=json.dumps(
            {
                'username': 'jane@doe.com',
                'name': 'Jane Doe'
            }
        ),
        content_type='application/json'
    )

    assert response.status_code == 200
    validate(
        response.json,
        response_schema
    )


def test_add_user_missing_name(client, database_table):
    response = client.post(
        '/add-user/',
        data=json.dumps(
            {
                'username': 'jane@doe.com',
            }
        ),
        content_type='application/json'
    )

    assert response.status_code == 400


def test_add_user_already_exists(client, database_table):
    User(name='John Doe', username='john@doe.com', is_active=True).save()
    response = client.post(
        '/add-user/',
        data=json.dumps(
            {
                'username': 'john@doe.com',
                'name': 'John Doe'
            }
        ),
        content_type='application/json'
    )

    assert response.status_code == 400