Introduction

Application Overview

After completing this course, we will have built a RESTful API using Test Driven Development. The API will following RESTful design principle, using basic HTTP Methods.

API will be built with Flask as framework and SQLAlchemy as ORM.

Chapter 1

In chapter 1 we'll structurize our project with Flask, SQLAlchemy, Postgres and Docker.

• Getting Started
• Docker Setup
• PostgreSQL Setup

Getting Started

In this chapter we'll setup basic structure for our projects to help our test driven development flow.

Setup

Create new project and requirements with

$ mkdir flask-todo
$ cd flask-todo

We need to initialize our project directory as git repository with command

$ git init

Then we'll use python virtual environment to keep dependencies required by our projects isolated.

$ python -m venv venv
$ source venv/bin/activate

After activating python virtual environment, install required package

(venv)$ pip install flask==2.2.2

To save our project dependencies we can create requirements.txt containing

Flask==2.2.2

Create app package inside our project directory with command

(venv)$ mkdir app
(venv)$ touch app/__init__.py

Create Route helper class in http package inside our app

(venv)$ mkdir app/http
(venv)$ touch app/http/__init__.py

# app/http/route.py

from __future__ import annotations

from typing import List, Optional, Type, Union

from flask import Blueprint
from flask.typing import RouteCallable
from flask.views import View


class Route:
    def __init__(
        self,
        url_rule: str,
        view: Type[View],
        method: Optional[str] = None,
        name_alias: Optional[str] = None,
    ) -> None:
        self.url_rule = url_rule
        self.view = view
        self.method = method
        self.name_alias = name_alias

    def view_func(self) -> RouteCallable:
        view_name = self.name_alias or self.view.__name__
        return self.view.as_view(view_name)

    def methods(self) -> List[str]:
        if self.method is None:
            methods = getattr(self.view, "methods", None) or ("GET",)
        else:
            methods = [self.method]
        return methods

    @classmethod
    def get(cls, url_rule: str, view: Type[View], name_alias: Optional[str] = None):
        return cls(url_rule=url_rule, view=view, method="GET", name_alias=name_alias)

    @classmethod
    def post(cls, url_rule: str, view: Type[View], name_alias: Optional[str] = None):
        return cls(url_rule=url_rule, view=view, method="POST", name_alias=name_alias)

    @classmethod
    def put(cls, url_rule: str, view: Type[View], name_alias: Optional[str] = None):
        return cls(url_rule=url_rule, view=view, method="PUT", name_alias=name_alias)

    @classmethod
    def delete(cls, url_rule: str, view: Type[View], name_alias: Optional[str] = None):
        return cls(url_rule=url_rule, view=view, method="DELETE", name_alias=name_alias)

    @staticmethod
    def group(url_prefix_group: str, routes: List[Union[Route, Blueprint]], **kwargs):
        name = kwargs.pop("name", None) or url_prefix_group.replace("/", "")
        url_prefix = kwargs.pop("url_prefix", None) or url_prefix_group
        blueprint = Blueprint(name, __name__, url_prefix=url_prefix, **kwargs)

        for route in routes:
            if isinstance(route, Blueprint):
                blueprint.register_blueprint(route)
            else:
                blueprint.add_url_rule(
                    route.url_rule, view_func=route.view_func(), methods=route.methods()
                )

        return blueprint

To isolate our application configuration create Config object inside our app

# app/config.py

import os
from functools import lru_cache


class Config:
    DEBUG = False


class ProductionConfig(Config):
    pass


class DevelopmentConfig(Config):
    DEBUG = True


class TestingConfig(Config):
    pass


@lru_cache
def get_config():
    config = {
        "production": ProductionConfig,
        "development": DevelopmentConfig,
        "testing": TestingConfig,
    }
    env = os.getenv("APP_ENV", "development")
    return config.get(env, DevelopmentConfig)

Then we can create factory.py file inside app directory

# app/factory.py

from typing import List, Type, Union

from flask import Blueprint, Flask

from app.config import Config
from app.http.route import Route


def create_app(
    app_name: str, config: Type[Config], routes: List[Union[Route, Blueprint]]
):
    app = Flask(app_name)
    app.config.from_object(config)

    for route in routes:
        if isinstance(route, Blueprint):
            app.register_blueprint(route)
        else:
            app.add_url_rule(
                route.url_rule, view_func=route.view_func(), methods=route.methods()
            )

    return app

Next we will create basic handler for our application, by creating package views with main_api module

# app/views/main_api.py

from flask import jsonify
from flask.typing import ResponseReturnValue
from flask.views import View


class MainView(View):
    def dispatch_request(self) -> ResponseReturnValue:
        resp = {"message": "Application running."}
        return jsonify(resp)

Then we can create basic routing for our application, first create routes package with api module

# app/routes/api.py

from app.http.route import Route
from app.views.main_api import MainView

api_routes = [
    Route.get("/", MainView),
]

Next update our app package __init__.py scripts to glue all our code before

# app/__init__.py

from app.config import get_config
from app.factory import create_app
from app.routes.api import api_routes

app = create_app("flask-todo", config=get_config(), routes=api_routes)

Next we'll create script to run our application on root of project

# serve.py

from app import app

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

Run project by invoke command

(venv)$ python serve.py

Navigate to http://localhost:5000 then you should see

{
  "message": "Application running."
}

To exclude unwanted files in our codebase repository, we can create .gitignore file containing

__pycache__/
venv/

Your final project structure should look like this

flask-todo
├── app
│   ├── config.py
│   ├── factory.py
│   ├── http
│   │   ├── __init__.py
│   │   └── route.py
│   ├── __init__.py
│   ├── routes
│   │   ├── api.py
│   │   └── __init__.py
│   └── views
│       ├── __init__.py
│       └── main_api.py
├── .gitignore
├── requirements.txt
└── serve.py

4 directories, 12 files

We can commit our works before continuing to next chapter.

(venv)$ git add .
(venv)$ git commit -m "Initialize project structure"

Docker Setup

This chapter require docker and docker compose installed

Let's put our application inside docker container

Setup Application Container

Create Dockerfile in our project directory

FROM python:3.8-slim

# Set environment varibles
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

WORKDIR /usr/src/app

# Install dependencies
RUN pip install --upgrade pip
COPY ./requirements.txt .

RUN pip install -r requirements.txt

COPY . .

Create .dockerignore file to avoid unnecessarily sending large or sensitive files and directories to the daemon and potentially adding them to images.

__pycache__/
venv/
Dockerfile
env
.env

Create docker-compose.yml file to manage our container via docker compose

version: '3.8'

services:
  todos:
    build: .
    command: python serve.py
    volumes:
      - .:/usr/src/app/
    ports:
      - 5000:5000
    env_file:
      - ./.env.dev

Create file .env.dev to add env based configuration

APP_ENV=development

Before build our image, we should add slight changes to our serve.py

from app import app

if __name__ == "__main__":
    app.run(host="0.0.0.0")

This will tell our application to use address 0.0.0.0 instead of 127.0.0.1.

Next we can build our docker with command

