Skip to content

Latest commit

 

History

History
304 lines (216 loc) · 11.6 KB

02-crud-api.md

File metadata and controls

304 lines (216 loc) · 11.6 KB

02 - A basic CRUD API with TDD

Code for this chapter available here.

With our fresh environment we will start to get hands on with API Star. In this section we will construct a simple CRUD (Create, Read, Update and Delete) api to manage a TODO list.

We will put special enfasis on testing, using py.test testing framework, which is included in API Star. We will follow some principles of TDD (Test Driven Development).

Just another TODO list API

The TODO List is the 'hello world' of the XXI century. In this basic example, a user must be able to create or delete a Task, as well as mark it as completed. It should also list every Task, and maybe allow to filter by a completed query string param.

The endpoints of our API that will be developed in this chapter are:

  • GET /task/: Retrieve a list of tasks.
  • POST /task/: Create a new task.
  • PATCH /task/{id}/: Update some field of an specific task by id (i.e. mark as completed)
  • DELETE /task/{id}/: Delete a task by id.

Testing a dumb view

We will start using the code structure generated in Chapter 1, adding the GET and POST verbs over the /task/ endpoint to allow to create new tasks and list them. To temporarily persist our list of tasks, we will use an in memory global variable. In next chapters we will substitute this using SQLAlchemy as database backend.

Furthermore we will create a Task schema. API Star provides a typing system to define your API interface. Schemas are defined as specification of input/output data contracts for your views, validating your input and serializing your output (as libraries such as marshmallow do).

Let's start adding some dumb views in our project/views.py file:

    def list_tasks():
        return {}

    def create_task():
        return {}

And add their routes in routes.py taking advantatge of Include to isolate the task related routes:

    [...]
    from project.views import list_tasks, create_task

    task_routes = [
        Route('/', 'GET', list_tasks),
        Route('/', 'POST', create_task),
    ]

    routes = [
        Route('/', 'GET', welcome),
        Include('/task', task_routes),
        Include('/docs', docs_routes),
        Include('/static', static_routes)
    ]

Now we have just created our first dumb connected views. But wait, were is the Red-Green-Refactor?

Some tests are generated by apistar in the example of chapter 1, so we can run them. Add this service to the docker-compose.yml, and we can move some of the parameters from the api service to the base:

version: '2'

services:
  base:
    build: .
    entrypoint: apistar
    working_dir: /app
    volumes:
      - "./api:/app"

  api:
    extends: base
    command: ["run", "--host", "0.0.0.0", "--port", "80"]
    ports:
      - "8080:80"

  test:
    extends: base
    command: ["test"]

And simply run the test suite with:

$ docker-compose run test
  ========================================================================= test session starts =========================================================================
  platform linux -- Python 3.6.1, pytest-3.1.2, py-1.4.34, pluggy-0.4.0
  rootdir: /app, inifile:
  collected 2 items

  tests/test_app.py ..

  ====================================================================== 2 passed in 0.02 seconds =======================================================================

py.test

Add two simple tests at tests/test_app.py that call the views and assert that they return an empty dictionary.

from project.views import list_tasks, add_task

def test_add_task():
    assert add_task() == {}

def test_list_tasks():
    assert list_tasks() == {}

Run again the tests and.. nice! We are green! Let's start implementing the views. List tasks initially should return an empty list.

# test_app.py
def test_list_tasks():
    assert list_tasks() == []

# views.py
tasks = []

def list_tasks():
    """ Return a list of tasks """
    return tasks

I will present both the test and the modified view. The idea behind is to drive our development thinking about the expected behaviour, by writting the test before the code.

TestClient

API Star also provides a test client wrapping requests to test your app. We have tested the views directly, but we could also add a few http tests using this client.

from apistar.test import TestClient

client = TestClient()

def test_http_list_tasks():
    response = client.get('/tasks/')
    assert response.status_code == 200
    assert response.json() == []

def test_http_add_task():
    response = client.post('/tasks/')
    assert response.status_code == 200
    assert response.json() == {}

In our case, we will define a pytest fixture at directory level to provide our client as parameters of the tests (see /tests/conftest.py).

Schemas

To define the interface of the views, create an schema of our Task object and a simple TaskDefinition, constraining the max length of the string on input.

# schemas.py
from apistar import schema

class TaskDefinition(schema.String):
    max_length = 128

class Task(schema.Object):
    properties = {
        'definition': TaskDefinition,
        'completed': schema.Boolean(default=False),
    }

This schemas allow us to validate the input of the create task, and serialize the output of the both views. API Star allows us to annotate the route handlers with the expected input and output, it's like magic :-)

# views.py
from typing import List

from project.schemas import Task, TaskDefinition