$ docker compose build

After finish the build, we can start our services with

$ docker compose up -d

Navigating to http://localhost:5000 should return same response as before

{
  "message": "Application running."
}

We can check our container logs with command docker compose logs {container_name} -f

$ docker compose logs todos -f

To shutdown our container we can invoke docker compose down -v

Your final project structure should look like this

flask-todo
├── app
│   ├── config.py
│   ├── factory.py
│   ├── http
│   │   ├── __init__.py
│   │   └── route.py
│   ├── __init__.py
│   ├── routes
│   │   ├── api.py
│   │   └── __init__.py
│   └── views
│       ├── __init__.py
│       └── main_api.py
├── docker-compose.yml
├── Dockerfile
├── .dockerignore
├── .env.dev
├── .gitignore
├── requirements.txt
└── serve.py

4 directories, 16 files

We can commit our works before continuing to next chapter.

(venv)$ git add .
(venv)$ git commit -m "Docker Setup"

PostgreSQL Setup

In this chapter we'll setup PostgreSQL as primary database in our application.

Setup PostgreSQL Container

Create db directory inside our project workspaces

(venv)$ mkdir db
(venv)$ touch ./db/create.sql

Update create.sql file with following content

CREATE DATABASE todos_production;
CREATE DATABASE todos_development;
CREATE DATABASE todos_testing;

Create Dockerfile inside db directory with following content

FROM postgres:14.6-alpine

COPY create.sql /docker-entrypoint-initdb.d

Update docker-compose.yml to

This is for educational purpose only, consider to hash your credential before commit your code.

version: '3.8'

services:
  postgres:
    build: ./db
    volumes:
      - postgres_data:/var/lib/postgresql/data/
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=password

  todos:
    build: .
    command: python serve.py
    volumes:
      - .:/usr/src/app/
    ports:
      - 5000:5000
    env_file:
      - ./.env.dev
    depends_on:
      postgres:
        condition: service_started

volumes:
  postgres_data:

After setup database in docker container, we can add required library to our app for this project. Update our requirements.txt to following content

Flask==2.2.2
SQLAlchemy==1.4.45
Flask-SQLAlchemy==3.0.2
Flask-Migrate==4.0.0
psycopg2-binary==2.9.5

Before build our image, update main Dockerfile to following content

FROM python:3.8-slim

# Set environment varibles
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

RUN mkdir -p /usr/src/app
WORKDIR /usr/src/app

# Install dependencies
RUN pip install --upgrade pip
COPY ./requirements.txt .

RUN pip install -r requirements.txt

COPY . .

Then we can rebuild our docker image by invoke command

$ docker compose down
$ docker compose build
$ docker compose up -d

To check wether our database created or not, we can invoke psql from our postgres container

$ docker compose exec postgres psql -U postgres

When in psql interface we can invoke \l to list database in the container

postgres=# \l
                                     List of databases
       Name        |  Owner   | Encoding |  Collate   |   Ctype    |   Access privileges
-------------------+----------+----------+------------+------------+-----------------------
 postgres          | postgres | UTF8     | en_US.utf8 | en_US.utf8 |
 template0         | postgres | UTF8     | en_US.utf8 | en_US.utf8 | =c/postgres          +
                   |          |          |            |            | postgres=CTc/postgres
 template1         | postgres | UTF8     | en_US.utf8 | en_US.utf8 | =c/postgres          +
                   |          |          |            |            | postgres=CTc/postgres
 todos_development | postgres | UTF8     | en_US.utf8 | en_US.utf8 |
 todos_production  | postgres | UTF8     | en_US.utf8 | en_US.utf8 |
 todos_testing     | postgres | UTF8     | en_US.utf8 | en_US.utf8 |
(6 rows)

Next we can setup database connection on our application config

This is for educational purpose only, consider to hash your credential before commit your code.

import os
from functools import lru_cache


class Config:
    DEBUG = False
    SQLALCHEMY_TRACK_MODIFICATIONS = False


class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = "postgresql://postgres:password@postgres:5432/todos_production"


class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = "postgresql://postgres:password@postgres:5432/todos_development"


class TestingConfig(Config):
    TESTING=True
    SQLALCHEMY_DATABASE_URI = "postgresql://postgres:password@postgres:5432/todos_testing"


@lru_cache
def get_config():
    config = {
        "production": ProductionConfig,
        "development": DevelopmentConfig,
        "testing": TestingConfig,
    }
    env = os.getenv("APP_ENV", "development")
    return config.get(env, DevelopmentConfig)

Then update our factory.py script to

from typing import List, Type, Union

from flask import Blueprint, Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

from app.config import Config
from app.http.route import Route

db = SQLAlchemy()
migrate = Migrate()

def create_app(
    app_name: str, config: Type[Config], routes: List[Union[Route, Blueprint]]
):
    app = Flask(app_name)
    app.config.from_object(config)

    db.init_app(app)
    migrate.init_app(app, db)

    for route in routes:
        if isinstance(route, Blueprint):
            app.register_blueprint(route)
        else:
            app.add_url_rule(
                route.url_rule, view_func=route.view_func(), methods=route.methods()
            )

    return app

Then run migration init to generate migration related scripts

(venv)$ flask db init

This will create directory migrations to store our next model migrations.

Your final project structure should be look like this

flask-todo
├── app
│   ├── config.py
│   ├── factory.py
│   ├── http
│   │   ├── __init__.py
│   │   └── route.py
│   ├── __init__.py
│   ├── routes
│   │   ├── api.py
│   │   └── __init__.py
│   └── views
│       ├── __init__.py
│       └── main_api.py
├── db
│   ├── create.sql
│   └── Dockerfile
├── docker-compose.yml
├── Dockerfile
├── .dockerignore
├── .env.dev
├── .gitignore
├── migrations
│   ├── alembic.ini
│   ├── env.py
│   ├── README
│   ├── script.py.mako
│   └── versions
├── requirements.txt
└── serve.py

7 directories, 22 files

We can commit our works before continuing to next chapter.

(venv)$ git add .
(venv)$ git commit -m "PostgreSQL Connection Setup"

Chapter 2

In this chapter we'll starting build our features about RESTful API

• Create To Do API
• List To Do API
• Detail To Do API
• Update To Do API

Create ToDo API Endpoint

In this chapter, we will create endpoint /api/todos to create Task based on user needs.

Write the Test

Writing test for Flask application actually like writing function, but it have a few rules

• Test file name should starts with test_*.py or end with *_test.py, in this tutorial we will use latest pattern *_test.py
• Test method or function should start with test_
• In this tutorial we will cover using python standard library for unit testing unittest

Before create test case for our project, we need to define our base class that hold our test app. Let's create tests/base.py

import unittest
from app.factory import create_app
from app.config import TestingConfig
from app.routes.api import api_routes


class BaseAPITestCase(unittest.TestCase):
    def setUp(self) -> None:
        self.app = create_app("test-app", config=TestingConfig, routes=api_routes)
        self.client = self.app.test_client()
        self.ctx = self.app.app_context()
        self.ctx.push()

    def tearDown(self) -> None:
        self.ctx.pop()