def list_task() -> List[Task]:
    return [Task(t) for t in tasks]

def add_task(definition: TaskDefinition) -> Task:
    new_task = Task({'definition': definition})
    task.append(new_task)
    return new_task

# test_task.py
from apistar.test import TestClient

task_endpoint = '/task/'
client = TestClient()

new_task = {'definition': 'test task'}
added_task = {'definition': 'test task', 'completed': False}

def test_list_tasks():
    response = client.get(task_endpoint)
    assert response.status_code == 200
    assert response.json() == []

def test_add_task():
    response = client.post(task_endpoint, new_task)
    assert response.status_code == 200

    assert response.json() == added_task

def test_list_an_added_task():
    response = client.get(task_endpoint)
    assert response.status_code == 200
    assert response.json() == [added_task]

    test_add_task()
    response = client.get(task_endpoint)
    assert response.status_code == 200
    assert response.json() == [added_task, added_task]

Note: We have extracted the tests related with the /task/ API endpoint in tests/test_task.py.

Take a look to the views annotated with the Schemas previously defined. list_task returns a list of serialized task items, and add_task gets input via type annotation, with a validated definition.

Until now we have just tested the happy path, and that's just naive! But we will let that as an exercise to the reader. Fork the project and add some tests.

Delete a task or mark it as completed

Let's add the pending functionality: mark a todo task as completed and delete a task.

Until now our schema does not include an unique id value for tasks, so we cannot identify a concrete task (because different tasks could have the same definition and completion state). Also the data structure that we have used to persist the tasks in memory, a list, it's not specially suited for indexing by key, deleting tasks, etc.

The first thing that we will do is write some tests for the new autoincremental id of the tasks, then change the schema (adding an 'id': schema.Integer(default=None) in the properties) and use a generator to implement an autoincremental counter for our tasks ids, before starting to write the new views.

# views.py
import itertools
counter = itertools.count(1).__next__

def add_task(definition: TaskDefinition) -> Task:
    """
    Add a new task. It receives its definition as an argument
    and sets an autoincremental id in the Task constructor.
    """
    new_task = Task({'id': counter(), 'definition': definition})
    tasks.append(new_task)
    return new_task

A lot of people argue that APIs should not expose internal indexes to the public, and some propose to use uids or other mechanisms to expose ours objects to the world. They are right. But this is a tutorial from scratch, let's go step by step!

And voilà, now the tasks have an id and are uniquely identifiable. See the tests for the simple case of adding two times the same task.

Another refactor that will become extremly handy in the implementation of our views is replacing our list of tasks by a dictionary of tasks, with their brand new ids as key. Obviously updating and deleting operations would benefit from the direct access of a dictionary, both in code simplicity and computational complexity.

# views.py
tasks = {}

def list_tasks() -> List[Task]:
    """ Return a list of tasks """
    return [Task(tasks[id]) for id in tasks]

def add_task(definition: TaskDefinition) -> Task:
    """
    Add a new task. It receives its definition as an argument
    and sets an autoincremental id in the Task constructor.
    """
    id = counter()
    tasks[id] = Task({
        'id': id,
        'definition': definition,
        'completed' = False,
    })
    return tasks[id]

The tests should still be green. But now, with this small changes, we should be capable of easily implement the delete and patch views.

In the routes file we will be using the curly braces syntax for the url id parameter, to pass it to the delete_task method. API Start by default returns plain text and a 200 status code. But when a delete action is successfull a 204 should be returned, and a 404 if the specified task does not exist. Responses allows us to customize our responses.

# routes.py
from project.views import list_tasks, add_task, delete_task

task_routes = [
    Route('/', 'GET', list_tasks),
    Route('/', 'POST', add_task),
    Route('/{task_id}/', 'DELETE', delete_task),
]

# views.py
from apistar.http import Response

def delete_task(task_id: int) -> Response:
    if task_id not in tasks:
        return Response({}, status=404)

    del tasks[task_id]
    return Response({}, status=204)

Note: I have also rewritten the view add_task with this recently introduced Response. A POST to an API endpoint that is successfull should return a 201 status code and the created object.

And now as an exercise, why you don't write your own method and routing to mark a task as completed? You can see our simple proposal in the final source code of this chapter, method patch_task.

Note that in this implementation, we are using a PATCH method with a task_id parameter from the URL and a completed boolean from the body of the request. As the current README of apistar reads:

Parameters which do not correspond to a URL path parameter will be treated as query parameters for GET and DELETE requests, or part of the request body for POST, PUT, and PATCH requests.

  • TODO appendix for the filtering of completed/uncompleted tasks via query string

Next section: 03 - Database backend

Back to the table of contents.