This class will hold our app with TestingConfig, which connected to our testing specific database todos_testing.

Let's write our first test by create new file create_task_api_test.py to tests/features/tasks directory.

# tests/features/task/create_task_api_test.py

from http import HTTPStatus
from tests.base import BaseAPITestCase
from app.tasks.models.task import Task
from app.factory import db


class CreateTaskAPITest(BaseAPITestCase):
    def test_create_task(self):
        endpoint = "/api/todos"
        json_data = {
            "name": "Fix handrawer",
        }

        resp = self.client.post(
            endpoint,
            json=json_data
        )
        self.assertEqual(resp.status_code, HTTPStatus.CREATED)

        tasks = Task.query.all()
        self.assertEqual(1, len(tasks))

        Task.query.delete()
        db.session.commit()

Ensure your container is up and running by invoke command.

$ docker compose up -d

Then run the test with python unittest module.

$ docker compose exec todos python -m unittest tests/features/task/create_task_api.py

It will return failure message

F
======================================================================
FAIL: test_create_task (tests.features.task.create_task_api.CreateTaskAPITest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/src/app/tests/features/task/create_task_api.py", line 20, in test_create_task
    self.assertEqual(resp.status_code, HTTPStatus.CREATED)
AssertionError: 404 != <HTTPStatus.CREATED: 201>

----------------------------------------------------------------------
Ran 1 test in 0.006s

FAILED (failures=1)

This occured because we not define our Task resource yet. To define Task resource, first we need to create table to store the data by creating new migration, first we need to define our models.

First we need to create test againts our Task model, create file tests/unit/task/task_model_test.py

from app.factory import db
from app.tasks.models.task import Task
from tests.base import BaseAPITestCase


class TaskModelTest(BaseAPITestCase):
    def test_create_task(self):
        task = Task(name="Finish Homework")
        db.session.add(task)
        db.session.commit()

        tasks = Task.query.all()
        assert len(tasks) == 1

        db.session.delete(task)
        db.session.commit()

Then run the test

$ docker compose exec todos python -m unittest tests/unit/task/task_model_test.py

It will return failure message

E
======================================================================
ERROR: task_model_test (unittest.loader._FailedTest)
----------------------------------------------------------------------
ImportError: Failed to import test module: task_model_test
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/unittest/loader.py", line 154, in loadTestsFromName
    module = __import__(module_name)
  File "/usr/src/app/tests/unit/task/task_model_test.py", line 2, in <module>
    from app.modules.tasks.models import Task
ImportError: cannot import name 'Task' from 'app.modules.tasks.models' (/usr/src/app/app/modules/tasks/models.py)


----------------------------------------------------------------------
Ran 1 test in 0.000s

FAILED (errors=1)

Create Task model to file app/tasks/models/task.py

# app/tasks/models/task.py

from app.factory import db


class Task(db.Model): # type: ignore
    __tablename__ = "tasks"

    id = db.Column(db.BigInteger, primary_key=True)
    name = db.Column(db.String(255))
    completed = db.Column(db.Boolean, default=False)
    created_at = db.Column(db.DateTime, default=db.func.current_timestamp())
    updated_at = db.Column(
        db.DateTime,
        default=db.func.current_timestamp(),
        onupdate=db.func.current_timestamp(),
    )

Apparently our Models is not known to alembic migrations until it loaded to our app via routing registration, we need to create empty Handler for our routing. Create file app/tasks/views/create_task_api.py contains basic handler view.

# app/tasks/views/create_task_api.py

from flask import jsonify
from flask.typing import ResponseReturnValue
from flask.views import View

from app.tasks.models.task import Task

class CreateTaskAPI(View):
    def dispatch_request(self) -> ResponseReturnValue:
        return jsonify({})

Then register the handler to our app/routes/api.py

# app/routes/api.py

from app.http.route import Route
from app.tasks.views.create_task_api import CreateTaskAPI
from app.views.main_api import MainView

api_routes = [
    Route.get("/", MainView),
    Route.group("/api", routes=[
        Route.post("/todos", view=CreateTaskAPI)
    ]),
]

In this routing file, we group our /todos endpoint to /api group.

Generate db migration by running command

$ docker compose exec todos flask db migrate -m "create tasks table"

This will create new generated file on migrations/versions/{hash_number}_{message}.py

Then we can run our migration againts testing environment with command

$ docker compose exec -e APP_ENV=testing todos flask db upgrade

When we run our model test againts newly created table, it will return success

$ docker compose exec todos python -m unittest tests/unit/task/task_model_test.py

It will return message

.
----------------------------------------------------------------------
Ran 1 test in 0.027s

OK

Next try re-running our feature test tests/features/task/create_task_api_test.py

$ docker compose exec todos python -m unittest tests/features/task/create_task_api_test.py

It does not return HTTP Not Found Error (404) anymore but still fail our test

F
======================================================================
FAIL: test_create_task (tests.features.task.create_task_api_test.CreateTaskAPITest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/src/app/tests/features/task/create_task_api_test.py", line 16, in test_create_task
    self.assertEqual(resp.status_code, HTTPStatus.CREATED)
AssertionError: 200 != <HTTPStatus.CREATED: 201>

----------------------------------------------------------------------
Ran 1 test in 0.007s

FAILED (failures=1)

Let's make it green by implementing code at app/tasks/views/create_task_api.py

# app/tasks/views/create_task_api.py

from http import HTTPStatus

from flask import jsonify, request
from flask.typing import ResponseReturnValue
from flask.views import View

from app.factory import db
from app.tasks.models.task import Task


class CreateTaskAPI(View):
    def dispatch_request(self) -> ResponseReturnValue:
        json_data = request.get_json()

        task = Task(**json_data)

        db.session.add(task)
        db.session.commit()
        resp = {
            "msg": "Created"
        }
        return jsonify(resp), HTTPStatus.CREATED

Re-run the test tests/features/task/create_task_api_test.py

$ docker compose exec todos python -m unittest tests/features/task/create_task_api_test.py

It will pass

.
----------------------------------------------------------------------
Ran 1 test in 0.031s

OK

But this test only covers happy path only. We need to add tests for following scenario

• Empty json paylod
• Wrong json attributes

Before adding new tests, lets create helper to help invoke our test. Create file bin/test on our application directory.

#!/usr/bin/env bash

docker compose exec todos python -m unittest $@

Make file executable with

$ chmod +x bin/test

Now we can invoke our test with bin/test unit_test.py.

Let's add test for empty json payload to our test case

# tests/features/task/create_task_api_test.py
    ...
    def test_create_task_when_payload_empty(self):
        endpoint = "/api/todos"
        json_data = {}

        resp = self.client.post(
            endpoint,
            json=json_data
        )
        self.assertEqual(resp.status_code, HTTPStatus.BAD_REQUEST)

        Task.query.delete()
        db.session.commit()

Re-run test with

$ bin/test tests/features/task/create_task_api_test.py

It will raise error NotNullViolation, because attribute name is not allowed to be null.

.E
======================================================================
ERROR: test_create_task_when_payload_empty (tests.features.task.create_task_api_test.CreateTaskAPITest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1900, in _execute_context
    self.dialect.do_execute(
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
    cursor.execute(statement, parameters)
psycopg2.errors.NotNullViolation: null value in column "name" of relation "tasks" violates not-null constraint

To validate user request schema, let's add marshmallow package to our application code base by update requirements.txt.

Flask==2.2.2
SQLAlchemy==1.4.45
Flask-SQLAlchemy==3.0.2
Flask-Migrate==4.0.0
psycopg2==2.9.5
marshmallow==3.19.0
marshmallow-sqlalchemy==0.28.1
flask-marshmallow==0.14.0

Then rebuild our container with docker compose build command, then update our running container with docker compose up -d. After add flask-marshmallow package, then register to our application factory.

# app/factory.py

from typing import List, Type, Union

from flask import Blueprint, Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_marshmallow import Marshmallow

from app.config import Config
from app.http.route import Route

db = SQLAlchemy()
migrate = Migrate()
ma = Marshmallow()

def create_app(
    app_name: str, config: Type[Config], routes: List[Union[Route, Blueprint]]
):
    app = Flask(app_name)
    app.config.from_object(config)

    db.init_app(app)
    ma.init_app(app)
    migrate.init_app(app, db)

    for route in routes:
        if isinstance(route, Blueprint):
            app.register_blueprint(route)
        else:
            app.add_url_rule(
                route.url_rule, view_func=route.view_func(), methods=route.methods()
            )

    return app

Then create our TaskSchema on app/tasks/schemas/task_schema.py

# app/tasks/schemas/task_schema.py

from marshmallow import fields, Schema


class TaskSchema(Schema):
    id = fields.Integer(dump_only=True)
    name = fields.String(required=True)
    completed = fields.Boolean()
    created_at = fields.DateTime(dump_only=True)
    updated_at = fields.DateTime(dump_only=True)

Next implement validation on our view at app/tasks/views/create_task_api.py

# app/tasks/views/create_task_api.py

from http import HTTPStatus

from flask import jsonify, make_response, request
from flask.typing import ResponseReturnValue
from flask.views import View

from app.factory import db
from app.tasks.models.task import Task
from app.tasks.schemas.task_schema import TaskSchema


class CreateTaskAPI(View):
    def dispatch_request(self) -> ResponseReturnValue:
        task_schema = TaskSchema()
        json_data = request.get_json() or dict()
        errors = task_schema.validate(json_data)
        if errors:
            return make_response(
                jsonify(errors=str(errors)),
                HTTPStatus.BAD_REQUEST
            )

        task = Task(**json_data)

        db.session.add(task)
        db.session.commit()
        resp = {
            "data": task_schema.dump(task)
        }
        return jsonify(resp), HTTPStatus.CREATED

Re-run test with

$ bin/test tests/features/task/create_task_api_test.py

Test passing

..
----------------------------------------------------------------------
Ran 2 tests in 0.041s

OK

Next let's create test by passing wrong attributes

    ...
    def test_create_task_when_payload_has_unknown_attribute(self):
        endpoint = "/api/todos"
        json_data = {
            "name": "Fix new car",
            "completed": False,
            "other": "value",
        }

        resp = self.client.post(
            endpoint,
            json=json_data
        )
        self.assertEqual(resp.status_code, HTTPStatus.BAD_REQUEST)

        Task.query.delete()
        db.session.commit()

Re-run test with

$ bin/test tests/features/task/create_task_api_test.py

Test passing

...
----------------------------------------------------------------------
Ran 3 tests in 0.052s

OK

After our scenario has been implemented, we can run all our test we've created so far by running command

$ bin/test discover -s tests -p '*_test.py'

Ensure all test are passed

....
----------------------------------------------------------------------
Ran 4 tests in 0.064s

OK

Your final project structure should be look like this

flask-todo
├── app
│   ├── config.py
│   ├── factory.py
│   ├── http
│   │   ├── __init__.py
│   │   └── route.py
│   ├── __init__.py
│   ├── routes
│   │   ├── api.py
│   │   └── __init__.py
│   ├── tasks
│   │   ├── __init__.py
│   │   ├── models
│   │   │   ├── __init__.py
│   │   │   └── task.py
│   │   ├── schemas
│   │   │   ├── __init__.py
│   │   │   └── task_schema.py
│   │   └── views
│   │       ├── create_task_api.py
│   │       └── __init__.py
│   └── views
│       ├── __init__.py
│       └── main_api.py
├── bin
│   └── test
├── db
│   ├── create.sql
│   └── Dockerfile
├── docker-compose.yml
├── Dockerfile
├── .dockerignore
├── .env.dev
├── .gitignore
├── migrations
│   ├── alembic.ini
│   ├── env.py
│   ├── README
│   ├── script.py.mako
│   └── versions
│       └── 213adaf34061_create_tasks_tables.py
├── requirements.txt
├── serve.py
└── tests
    ├── base.py
    ├── features
    │   ├── __init__.py
    │   └── task
    │       ├── create_task_api_test.py
    │       └── __init__.py
    ├── __init__.py
    └── unit
        ├── __init__.py
        └── task
            ├── __init__.py
            └── task_model_test.py

17 directories, 39 files

We can commit our works before continuing to next chapter.

(venv)$ git add .
(venv)$ git commit -m "Create CreateTaskAPI endpoint"

List ToDo API

In this chapter we will create endpoint /api/todos to list all created Task in our database.

Write the Test

First let's create test at tests/features/task/list_task_api_test.py

from http import HTTPStatus
from tests.base import BaseAPITestCase
from app.tasks.models.task import Task
from app.factory import db


class ListTaskAPITest(BaseAPITestCase):
    def test_list_task_api(self):
        endpoint = "/api/todos"
        task_1 = Task(name="Finish Homework")
        task_2 = Task(name="Clean up closet")
        db.session.add_all([task_1, task_2])
        db.session.commit()

        resp = self.client.get(endpoint)

        self.assertEqual(resp.status_code, HTTPStatus.OK)

        resp_json = resp.json or dict()
        data = resp_json["data"]

        self.assertEqual(data[0]["name"], "Finish Homework")
        self.assertEqual(data[1]["name"], "Clean up closet")

        Task.query.delete()
        db.session.commit()

Invoke test by running command

$ bin/test tests/features/task/list_task_api_test.py

It will return error

F
======================================================================
FAIL: test_list_task_api (tests.features.task.list_task_api_test.ListTaskAPITest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/src/app/tests/features/task/list_task_api_test.py", line 17, in test_list_task_api
    self.assertEqual(resp.status_code, HTTPStatus.OK)
AssertionError: 405 != <HTTPStatus.OK: 200>

----------------------------------------------------------------------
Ran 1 test in 0.019s

FAILED (failures=1)

It raise error 405 (HTTP Method Not Allowed) since we not implement our routes yet. Let's implement our routes at file app/tasks/views/list_task_api.py

from http import HTTPStatus

from flask import jsonify
from flask.typing import ResponseReturnValue
from flask.views import View

from app.factory import db
from app.tasks.models.task import Task
from app.tasks.schemas.task_schema import TaskSchema


class ListTaskAPI(View):
    def dispatch_request(self) -> ResponseReturnValue:
        tasks_schema = TaskSchema(many=True)
        tasks = Task.query.all()

        resp = {
            "data": tasks_schema.dump(tasks)
        }
        return jsonify(resp), HTTPStatus.OK

Then register it app/routes/api.py

from app.http.route import Route
from app.tasks.views.list_task_api import ListTaskAPI
from app.tasks.views.create_task_api import CreateTaskAPI
from app.views.main_api import MainView

api_routes = [
    Route.get("/", MainView),
    Route.group("/api", routes=[
        Route.post("/todos", view=CreateTaskAPI),
        Route.get("/todos", view=ListTaskAPI),
    ]),
]

Let's re run the test

$ bin/test tests/features/task/list_task_api_test.py

It should passed

.
----------------------------------------------------------------------
Ran 1 test in 0.031s

OK

Re-run all test to ensure our new feature not breaking existing feature

$ bin/test discover -s tests -p '*_test.py'

It should pass all tests

.....
----------------------------------------------------------------------
Ran 5 tests in 0.084s

OK

Your final project structure should be look like this

flask-todo
├── app
│   ├── config.py
│   ├── factory.py
│   ├── http
│   │   ├── __init__.py
│   │   └── route.py
│   ├── __init__.py
│   ├── routes
│   │   ├── api.py
│   │   └── __init__.py
│   ├── tasks
│   │   ├── __init__.py
│   │   ├── models
│   │   │   ├── __init__.py
│   │   │   └── task.py
│   │   ├── schemas
│   │   │   ├── __init__.py
│   │   │   └── task_schema.py
│   │   └── views
│   │       ├── create_task_api.py
│   │       ├── __init__.py
│   │       └── list_task_api.py
│   └── views
│       ├── __init__.py
│       └── main_api.py
├── bin
│   └── test
├── db
│   ├── create.sql
│   └── Dockerfile
├── docker-compose.yml
├── Dockerfile
├── .dockerignore
├── .env.dev
├── .gitignore
├── migrations
│   ├── alembic.ini
│   ├── env.py
│   ├── README
│   ├── script.py.mako
│   └── versions
│       └── 213adaf34061_create_tasks_tables.py
├── requirements.txt
├── serve.py
└── tests
    ├── base.py
    ├── features
    │   ├── __init__.py
    │   └── task
    │       ├── create_task_api_test.py
    │       ├── __init__.py
    │       └── list_task_api_test.py
    ├── __init__.py
    └── unit
        ├── __init__.py
        └── task
            ├── __init__.py
            └── task_model_test.py

17 directories, 41 files

We can commit our works before continuing to next chapter.

(venv)$ git add .
(venv)$ git commit -m "Create ListTaskAPI endpoint"

Detail ToDo API

In this chapter we will create endpoint /api/todos/{:id} to get detail about specific Task

Write the Test

Write test at tests/features/task/detail_task_api_test.py

from http import HTTPStatus
from tests.base import BaseAPITestCase
from app.tasks.models.task import Task
from app.factory import db


class DetailTaskAPITest(BaseAPITestCase):
    def test_detail_task(self):
        endpoint = "/api/todos"
        task = Task(name="Buy groceries")
        db.session.add(task)
        db.session.commit()

        db.session.refresh(task)

        resp = self.client.get(f"{endpoint}/{task.id}")
        self.assertEqual(resp.status_code, HTTPStatus.OK)

        resp_json = resp.json or dict()
        data = resp_json["data"]
        self.assertEqual(data["name"], "Buy groceries")

        Task.query.delete()
        db.session.commit()

It will raise error HTTP_NOT_FOUND (404), since we not implement our code

F
======================================================================
FAIL: test_create_task (tests.features.task.detail_task_api_test.CreateTaskAPITest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/src/app/tests/features/task/detail_task_api_test.py", line 17, in test_create_task
    self.assertEqual(resp.status_code, HTTPStatus.OK)
AssertionError: 404 != <HTTPStatus.OK: 200>

----------------------------------------------------------------------
Ran 1 test in 0.029s

FAILED (failures=1)

Let's implement our code at app/tasks/views/detail_task_api.py

from http import HTTPStatus

from flask import jsonify
from flask.typing import ResponseReturnValue
from flask.views import View

from app.tasks.models.task import Task
from app.tasks.schemas.task_schema import TaskSchema


class DetailTaskAPI(View):
    def dispatch_request(self, task_id) -> ResponseReturnValue:
        tasks_schema = TaskSchema()
        task = Task.query.filter(
            Task.id == task_id
        ).first()

        resp = {
            "data": tasks_schema.dump(task)
        }
        return jsonify(resp), HTTPStatus.OK

And update our api routes to

from app.http.route import Route
from app.tasks.views.detail_task_api import DetailTaskAPI
from app.tasks.views.list_task_api import ListTaskAPI
from app.tasks.views.create_task_api import CreateTaskAPI
from app.views.main_api import MainView

api_routes = [
    Route.get("/", MainView),
    Route.group("/api", routes=[
        Route.post("/todos", view=CreateTaskAPI),
        Route.get("/todos", view=ListTaskAPI),
        Route.get("/todos/<task_id>", view=DetailTaskAPI),
    ]),
]

Re-run our test file

$ bin/test tests/features/task/detail_task_api_test.py

It should pass

.
----------------------------------------------------------------------
Ran 1 test in 0.033s

OK

Let's add test to cover when id passed not available

    ...
    def test_detail_task_when_id_not_exists(self):
        endpoint = "/api/todos"
        resp = self.client.get(f"{endpoint}/123")
        self.assertEqual(resp.status_code, HTTPStatus.NOT_FOUND)

Re-run our test file

$ bin/test tests/features/task/detail_task_api_test.py

It should raise error

.F
======================================================================
FAIL: test_detail_task_when_id_not_exists (tests.features.task.detail_task_api_test.DetailTaskAPITest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/src/app/tests/features/task/detail_task_api_test.py", line 29, in test_detail_task_when_id_not_exists
    self.assertEqual(resp.status_code, HTTPStatus.NOT_FOUND)
AssertionError: 200 != <HTTPStatus.NOT_FOUND: 404>

----------------------------------------------------------------------
Ran 2 tests in 0.046s

FAILED (failures=1)

Let's change our view to

from http import HTTPStatus

from flask import jsonify
from flask.typing import ResponseReturnValue
from flask.views import View

from app.tasks.models.task import Task
from app.tasks.schemas.task_schema import TaskSchema


class DetailTaskAPI(View):
    def dispatch_request(self, task_id) -> ResponseReturnValue:
        tasks_schema = TaskSchema()
        task = Task.query.filter(
            Task.id == task_id
        ).first_or_404()

        resp = {
            "data": tasks_schema.dump(task)
        }
        return jsonify(resp), HTTPStatus.OK

Re-run our test file

$ bin/test tests/features/task/detail_task_api_test.py

It should be pass

..
----------------------------------------------------------------------
Ran 2 tests in 0.047s

OK

Let's re-run all of our tests

$ bin/test discover -s tests -p '*_test.py'

It should pass

.......
----------------------------------------------------------------------
Ran 7 tests in 0.123s

OK

Your final project structure should be look like this

flask-todo
├── app
│   ├── config.py
│   ├── factory.py
│   ├── http
│   │   ├── __init__.py
│   │   └── route.py
│   ├── __init__.py
│   ├── routes
│   │   ├── api.py
│   │   └── __init__.py
│   ├── tasks
│   │   ├── __init__.py
│   │   ├── models
│   │   │   ├── __init__.py
│   │   │   └── task.py
│   │   ├── schemas
│   │   │   ├── __init__.py
│   │   │   └── task_schema.py
│   │   └── views
│   │       ├── create_task_api.py
│   │       ├── detail_task_api.py
│   │       ├── __init__.py
│   │       └── list_task_api.py
│   └── views
│       ├── __init__.py
│       └── main_api.py
├── bin
│   └── test
├── db
│   ├── create.sql
│   └── Dockerfile
├── docker-compose.yml
├── Dockerfile
├── .dockerignore
├── .env.dev
├── .gitignore
├── migrations
│   ├── alembic.ini
│   ├── env.py
│   ├── README
│   ├── script.py.mako
│   └── versions
│       └── 213adaf34061_create_tasks_tables.py
├── requirements.txt
├── serve.py
└── tests
    ├── base.py
    ├── features
    │   ├── __init__.py
    │   └── task
    │       ├── create_task_api_test.py
    │       ├── detail_task_api_test.py
    │       ├── __init__.py
    │       └── list_task_api_test.py
    ├── __init__.py
    └── unit
        ├── __init__.py
        └── task
            ├── __init__.py
            └── task_model_test.py

17 directories, 43 files

We can commit our works before continuing to next chapter.

(venv)$ git add .
(venv)$ git commit -m "Create DetailTaskAPI endpoint"

Update ToDo API

In this chapter we will create endpoint /api/todos/{:id} to update our Task model

Write the Test

Write test at tests/features/task/update_task_api_test.py

from http import HTTPStatus
from tests.base import BaseAPITestCase
from app.tasks.models.task import Task
from app.factory import db


class UpdateTaskAPITest(BaseAPITestCase):
    def test_update_task(self):
        endpoint = "/api/todos"
        task = Task(name="Buy groceries")
        db.session.add(task)
        db.session.commit()

        db.session.refresh(task)

        json_data = {
            "name": "Buy milk",
            "completed": True,
        }
        resp = self.client.post(
            f"{endpoint}/{task.id}",
            json=json_data
        )
        self.assertEqual(resp.status_code, HTTPStatus.OK)

        resp_json = resp.json or dict()
        data = resp_json["data"]
        self.assertEqual(data["name"], "Buy milk")
        self.assertEqual(data["completed"], True)

        Task.query.delete()
        db.session.commit()

Run our test

$ bin/test tests/features/task/update_task_api_test.py

It will error

F
======================================================================
FAIL: test_update_task (tests.features.task.update_task_api_test.UpdateTaskAPITest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/src/app/tests/features/task/update_task_api_test.py", line 24, in test_update_task
    self.assertEqual(resp.status_code, HTTPStatus.OK)
AssertionError: 405 != <HTTPStatus.OK: 200>

----------------------------------------------------------------------
Ran 1 test in 0.032s

FAILED (failures=1)

Let's implement our view at app/tasks/views/update_task_api.py

from http import HTTPStatus

from flask import jsonify, make_response, request
from flask.typing import ResponseReturnValue
from flask.views import View

from app.factory import db
from app.tasks.models.task import Task
from app.tasks.schemas.task_schema import TaskSchema


class UpdateTaskAPI(View):
    def dispatch_request(self, task_id) -> ResponseReturnValue:
        task_schema = TaskSchema()
        json_data = request.get_json() or dict()

        task = Task.query.filter(
            Task.id == task_id
        ).first()

        task.query.update(json_data)
        db.session.commit()

        resp = {
            "data": task_schema.dump(task)
        }
        return jsonify(resp), HTTPStatus.OK

Update our routes to

from app.http.route import Route
from app.tasks.views.detail_task_api import DetailTaskAPI
from app.tasks.views.list_task_api import ListTaskAPI
from app.tasks.views.create_task_api import CreateTaskAPI
from app.tasks.views.update_task_api import UpdateTaskAPI
from app.views.main_api import MainView

api_routes = [
    Route.get("/", MainView),
    Route.group("/api", routes=[
        Route.post("/todos", view=CreateTaskAPI),
        Route.get("/todos", view=ListTaskAPI),
        Route.get("/todos/<task_id>", view=DetailTaskAPI),
        Route.put("/todos/<task_id>", view=UpdateTaskAPI),
    ]),
]

Re-run the test

$ bin/test tests/features/task/update_task_api_test.py

It should pass

.
----------------------------------------------------------------------
Ran 1 test in 0.041s

OK

Let's add test when updated Task does not exists

    ...
    def test_update_task_when_id_not_exists(self):
    endpoint = "/api/todos"
    json_data = {
        "name": "Buy milk tea",
    }
    resp = self.client.put(f"{endpoint}/123", json=json_data)
    self.assertEqual(resp.status_code, HTTPStatus.NOT_FOUND)

Re-run the test

$ bin/test tests/features/task/update_task_api_test.py

It should fail

.E
======================================================================
ERROR: test_update_task_when_id_not_exists (tests.features.task.update_task_api_test.UpdateTaskAPITest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/src/app/tests/features/task/update_task_api_test.py", line 39, in test_update_task_when_id_not_exists
    resp = self.client.put(f"{endpoint}/123", json=json_data)
  File "/usr/local/lib/python3.8/site-packages/werkzeug/test.py", line 1150, in put
    return self.open(*args, **kw)
  File "/usr/local/lib/python3.8/site-packages/flask/testing.py", line 223, in open
    response = super().open(
  File "/usr/local/lib/python3.8/site-packages/werkzeug/test.py", line 1094, in open
    response = self.run_wsgi_app(request.environ, buffered=buffered)
  File "/usr/local/lib/python3.8/site-packages/werkzeug/test.py", line 961, in run_wsgi_app
    rv = run_wsgi_app(self.application, environ, buffered=buffered)
  File "/usr/local/lib/python3.8/site-packages/werkzeug/test.py", line 1242, in run_wsgi_app
    app_rv = app(environ, start_response)
  File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 2548, in __call__
    return self.wsgi_app(environ, start_response)
  File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 2528, in wsgi_app
    response = self.handle_exception(e)
  File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 2525, in wsgi_app
    response = self.full_dispatch_request()
  File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 1822, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 1820, in full_dispatch_request
    rv = self.dispatch_request()
  File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 1796, in dispatch_request
    return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)
  File "/usr/local/lib/python3.8/site-packages/flask/views.py", line 107, in view
    return current_app.ensure_sync(self.dispatch_request)(**kwargs)
  File "/usr/src/app/app/tasks/views/update_task_api.py", line 28, in dispatch_request
    task.query.update(json_data)
AttributeError: 'NoneType' object has no attribute 'query'

----------------------------------------------------------------------
Ran 2 tests in 0.057s

FAILED (errors=1)

Fix our implementation to

from http import HTTPStatus

from flask import jsonify, make_response, request
from flask.typing import ResponseReturnValue
from flask.views import View

from app.factory import db
from app.tasks.models.task import Task
from app.tasks.schemas.task_schema import TaskSchema


class UpdateTaskAPI(View):
    def dispatch_request(self, task_id) -> ResponseReturnValue:
        task_schema = TaskSchema()
        json_data = request.get_json() or dict()

        task = Task.query.filter(
            Task.id == task_id
        ).first_or_404()

        task.query.update(json_data)
        db.session.commit()

        resp = {
            "data": task_schema.dump(task)
        }
        return jsonify(resp), HTTPStatus.OK

Re-run the test

$ bin/test tests/features/task/update_task_api_test.py

It should pass

..
----------------------------------------------------------------------
Ran 2 tests in 0.056s

OK

Add more test to validate json schema

    ...
    def test_update_task_when_required_payload_empty(self):
        endpoint = "/api/todos"
        task = Task(name="Buy groceries")
        db.session.add(task)
        db.session.commit()

        db.session.refresh(task)

        json_data = {}
        resp = self.client.put(
            f"{endpoint}/{task.id}",
            json=json_data
        )
        self.assertEqual(resp.status_code, HTTPStatus.BAD_REQUEST)

        Task.query.delete()
        db.session.commit()

Re-run the test

$ bin/test tests/features/task/update_task_api_test.py

It should fail

..F
======================================================================
FAIL: test_update_task_when_required_payload_empty (tests.features.task.update_task_api_test.UpdateTaskAPITest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/src/app/tests/features/task/update_task_api_test.py", line 55, in test_update_task_when_required_payload_empty
    self.assertEqual(resp.status_code, HTTPStatus.BAD_REQUEST)
AssertionError: 200 != <HTTPStatus.BAD_REQUEST: 400>

----------------------------------------------------------------------
Ran 3 tests in 0.074s

FAILED (failures=1)

Update our implementation to

from http import HTTPStatus

from flask import jsonify, make_response, request
from flask.typing import ResponseReturnValue
from flask.views import View

from app.factory import db
from app.tasks.models.task import Task
from app.tasks.schemas.task_schema import TaskSchema


class UpdateTaskAPI(View):
    def dispatch_request(self, task_id) -> ResponseReturnValue:
        task_schema = TaskSchema()

        json_data = request.get_json() or dict()
        errors = task_schema.validate(json_data)
        if errors:
            return make_response(
                jsonify(errors=str(errors)),
                HTTPStatus.BAD_REQUEST
            )

        task = Task.query.filter(
            Task.id == task_id
        ).first_or_404()

        task.query.update(json_data)
        db.session.commit()

        resp = {
            "data": task_schema.dump(task)
        }
        return jsonify(resp), HTTPStatus.OK

Re-run the test

$ bin/test tests/features/task/update_task_api_test.py

It should pass

...
----------------------------------------------------------------------
Ran 3 tests in 0.072s

OK

Let's re-run all of our tests

$ bin/test discover -s tests -p '*_test.py'

It should pass

..........
----------------------------------------------------------------------
Ran 10 tests in 0.186s

OK

Your final project structure should be look like this

flask-todo
├── app
│   ├── config.py
│   ├── factory.py
│   ├── http
│   │   ├── __init__.py
│   │   └── route.py
│   ├── __init__.py
│   ├── routes
│   │   ├── api.py
│   │   └── __init__.py
│   ├── tasks
│   │   ├── __init__.py
│   │   ├── models
│   │   │   ├── __init__.py
│   │   │   └── task.py
│   │   ├── schemas
│   │   │   ├── __init__.py
│   │   │   └── task_schema.py
│   │   └── views
│   │       ├── create_task_api.py
│   │       ├── detail_task_api.py
│   │       ├── __init__.py
│   │       ├── list_task_api.py
│   │       └── update_task_api.py
│   └── views
│       ├── __init__.py
│       └── main_api.py
├── bin
│   └── test
├── db
│   ├── create.sql
│   └── Dockerfile
├── docker-compose.yml
├── Dockerfile
├── .dockerignore
├── .env.dev
├── .gitignore
├── migrations
│   ├── alembic.ini
│   ├── env.py
│   ├── README
│   ├── script.py.mako
│   └── versions
│       └── 213adaf34061_create_tasks_tables.py
├── requirements.txt
├── serve.py
└── tests
    ├── base.py
    ├── features
    │   ├── __init__.py
    │   └── task
    │       ├── create_task_api_test.py
    │       ├── detail_task_api_test.py
    │       ├── __init__.py
    │       ├── list_task_api_test.py
    │       └── update_task_api_test.py
    ├── __init__.py
    └── unit
        ├── __init__.py
        └── task
            ├── __init__.py
            └── task_model_test.py

17 directories, 45 files

We can commit our works before continuing to next chapter.

(venv)$ git add .
(venv)$ git commit -m "Create UpdateTaskAPI endpoint"

Delete ToDo API

In this chapter we will create endpoint /api/todos/{:id} to delete specific Task

Write the Test

Write our test at tests/features/task/delete_task_api_test.py

from http import HTTPStatus
from tests.base import BaseAPITestCase
from app.tasks.models.task import Task
from app.factory import db


class DeleteTaskAPITest(BaseAPITestCase):
    def test_delete_task(self):
        endpoint = "/api/todos"
        task = Task(name="Buy groceries")
        db.session.add(task)
        db.session.commit()

        db.session.refresh(task)

        resp = self.client.delete(f"{endpoint}/{task.id}")
        self.assertEqual(resp.status_code, HTTPStatus.NO_CONTENT)

        tasks = Task.query.all()
        self.assertEqual(len(tasks), 0)

        Task.query.delete()
        db.session.commit()

Run the test

$ bin/test tests/features/task/delete_task_api_test.py

It should fail

F
======================================================================
FAIL: test_delete_task (tests.features.task.delete_task_api_test.DeleteTaskAPITest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/src/app/tests/features/task/delete_task_api_test.py", line 17, in test_delete_task
    self.assertEqual(resp.status_code, HTTPStatus.NO_CONTENT)
AssertionError: 405 != <HTTPStatus.NO_CONTENT: 204>

----------------------------------------------------------------------
Ran 1 test in 0.033s

FAILED (failures=1)

Create our view at app/tasks/views/delete_task_api.py

from http import HTTPStatus

from flask.typing import ResponseReturnValue
from flask.views import View

from app.factory import db
from app.tasks.models.task import Task


class DeleteTaskAPI(View):
    def dispatch_request(self, task_id) -> ResponseReturnValue:
        task = Task.query.filter(
            Task.id == task_id
        ).first()

        db.session.delete(task)
        db.session.commit()
        return "", HTTPStatus.NO_CONTENT

Update our routes to

from app.http.route import Route
from app.tasks.views.delete_task_api import DeleteTaskAPI
from app.tasks.views.detail_task_api import DetailTaskAPI
from app.tasks.views.list_task_api import ListTaskAPI
from app.tasks.views.create_task_api import CreateTaskAPI
from app.tasks.views.update_task_api import UpdateTaskAPI
from app.views.main_api import MainView

api_routes = [
    Route.get("/", MainView),
    Route.group("/api", routes=[
        Route.post("/todos", view=CreateTaskAPI),
        Route.get("/todos", view=ListTaskAPI),
        Route.get("/todos/<task_id>", view=DetailTaskAPI),
        Route.put("/todos/<task_id>", view=UpdateTaskAPI),
        Route.delete("/todos/<task_id>", view=DeleteTaskAPI),
    ]),
]

Re-run the test

$ bin/test tests/features/task/delete_task_api_test.py

It should pass

.
----------------------------------------------------------------------
Ran 1 test in 0.038s

OK

Let's add test when target id does not exists

    ...
    def test_delete_task_when_id_not_exists(self):
        endpoint = "/api/todos"
        resp = self.client.delete(f"{endpoint}/123")
        self.assertEqual(resp.status_code, HTTPStatus.NOT_FOUND)

Re-run the test

$ bin/test tests/features/task/delete_task_api_test.py

It should fail

.E
======================================================================
ERROR: test_delete_task_when_id_not_exists (tests.features.task.delete_task_api_test.DeleteTaskAPITest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 2700, in delete
    state = attributes.instance_state(instance)
AttributeError: 'NoneType' object has no attribute '_sa_instance_state'

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/src/app/tests/features/task/delete_task_api_test.py", line 27, in test_delete_task_when_id_not_exists
    resp = self.client.delete(f"{endpoint}/123")
  File "/usr/local/lib/python3.8/site-packages/werkzeug/test.py", line 1155, in delete
    return self.open(*args, **kw)
  File "/usr/local/lib/python3.8/site-packages/flask/testing.py", line 223, in open
    response = super().open(
  File "/usr/local/lib/python3.8/site-packages/werkzeug/test.py", line 1094, in open
    response = self.run_wsgi_app(request.environ, buffered=buffered)
  File "/usr/local/lib/python3.8/site-packages/werkzeug/test.py", line 961, in run_wsgi_app
    rv = run_wsgi_app(self.application, environ, buffered=buffered)
  File "/usr/local/lib/python3.8/site-packages/werkzeug/test.py", line 1242, in run_wsgi_app
    app_rv = app(environ, start_response)
  File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 2548, in __call__
    return self.wsgi_app(environ, start_response)
  File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 2528, in wsgi_app
    response = self.handle_exception(e)
  File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 2525, in wsgi_app
    response = self.full_dispatch_request()
  File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 1822, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 1820, in full_dispatch_request
    rv = self.dispatch_request()
  File "/usr/local/lib/python3.8/site-packages/flask/app.py", line 1796, in dispatch_request
    return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)
  File "/usr/local/lib/python3.8/site-packages/flask/views.py", line 107, in view
    return current_app.ensure_sync(self.dispatch_request)(**kwargs)
  File "/usr/src/app/app/tasks/views/delete_task_api.py", line 16, in dispatch_request
    db.session.delete(task)
  File "<string>", line 2, in delete
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 2702, in delete
    util.raise_(
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
sqlalchemy.orm.exc.UnmappedInstanceError: Class 'builtins.NoneType' is not mapped

----------------------------------------------------------------------
Ran 2 tests in 0.056s

FAILED (errors=1)

Update our implementation to

from http import HTTPStatus

from flask.typing import ResponseReturnValue
from flask.views import View

from app.factory import db
from app.tasks.models.task import Task


class DeleteTaskAPI(View):
    def dispatch_request(self, task_id) -> ResponseReturnValue:
        task = Task.query.filter(
            Task.id == task_id
        ).first_or_404()

        db.session.delete(task)
        db.session.commit()
        return "", HTTPStatus.NO_CONTENT

Re-run the test

$ bin/test tests/features/task/delete_task_api_test.py

It should pass

..
----------------------------------------------------------------------
Ran 2 tests in 0.057s

OK

Re-run all test to ensure our new feature not breaking existing feature

$ bin/test discover -s tests -p '*_test.py'

It should pass all tests

............
----------------------------------------------------------------------
Ran 12 tests in 0.233s

OK

Your final project structure should be look like this

flask-todo
├── app
│   ├── config.py
│   ├── factory.py
│   ├── http
│   │   ├── __init__.py
│   │   └── route.py
│   ├── __init__.py
│   ├── routes
│   │   ├── api.py
│   │   └── __init__.py
│   ├── tasks
│   │   ├── __init__.py
│   │   ├── models
│   │   │   ├── __init__.py
│   │   │   └── task.py
│   │   ├── schemas
│   │   │   ├── __init__.py
│   │   │   └── task_schema.py
│   │   └── views
│   │       ├── create_task_api.py
│   │       ├── delete_task_api.py
│   │       ├── detail_task_api.py
│   │       ├── __init__.py
│   │       ├── list_task_api.py
│   │       └── update_task_api.py
│   └── views
│       ├── __init__.py
│       └── main_api.py
├── bin
│   └── test
├── db
│   ├── create.sql
│   └── Dockerfile
├── docker-compose.yml
├── Dockerfile
├── .dockerignore
├── .env.dev
├── .gitignore
├── migrations
│   ├── alembic.ini
│   ├── env.py
│   ├── README
│   ├── script.py.mako
│   └── versions
│       └── 213adaf34061_create_tasks_tables.py
├── requirements.txt
├── serve.py
└── tests
    ├── base.py
    ├── features
    │   ├── __init__.py
    │   └── task
    │       ├── create_task_api_test.py
    │       ├── delete_task_api_test.py
    │       ├── detail_task_api_test.py
    │       ├── __init__.py
    │       ├── list_task_api_test.py
    │       └── update_task_api_test.py
    ├── __init__.py
    └── unit
        ├── __init__.py
        └── task
            ├── __init__.py
            └── task_model_test.py

17 directories, 47 files

We can commit our works before continuing to next chapter.

(venv)$ git add .
(venv)$ git commit -m "Create DeleteTaskAPI endpoint"

Review

After completing Chapter 2 there are few things that we can review

Flask Class-based Views

In this course, we are utilizing Flask's Class-based Views instead of the route decorator.

Flask-SQLAlchemy

Instead defining our own utilization, we utilize Flask-SQLAlchemy to handle our connection and session to database.

Flask-Migrate

To have consistent database structure, we utilize Flask-Migrate to handle our model and migration.

About

Author

Ngalim Siregar is a software engineer who lives and works in Indonesia. He enjoys software development, and loves playing retro games.

• Twitter