3
Building the API
In the previous chapter, we built a backend that connects to the database, manages user sessions, and sends emails. Now, we will add a specific API to the backend that tracks the member’s to-do’s. This will require an API that allows members, sessions, and to-dos to be managed.
In this chapter, you’ll learn how to build a RESTful API, which is a very popular style of API and one you’ll likely use and come across in your career. You’ll also build an API to manage members and authenticate their actions, which could be used in any other app with minimal changes. Finally, we will also build an API to track the to-dos, which, again, could be adapted for other uses.
We’ll build the API using a RESTful style as it works very well with web apps and can be expressed very easily with Quart. A RESTful API is where the functionality is grouped by resource with each function being an action acting on the resource. For example, the functionality to log in is described as creating a session and log out as deleting a session. For a RESTful web app, the action is represented by the HTTP verb and the resource by the HTTP path. In addition, the response status code is used to indicate the effect of the functionality, with 2XX codes indicating success and 4XX codes indicating different types of errors.
Alternatives to RESTful APIs
While RESTful APIs utilize many HTTP verbs and paths to describe the functionality, a more basic style is to have a singular POST route. This route is then used for all the functionality with the request body describing the function and data. A good example is GraphQL, which typically uses only POST /graphql with a defined message structure. If you’d prefer to use GraphQL take a look at https://strawberry.rocks.
So, in this chapter, we will cover the following topics:
- Creating the database schema and models
- Building the session API
- Building the member API
- Building the to-do API
Technical requirements
The following additional folders are required in this chapter and should be created:
tozo └── backend ├── src │ └── backend │ ├── migrations │ └── models └── tests └── models
Empty backend/src/backend/models/__init__.py and backend/tests/models/__init__.py files should also be created.
To follow the development in this chapter, use the companion repository, https://github.com/pgjones/tozo, and see the commits between the tags r1-ch3-start and r1-ch3-end.
Creating the database schema and models
In this book, we are building a to-do tracking app, which means we need to store data about the member and their to-dos. We will do so by placing the data into the database, which means we need to define the structure of the data. This structure is called the schema and describes the tables in the database.
While the schema describes the structure of the data in the database, we will use models in the backend and frontend. Each model is a class that represents a row in the database. For example, a table with only an ID could be represented by a class with a single id attribute.
ORMs
Schemas and models are often conflated as the same thing, especially when an Object Relational Model (ORM) is used. While using an ORM is simpler to begin with, I find it hides important details and makes development harder after a short while. This is why, in this book, the model and schema are related but different. This also means that we’ll write SQL queries for all interactions with the database.
We’ll start by defining the member data and to-do data as both models and schemas in a migration, before adding some initial test and development data.
Creating the member schema and model
We need to store information for each member so that we can associate their to-dos with them, via a foreign key reference. In addition, we need to store enough information so that the member can log in and prove who they are (authenticate themselves), which means we need to store their email and password hash. Finally, we’ll also store when their account was created and when they verified their email – the latter being important if we want to send them emails.
The schema for this data is given by the following SQL, which is given for reference and will be used in the Running the first migration section:
CREATE TABLE members ( id INT PRIMARY KEY GENERATED ALWAYS AS IDENTITY, created TIMESTAMP NOT NULL DEFAULT now(), email TEXT NOT NULL, email_verified TIMESTAMP, password_hash TEXT NOT NULL ); CREATE UNIQUE INDEX members_unique_email_idx on members (LOWER(email));
The unique index highlighted ensures that there is only one member account per email, with email casing being ignored.
SQL formatting
In Chapter 1 Setting Up Our System for Development, I mentioned the importance of code formatting and autoformatters. Sadly, I haven’t found an autoformatter that works for SQL embedded in Python code. However, I recommend you follow the style guide given at http://sqlstyle.guide/ as I will in this book.
We can represent the database table with a Python dataclass, which includes each column as an attribute with the relevant Python type. This is the model shown in the following code, which should be added to backend/src/backend/models/member.py:
from dataclasses import dataclass from datetime import datetime @dataclass class Member: id: int email: str password_hash: str created: datetime email_verified: datetime | None
In addition to the model, we can add the following functions to backend/src/backend/models/member.py in order to convert between the backend model and the SQL that reads from the database:
from quart_db import Connection
async def select_member_by_email(
db: Connection, email: str
) -> Member | None:
result = await db.fetch_one(
"""SELECT id, email, password_hash, created, email_verified
FROM members
WHERE LOWER(email) = LOWER(:email)""",
{"email": email},
)
return None if result is None else Member(**result)
async def select_member_by_id(
db: Connection, id: int
) -> Member | None:
result = await db.fetch_one(
"""SELECT id, email, password_hash, created, email_verified
FROM members
WHERE id = :id""",
{"id": id},
)
return None if result is None else Member(**result)These functions allow member information to be read from the database. The highlighted line ensures that emails are considered a match if the lowercased email matches.
Email case sensitivity
In our app, we store the email in the case given by the user, while comparing lowercased emails. This is the most user-friendly and secure solution, as emails can have a case-sensitive local part (before the @) but rarely do and must be case insensitive for the domain part (after the @). Therefore, by storing the given casing we ensure the email is delivered while ensuring there is one account per email address. More information is available at https://stackoverflow.com/questions/9807909/are-email-addresses-case-sensitive.
Next, we need to add functions that can alter the data in the database by adding the following to backend/src/models/member.py:
async def insert_member(
db: Connection, email: str, password_hash: str
) -> Member:
result = await db.fetch_one(
"""INSERT INTO members (email, password_hash)
VALUES (:email, :password_hash)
RETURNING id, email, password_hash, created,
email_verified""",
{"email": email, "password_hash": password_hash},
)
return Member(**result)
async def update_member_password(
db: Connection, id: int, password_hash: str
) -> None:
await db.execute(
"""UPDATE members
SET password_hash = :password_hash
WHERE id = :id""",
{"id": id, "password_hash": password_hash},
)
async def update_member_email_verified(
db: Connection, id: int
) -> None:
await db.execute(
"UPDATE members SET email_verified = now() WHERE id = :id",
{"id": id},
)These functions match the functionality we’ll shortly add to the API.
The case sensitivity is something we should test, by adding the following to backend/tests/models/test_member.py:
import pytest from asyncpg.exceptions import UniqueViolationError # type: ignore from quart_db import Connection from backend.models.member import insert_member async def test_insert_member(connection: Connection) -> None: await insert_member(connection, "[email protected]", "") with pytest.raises(UniqueViolationError): await insert_member(connection, "[email protected]", "")
Firstly, we want a test to ensure that insert_member correctly rejects a second member with an email that differs only by casing. The highlighted line ensures that the lines within, when executed, raise a UniqueViolationError and hence prevents the member from being inserted again.
We also need to test that the select_member_by_email function compares case insensitively by adding the following to backend/tests/models/test_member.py:
from backend.models.member import select_member_by_email async def test_select_member_by_email (connection: Connection) -> None: await insert_member(connection, "[email protected]", "") member = await select_member_by_email( connection, "[email protected]" ) assert member is not None
With the model code set up this way, we’ll be able to use these functions and the class instance directly wherever required in the backend code.
Creating the to-do schema and model
We also want to store information for each to-do, specifically the to-do task as text, when the to-do is due to be completed (although this should be optional), and if the to-do is complete. In addition, every to-do should be linked to the member that owns it.
The schema for this data is given by the following SQL, which is given for reference and will be used in the Running the first migration section:
CREATE TABLE todos ( id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY, complete BOOLEAN NOT NULL DEFAULT FALSE, due TIMESTAMPTZ, member_id INT NOT NULL REFERENCES members(id), task TEXT NOT NULL );
The corresponding backend model for this table is given by the following code, which should be added to backend/src/backend/models/todo.py:
from dataclasses import dataclass from datetime import datetime from pydantic import constr @dataclass class Todo: complete: bool due: datetime | None id: int task: constr(strip_whitespace=True, min_length=1) # type: ignore
Here, constr is used in place of str in order to ensure that empty strings are not considered valid. In addition to the model, we can add the following functions to backend/src/backend/models/todo.py in order to convert between the backend model and the SQL that reads from the database:
from quart_db import Connection
async def select_todos(
connection: Connection,
member_id: int,
complete: bool | None = None,
) -> list[Todo]:
if complete is None:
query = """SELECT id, complete, due, task
FROM todos
WHERE member_id = :member_id"""
values = {"member_id": member_id}
else:
query = """SELECT id, complete, due, task
FROM todos
WHERE member_id = :member_id
AND complete = :complete"""
values = {"member_id": member_id, "complete": complete}
return [
Todo(**row)
async for row in connection.iterate(query, values)
]
async def select_todo(
connection: Connection, id: int, member_id: int,
) -> Todo | None:
result = await connection.fetch_one(
"""SELECT id, complete, due, task
FROM todos
WHERE id = :id AND member_id = :member_id""",
{"id": id, "member_id": member_id},
)
return None if result is None else Todo(**result)These functions allow to-dos to be read from the database, but will only return to-dos that are owned by the given member_id. Using these functions should ensure that we don’t return to-dos to the wrong members.
Next, we need to add functions that can alter the data in the database by adding the following to backend/src/models/todo.py:
async def insert_todo(
connection: Connection,
member_id: int,
task: str,
complete: bool,
due: datetime | None,
) -> Todo:
result = await connection.fetch_one(
"""INSERT INTO todos (complete, due, member_id, task)
VALUES (:complete, :due, :member_id, :task)
RETURNING id, complete, due, task""",
{
"member_id": member_id,
"task": task,
"complete": complete,
"due": due,
},
)
return Todo(**result)
async def update_todo(
connection: Connection,
id: int,
member_id: int,
task: str,
complete: bool,
due: datetime | None,
) -> Todo | None:
result = await connection.fetch_one(
"""UPDATE todos
SET complete = :complete, due = :due,
task = :task
WHERE id = :id AND member_id = :member_id
RETURNING id, complete, due, task""",
{
"id": id,
"member_id": member_id,
"task": task,
"complete": complete,
"due": due,
},
)
return None if result is None else Todo(**result)
async def delete_todo(
connection: Connection, id: int, member_id: int,
) -> None:
await connection.execute(
"DELETE FROM todos WHERE id = :id AND member_id = :member_id",
{"id": id, "member_id": member_id},
)Note that all these functions also take a member_id argument and only affect the to-dos that belong to the given member_id. This will help us avoid authorization errors whereby we write code that mistakenly allows a member to access or modify another member’s to-do.
This is something we should test, by adding the following to backend/tests/models/test_todo.py. Firstly, we want a test to ensure that delete_todo correctly deletes the to-do:
import pytest from quart_db import Connection from backend.models.todo import ( delete_todo, insert_todo, select_todo, update_todo ) @pytest.mark.parametrize( "member_id, deleted", [(1, True), (2, False)], ) async def test_delete_todo( connection: Connection, member_id: int, deleted: bool ) -> None: todo = await insert_todo( connection, 1, "Task", False, None ) await delete_todo(connection, todo.id, member_id) new_todo = await select_todo(connection, todo.id, 1) assert (new_todo is None) is deleted
The highlighted parametrization provides two tests. The first test ensures that member_id 1 can delete their to-do, and the second test ensures that member_id 2 cannot delete another user’s to-do.
We should also add a similar test to ensure that the update works as expected:
@pytest.mark.parametrize( "member_id, complete", [(1, True), (2, False)], ) async def test_update_todo( connection: Connection, member_id: int, complete: bool ) -> None: todo = await insert_todo( connection, 1, "Task", False, None ) await update_todo( connection, todo.id, member_id, "Task", True, None ) new_todo = await select_todo(connection, todo.id, 1) assert new_todo is not None assert new_todo.complete is complete
The parametrization provides two tests. The first test ensures that the member with member_id 1 can update their to-do, and the second test ensures that the member with member_id 2 cannot update another user’s to-do.
While we have these important tests in place, we can’t run them until we create the database tables via a migration.
Running the first migration
While we’ve written the SQL queries required to create the database schema, they haven’t run against the database. To run these, Quart-DB provides a migration system, whereby we can run queries as the backend starts, but only if they haven’t already run. To make use of this, we can add the following code to backend/src/backend/migrations/0.py:
from quart_db import Connection async def migrate(connection: Connection) -> None: await connection.execute( """CREATE TABLE members ( id INT PRIMARY KEY GENERATED ALWAYS AS IDENTITY, created TIMESTAMP NOT NULL DEFAULT now(), email TEXT NOT NULL, email_verified TIMESTAMP, password_hash TEXT NOT NULL )""", ) await connection.execute( """CREATE UNIQUE INDEX members_unique_email_idx ON members (LOWER(email) )""" ) await connection.execute( """CREATE TABLE todos ( id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY, complete BOOLEAN NOT NULL DEFAULT FALSE, due TIMESTAMPTZ, member_id INT NOT NULL REFERENCES members(id), task TEXT NOT NULL )""", ) async def valid_migration(connection: Connection) -> bool: return True
To see this migration take effect, you can run pdm run recreate-db and then start the backend (as the migration will run as the backend starts up). You can then use psql –U tozo to inspect the database and see the two new tables as shown in Figure 3.1:
Figure 3.1: The database schema after the migration.
There is a one-to-many relationship between the members and todos tables such that one member has many to-dos. Also note the schema_migration table is created and managed by Quart-DB to track migrations.
Adding test and development data
It is helpful to have some standardized initial data in the database when developing and running tests; for example, we can add a standard member with known credentials to log in, rather than have to create a new member every time the database is recreated. To do this, we can utilize the data path feature in Quart-DB.
For ease of use, we’ll add a single member to the database by adding the following to backend/src/backend/migrations/data.py:
from quart_db import Connection
async def execute(connection: Connection) -> None:
await connection.execute(
"""INSERT INTO members (email, password_hash)
VALUES ('[email protected]', '$2b$14$6yXjNza30kPCg3LhzZJfqeCWOLM.zyTiQFD4rdWlFHBTfYzzKJMJe'
)"""
)
await connection.execute(
"""INSERT INTO todos (member_id, task)
VALUES (1, 'Test Task')"""
)The password hash value corresponds to a value of password, which means the login will be with the email, password combination of [email protected], password.
To instruct Quart-DB to load and run this file, we need to add the following configuration variable to backend/development.env and backend/testing.env:
TOZO_QUART_DB_DATA_PATH="migrations/data.py"
We can now run the tests and check that they pass by running the following in the backend directory:
pdm run test
Now we’ve defined the data stored by the backend, we can focus on the API, starting with session management.
Building the session API
To manage user sessions, we need a session (authentication) API that provides routes to log in and log out (i.e., to create and delete sessions). Login should result in a cookie being set, and logout results in the cookie being deleted. As per the authentication setup, login should require an email and matching password. We’ll add this API via a sessions blueprint containing login, logout, and status functionality.
Creating the blueprint
A blueprint is a collection of route handlers and is useful to associate the related session functionality. It can be created with the following code in backend/src/backend/blueprints/sessions.py:
from quart import Blueprint
blueprint = Blueprint("sessions", __name__)The blueprint then needs to be registered with the app, by adding the following to backend/src/backend/run.py:
from backend.blueprints.sessions import blueprint as sessions_blueprint app.register_blueprint(sessions_blueprint)
With the blueprint created, we can now add specific functionality as routes.
Adding login functionality
The login functionality is described in a RESTful style as creating a session, and hence the route should be POST, expecting an email, a password, and a remember flag returning 200 on success and 401 on invalid credentials. This is done via the following, which should be added to backend/src/backend/blueprints/sessions.py:
from dataclasses import dataclass
from datetime import timedelta
import bcrypt
from pydantic import EmailStr
from quart import g, ResponseReturnValue
from quart_auth import AuthUser, login_user
from quart_rate_limiter import rate_limit
from quart_schema import validate_request
from backend.lib.api_error import APIError
from backend.models.member import select_member_by_email
@dataclass
class LoginData:
email: EmailStr
password: str
remember: bool = False
@blueprint.post("/sessions/")
@rate_limit(5, timedelta(minutes=1))
@validate_request(LoginData)
async def login(data: LoginData) -> ResponseReturnValue:
"""Login to the app.
By providing credentials and then saving the returned cookie.
"""
result = await select_member_by_email( g.connection, data.email )
if result is None:
raise APIError(401, "INVALID_CREDENTIALS")
passwords_match = bcrypt.checkpw(
data.password.encode("utf-8"),
result.password_hash.encode("utf-8"),
)
if passwords_match:
login_user(AuthUser(str(result.id)), data.remember)
return {}, 200
else:
raise APIError(401, "INVALID_CREDENTIALS")This route is rate limited to a lower limit than others (five requests a minute) to prevent malicious actors from brute forcing the login. This is where the malicious actor keeps trying different passwords in the hope that one will eventually be correct and allow login.
The route also validates the request data has the correct LoginData structure, which ensures that users correctly use this route, and prevents invalid data from causing errors in the route handler code.
The route itself tries to fetch the member’s details from the database given the email provided in the request data. If there is no data, a 401 response is returned. The password provided in the request data is then checked against the password hash in the database, with a match resulting in the member being logged in with a 200 response. If the passwords don’t match, a 401 response is returned.
Trailing slashes
For this route, and for all others in the app, I’ve added a trailing slash so that the path is /sessions/ rather than /sessions. This is a useful convention to follow as requests to /sessions will be automatically redirected to /sessions/ and hence work despite the missing slash, whereas requests to /sessions/ would not be redirected to /session if the route was defined without the trailing slash.
Logging in results in a cookie being stored in the member’s browser, which is then sent in every subsequent request. The presence and value of this cookie are used to determine whether the member is logged in, and which member made the request.
Account enumeration
This implementation will allow an attacker to enumerate emails present in the database, which can be considered a security issue. See Chapter 7, Securing and Packaging the App, for how to mitigate against this.
Adding logout functionality
A logout route is described as a session deletion in a RESTful style, therefore the route should be DELETE, returning 200. The following should be added to backend/src/backend/blueprints/sessions.py:
from quart_auth import logout_user
from quart_rate_limiter import rate_exempt
@blueprint.delete("/sessions/")
@rate_exempt
async def logout() -> ResponseReturnValue:
"""Logout from the app.
Deletes the session cookie.
"""
logout_user()
return {}This route is rate exempt as nothing should prevent a member from logging out – it is important that the logout function works so that members are logged out when they want to be. The route then only needs to call logout_user, which results in the cookie being deleted.
Idempotent routes
Idempotence is a property of a route where the final state is achieved no matter how many times the route is called, that is, calling the route once or 10 times has the same effect. This is a useful property as it means the route can be safely retried if the request fails. For RESTful and HTTP APIs, the routes using GET, PUT, and DELETE verbs are expected to be idempotent. In this book, the routes using the GET, PUT, and DELETE verbs are idempotent.
Adding status functionality
It is useful to have a route that returns the current session (status) as we’ll use it for debugging and testing. For a RESTful API, this should be a GET route, and the following should be added to backend/src/backend/blueprints/sessions.py:
from quart_auth import current_user, login_required
from quart_schema import validate_response
@dataclass
class Status:
member_id: int
@blueprint.get("/sessions/")
@rate_limit(10, timedelta(minutes=1))
@login_required
@validate_response(Status)
async def status() -> ResponseReturnValue:
assert current_user.auth_id is not None # nosec
return Status(member_id=int(current_user.auth_id))The highlighted assertion is used to inform the type checker that current_user.auth_id cannot be None in this function, and hence prevents the type checker from considering the subsequent line as an error. The # nosec comment informs the bandit security checker that this use of assert is not a security risk.
The route is rate limited for protection and will only run if the request has the correct cookie present from login. The route returns the member’s ID based on the value in the cookie as this is also very useful.
Testing the routes
We should test that these routes work as a user would expect, starting by testing that a user can log in, get their status, and then log out as a complete flow. This is tested by adding the following to backend/tests/blueprints/test_sessions.py:
from quart import Quart
async def test_session_flow(app: Quart) -> None:
test_client = app.test_client()
await test_client.post(
"/sessions/",
json={
"email": "[email protected]", "password": "password"
},
)
response = await test_client.get("/sessions/")
assert (await response.get_json())["memberId"] == 1
await test_client.delete("/sessions/")
response = await test_client.get("/sessions/")
assert response.status_code == 401This test ensures that a member can log in and then access routes that require them to be logged in. It then logs the member out and checks that they can no longer access the route.
We should also test that the login route returns the correct response if the wrong credentials are provided by adding the following test to backend/tests/blueprints/test_sessions.py:
async def test_login_invalid_password(app: Quart) -> None:
test_client = app.test_client()
await test_client.post(
"/sessions/",
json={
"email": "[email protected]", "password": "incorrect"
},
)
response = await test_client.get("/sessions/")
assert response.status_code == 401This is all we need to allow members to log in and log out. Next, we can focus on managing members.
Building the member API
To manage members, we need an API that provides routes to create a member (register), confirm the email address, change the password, request a password reset, and reset a password.
We’ll add this API via a blueprint for the member, containing registration, email confirmation, changing password, and password reset functionality.
Creating the members blueprint
To begin, we should create a blueprint for all the member routes, it is created with the following code in backend/src/backend/blueprints/members.py:
from quart import Blueprint
blueprint = Blueprint("members", __name__)The blueprint then needs to be registered with the app, by adding the following to backend/src/backend/run.py:
from backend.blueprints.members import blueprint as members_blueprint app.register_blueprint(members_blueprint)
With the blueprint created, we can now add the specific functionality as routes.
Creating a member
In our app, we want users to be able to register as members. This requires a route that accepts an email and a password. The route should then check the password is sufficiently complex, create a new member, and send a welcome email. As the route creates a member, it should use the POST method to be in the RESTful style.
We’ll add a link to the welcome email that the recipient of the email can visit to prove they registered with our app. This way, we have verified that the email address owner is the same user that registered. The link will work by including an authentication token in the path, with the token working as explained in Chapter 2, Creating a Reusable Backend with Quart.
We can do this by first creating an email template by adding the following to backend/src/backend/templates/welcome.html:
{% extends "email.html" %}
{% block welcome %}
Hello and welcome to tozo!
{% endblock %}
{% block content %}
Please confirm you signed up by following this
<a href="{{ config['BASE_URL'] }}/confirm-email/{{ token }}/">
link
</a>.
{% endblock %}The route itself should return 201 on success, as this status code indicates a successful creation. This is all achieved by adding the following to backend/src/backend/blueprints/members.py:
from dataclasses import dataclass
from datetime import timedelta
import asyncpg # type: ignore
import bcrypt
from itsdangerous import URLSafeTimedSerializer
from quart import current_app, g, ResponseReturnValue
from quart_schema import validate_request
from quart_rate_limiter import rate_limit
from zxcvbn import zxcvbn # type: ignore
from backend.lib.api_error import APIError
from backend.lib.email import send_email
from backend.models.member import insert_member
MINIMUM_STRENGTH = 3
EMAIL_VERIFICATION_SALT = "email verify"
@dataclass
class MemberData:
email: str
password: str
@blueprint.post("/members/")
@rate_limit(10, timedelta(seconds=10))
@validate_request(MemberData)
async def register(data: MemberData) -> ResponseReturnValue:
"""Create a new Member.
This allows a Member to be created.
"""
strength = zxcvbn(data.password)
if strength["score"] < MINIMUM_STRENGTH:
raise APIError(400, "WEAK_PASSWORD")
hashed_password = bcrypt.hashpw(
data.password.encode("utf-8"),
bcrypt.gensalt(14),
)
try:
member = await insert_member(
g.connection,
data.email,
hashed_password.decode(),
)
except asyncpg.exceptions.UniqueViolationError:
pass
else:
serializer = URLSafeTimedSerializer(
current_app.secret_key, salt=EMAIL_VERIFICATION_SALT,
)
token = serializer.dumps(member.id)
await send_email(
member.email,
"Welcome",
"welcome.html",
{"token": token},
)
return {}, 201As can be seen, the password strength is first checked, using zxcvbn, with weak passwords resulting in a 400 response. The password is then hashed and used with the email to insert a member. The new member’s ID is then used to create an email verification token, which is rendered into the email body before being sent to the given email address.
When the user follows the link, they will return to our app with the token for the email confirmation route to check.
Confirming the email address
When a user registers as a member, they are sent a link back to our app that includes an email verification token. The token identifies the member and hence confirms that the email address is correct. Therefore, we need a route that accepts the token and, if valid, confirms the email address. This updates the member’s email property in a RESTful sense and hence is achieved by adding the following to backend/src/backend/blueprints/members.py:
from itsdangerous import BadSignature, SignatureExpired
from backend.models.member import update_member_email_verified
ONE_MONTH = int(timedelta(days=30).total_seconds())
@dataclass
class TokenData:
token: str
@blueprint.put("/members/email/")
@rate_limit(5, timedelta(minutes=1))
@validate_request(TokenData)
async def verify_email(data: TokenData) -> ResponseReturnValue:
"""Call to verify an email.
This requires the user to supply a valid token.
"""
serializer = URLSafeTimedSerializer(
current_app.secret_key, salt=EMAIL_VERIFICATION_SALT
)
try:
member_id = serializer.loads( data.token, max_age=ONE_MONTH )
except SignatureExpired:
raise APIError(403, "TOKEN_EXPIRED")
except BadSignature:
raise APIError(400, "TOKEN_INVALID")
else:
await update_member_email_verified(g.connection, member_id)
return {} The token is checked via the loads method, and if it is expired a 403 response is returned, whereas if it is invalid a 400 response is returned. If the token is good, the member’s email is marked as verified in the database and a 200 response is returned.
Once a user has registered, and hopefully verified their email, they will want to be able to change their password.
Changing passwords
A user will want to change their password, which requires a route that accepts their new password and their old password. The old password is checked to make the member’s account more secure, as a malicious user gaining access via an unattended computer cannot change the member’s password (without also knowing the member’s password). The route will also need to check that the new password has sufficient complexity as with the registration route.
The route should also inform the user that the password has been changed by email. Doing so makes the member’s account more secure as the member can take corrective action if they are informed about a password change that they didn’t authorize. This email is defined by adding the following to backend/src/backend/templates/password_changed.html:
{% extends "email.html" %}
{% block content %}
Your Tozo password has been successfully changed.
{% endblock %}This route will update the password, which in a RESTful style means a PUT route on the member’s password resource that returns 200 on success. It should return a 400 response if the password is not complex enough and a 401 response if the old password is incorrect. The following should be added to backend/src/backend/blueprints/members.py:
from typing import cast
from quart_auth import current_user, login_required
from backend.models.member import select_member_by_id, update_member_password
@dataclass
class PasswordData:
current_password: str
new_password: str
@blueprint.put("/members/password/")
@rate_limit(5, timedelta(minutes=1))
@login_required
@validate_request(PasswordData)
async def change_password(data: PasswordData) -> ResponseReturnValue:
"""Update the members password.
This allows the user to update their password.
"""
strength = zxcvbn(data.new_password)
if strength["score"] < MINIMUM_STRENGTH:
raise APIError(400, "WEAK_PASSWORD")
member_id = int(cast(str, current_user.auth_id))
member = await select_member_by_id(
g.connection, member_id
)
assert member is not None # nosec
passwords_match = bcrypt.checkpw(
data.current_password.encode("utf-8"),
member.password_hash.encode("utf-8"),
)
if not passwords_match:
raise APIError(401, "INVALID_PASSWORD")
hashed_password = bcrypt.hashpw(
data.new_password.encode("utf-8"),
bcrypt.gensalt(14),
)
await update_member_password(
g.connection, member_id, hashed_password.decode()
)
await send_email(
member.email,
"Password changed",
"password_changed.html",
{},
)
return {}As with the login route, this route has a lower rate limit to mitigate against brute force attacks. The code then checks the password strength before checking that the old password matches the hash stored in the database. If these checks pass, the password hash in the database is updated and an email is sent to the member.
This functionality is intentionally not useful for member’s that have forgotten their password. In that case, they first need to request a password reset.
Requesting a password reset
If a member forgets their password, they’ll want a way to reset it. This is typically provided by emailing the member a link that they can follow to a password reset page with the link containing a token to authorize the reset – as with the email verification. For this to work, we first need a route that accepts the user’s email address and sends out the link. To start, let’s add the following email content to backend/src/backend/templates/forgotten_password.html:
{% extends "email.html" %}
{% block content %}
You can use this
<a href="{{ config['BASE_URL'] }}/reset-password/{{ token }}/">
link
</a>
to reset your password.
{% endblock %}The route itself should accept an email address, and in the RESTful style should be a PUT to the member email resource. The following should be added to backend/src/backend/blueprints/members.py:
from pydantic import EmailStr
from backend.models.member import select_member_by_email
FORGOTTEN_PASSWORD_SALT = "forgotten password" # nosec
@dataclass
class ForgottenPasswordData:
email: EmailStr
@blueprint.put("/members/forgotten-password/")
@rate_limit(5, timedelta(minutes=1))
@validate_request(ForgottenPasswordData)
async def forgotten_password(data: ForgottenPasswordData) -> ResponseReturnValue:
"""Call to trigger a forgotten password email.
This requires a valid member email.
"""
member = await select_member_by_email( g.connection, data.email )
if member is not None:
serializer = URLSafeTimedSerializer(
current_app.secret_key, salt=FORGOTTEN_PASSWORD_SALT,
)
token = serializer.dumps(member.id)
await send_email(
member.email,
"Forgotten password",
"forgotten_password.html",
{"token": token},
)
return {}This route creates a token using the forgotten-password salt. It is important that the salt differs to ensure that these tokens cannot be used in place of the email verification token and vice versa. The token is then rendered into the email and sent to the member.
Resetting the password
If the member follows the link emailed out by the previous route, they will visit a page that allows them to enter a new password. Therefore, we need a route that accepts the new password and the token. This is achieved by adding the following to backend/src/backend/blueprints/members.py:
ONE_DAY = int(timedelta(hours=24).total_seconds())
@dataclass
class ResetPasswordData:
password: str
token: str
@blueprint.put("/members/reset-password/")
@rate_limit(5, timedelta(minutes=1))
@validate_request(ResetPasswordData)
async def reset_password(data: ResetPasswordData) -> ResponseReturnValue:
"""Call to reset a password using a token.
This requires the user to supply a valid token and a
new password.
"""
serializer = URLSafeTimedSerializer(
current_app.secret_key, salt=FORGOTTEN_PASSWORD_SALT
)
try:
member_id = serializer.loads(data.token, max_age=ONE_ DAY)
except SignatureExpired:
raise APIError(403, "TOKEN_EXPIRED")
except BadSignature:
raise APIError(400, "TOKEN_INVALID")
else:
strength = zxcvbn(data.password)
if strength["score"] < MINIMUM_STRENGTH:
raise APIError(400, "WEAK_PASSWORD")
hashed_password = bcrypt.hashpw(
data.password.encode("utf-8"),
bcrypt.gensalt(14),
)
await update_member_password(
g.connection, member_id, hashed_password.decode()
)
member = await select_member_by_id(
g.connection, int(cast(str, current_user.auth_id))
)
assert member is not None # nosec
await send_email(
member.email,
"Password changed",
"password_changed.html",
{},
)
return {}This route checks whether the token is valid, returning either a 400 if it is not or a 403 if it has expired. The expiry is important as it protects against a member’s email being exposed in the future (as the token will have expired and hence is useless). Then, if the password is strong enough, the new hash is placed into the database.
Managing members
We’ve added functionality to create members and manage members’ passwords. However, we haven’t added functionality to manage a member’s account itself, for example, to close and delete it. This functionality will be dependent on the regulatory rules of your app as, for example, you may be required to keep data for a certain length of time.
With this route, we have all the functionality we require for member accounts and can now focus on testing the functionality.
Testing the routes
We should test that these routes work as a user would expect. Firstly, let’s test that new members can register and then log in by adding the following to backend/tests/blueprints/test_members.py:
import pytest
from quart import Quart
async def test_register(
app: Quart, caplog: pytest.LogCaptureFixture
) -> None:
test_client = app.test_client()
data = {
"email": "[email protected]",
"password": "testPassword2$",
}
await test_client.post("/members/", json=data)
response = await test_client.post("/sessions/", json=data)
assert response.status_code == 200
assert "Sending welcome.html to [email protected]" in caplog.textThis test registers a new member with the email [email protected] and then checks that the welcome email was sent to this address. Next, we need to check that the user can confirm their email address by adding the following to backend/tests/blueprints/test_members.py:
from itsdangerous import URLSafeTimedSerializer
from freezegun import freeze_time
from backend.blueprints.members import EMAIL_VERIFICATION_SALT
@pytest.mark.parametrize(
"time, expected",
[("2022-01-01", 403), (None, 200)],
)
async def test_verify_email(
app: Quart, time: str | None, expected: int
) -> None:
with freeze_time(time):
signer = URLSafeTimedSerializer(
app.secret_key, salt= EMAIL_VERIFICATION_SALT
)
token = signer.dumps(1)
test_client = app.test_client()
response = await test_client.put(
"/members/email/", json={"token": token}
)
assert response.status_code == expected
async def test_verify_email_invalid_token(app: Quart) -> None:
test_client = app.test_client()
response = await test_client.put(
"/members/email/", json={"token": "invalid"}
)
assert response.status_code == 400The highlighted line allows us to ensure that expired tokens result in a 403 response while current tokens succeed. The second test ensures that invalid tokens result in a 400 response.
Next, we will test that members can change their password by adding the following to backend/tests/blueprints/test_members.py:
async def test_change_password(
app: Quart, caplog: pytest.LogCaptureFixture
) -> None:
test_client = app.test_client()
data = {
"email": "[email protected]",
"password": "testPassword2$",
}
response = await test_client.post("/members/", json=data)
async with test_client.authenticated("2"): # type: ignore
response = await test_client.put(
"/members/password/",
json={
"currentPassword": data["password"],
"newPassword": "testPassword3$",
}
)
assert response.status_code == 200
assert "Sending password_changed.html to [email protected]" in caplog.textThis test registers a new member and then, while authenticated as that member, changes the password.
Then we can test that a user that has forgotten their password can request a reset link by adding the following to backend/tests/blueprints/test_members.py:
async def test_forgotten_password(
app: Quart, caplog: pytest.LogCaptureFixture
) -> None:
test_client = app.test_client()
data = {"email": "[email protected]"}
response = await test_client.put(
"/members/forgotten-password/", json=data
)
assert response.status_code == 200
assert "Sending forgotten_password.html to [email protected]" in caplog.textNow we have these simple tests in place, we can focus on the To-Do API.
Building the To-Do API
To manage to-dos, we need an API that provides functionality to create a new to-do, retrieve a to-do or to-dos, update a to-do, and delete a to-do (i.e., an API that has CRUD functionality). We’ll do this by creating a to-do blueprint with a route per CRUD function.
CRUD functionality
CRUD stands for Create, Read, Update, and Delete, and is used to describe a set of functionalities. It is often used to describe the functionality of RESTful APIs. Typically, for a RESTFul API, the Create route uses the POST HTTP method, Read uses GET, Update uses PUT, and Delete uses DELETE.
Creating the blueprint
The blueprint itself can be created with the following code in backend/src/backend/blueprints/todos.py:
from quart import Blueprint
blueprint = Blueprint("todos", __name__)The blueprint then needs to be registered with the app, by adding the following to backend/src/backend/run.py:
from backend.blueprints.todos import blueprint as todos_blueprint app.register_blueprint(todos_blueprint)
With the blueprint created, we can now add specific functionality as routes.
Creating a to-do
The first functionality we need is to create a to-do. The route should expect the to-do data and return the complete to-do with a 201 status code on success. Returning the complete to-do is useful as it contains the to-do’s ID and confirms that the data is added. A RESTful to-do creation route should use the POST verb and have a /todos/ path. The following should be added to backend/src/backend/blueprints/todos.py:
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import cast
from quart import g
from quart_auth import current_user, login_required
from quart_schema import validate_request, validate_response
from quart_rate_limiter import rate_limit
from backend.models.todo import insert_todo, Todo
@dataclass
class TodoData:
complete: bool
due: datetime | None
task: str
@blueprint.post("/todos/")
@rate_limit(10, timedelta(seconds=10))
@login_required
@validate_request(TodoData)
@validate_response(Todo, 201)
async def post_todo(data: TodoData) -> tuple[Todo, int]:
"""Create a new Todo.
This allows todos to be created and stored.
"""
todo = await insert_todo(
g.connection,
int(cast(str, current_user.auth_id)),
data.task,
data.complete,
data.due,
)
return todo, 201The route is rate limited to prevent malicious usage, with the assumption that normal users are unlikely to create more than 10 to-dos in 10 seconds (1 a second on average). It is also a route that requires the user to be logged in. The final two decorators ensure that the request and response data represent the to-do data and a complete to-do.
The route function simply inserts the data into the database and returns the complete to-do. Next, users will need to read a to-do from the backend.
Reading a to-do
Users will need to read a to-do based on its ID. This will be implemented as a GET route with the ID specified in the path. The route should then either return the to-do or a 404 response if the to-do does not exist. The following should be added to backend/src/backend/blueprints/todos.py:
from backend.lib.api_error import APIError
from backend.models.todo import select_todo
@blueprint.get("/todos/<int:id>/")
@rate_limit(10, timedelta(seconds=10))
@login_required
@validate_response(Todo)
async def get_todo(id: int) -> Todo:
"""Get a todo.
Fetch a Todo by its ID.
"""
todo = await select_todo(
g.connection, id, int(cast(str, current_user.auth_id))
)
if todo is None:
raise APIError(404, "NOT_FOUND")
else:
return todoAs with the creation route, this route includes rate limiting protection, requires the user to be logged in, and validates the response data. It then selects the to-do from the database based on the ID given in the path and returns it or a 404 response if no to-do exists. Note that the select_todo function requires the member’s ID, ensuring that members cannot read other members’ to-dos.
While reading a single to-do is useful, a user will also need to read all their to-dos in one call, which we’ll add next.
Reading the to-dos
A user will need to read all their to-dos, which for a RESTFul API should use the GET verb and return a list of to-dos on success. We’ll also allow the user to filter the to-dos based on the complete attribute, which should be optional and hence, in a RESTful API, is provided via a querystring. The querystring works via the request path, for example, /todos/?complete=true or /todos/?complete=false. The following should be added to backend/src/backend/blueprints/todos.py:
from quart_schema import validate_querystring
from backend.models.todo import select_todos
@dataclass
class Todos:
todos: list[Todo]
@dataclass
class TodoFilter:
complete: bool | None = None
@blueprint.get("/todos/")
@rate_limit(10, timedelta(seconds=10))
@login_required
@validate_response(Todos)
@validate_querystring(TodoFilter)
async def get_todos(query_args: TodoFilter) -> Todos:
"""Get the todos.
Fetch all the Todos optionally based on the complete status.
"""
todos = await select_todos(
g.connection,
int(cast(str, current_user.auth_id)),
query_args.complete,
)
return Todos(todos=todos)This route includes rate limit protection, requires logged-in usage, validates the response data, and includes validation of the querystring parameters. We can now move on to allowing updates to a to-do.
Updating a to-do
We need to provide functionality for members to update the data that makes up a to-do. For a RESTFul API, this route should use the PUT verb, expect the to-do data, and return the complete to-do on success or a 404 if the to-do does not exist. The following should be added to backend/src/backend/blueprints/todos.py:
from backend.models.todo import update_todo
@blueprint.put("/todos/<int:id>/")
@rate_limit(10, timedelta(seconds=10))
@login_required
@validate_request(TodoData)
@validate_response(Todo)
async def put_todo(id: int, data: TodoData) -> Todo:
"""Update the identified todo
This allows the todo to be replaced with the request data.
"""
todo = await update_todo(
g.connection,
id,
int(cast(str, current_user.auth_id)),
data.task,
data.complete,
data.due,
)
if todo is None:
raise APIError(404, "NOT_FOUND")
else:
return todoThis route includes rate limit protection, requires logged-in usage, and validates the request and response data. It then updates the to-do and returns the updated to-do or a 404 response if there is no to-do for the provided ID. Next, we’ll allow users to delete to-dos.
Deleting a to-do
For a RESTFul API, the to-do deletion route should use the DELETE verb, and return 202 whether the to-do exists or not. The following should be added to backend/src/backend/blueprints/todos.py:
from quart import ResponseReturnValue
from backend.models.todo import delete_todo
@blueprint.delete("/todos/<int:id>/")
@rate_limit(10, timedelta(seconds=10))
@login_required
async def todo_delete(id: int) -> ResponseReturnValue:
"""Delete the identified todo
This will delete the todo.
"""
await delete_todo(
g.connection, id, int(cast(str, current_user.auth_id))
)
return "", 202This route includes rate limit protection, requires logged-in usage, and deletes the to-do with the given ID as long as it belongs to the logged-in member.
With all the functionality for to-dos in place, we can now focus on testing that it works correctly.
Testing the routes
We should test that these routes work as a user would expect. Firstly, we need to ensure we can create new to-dos by adding the following to backend/tests/blueprints/test_todos.py:
from quart import Quart
async def test_post_todo(app: Quart) -> None:
test_client = app.test_client()
async with test_client.authenticated("1"): # type: ignore
response = await test_client.post(
"/todos/",
json={
"complete": False, "due": None, "task": "Test task"
},
)
assert response.status_code == 201
assert (await response.get_json())["id"] > 0Next, we can ensure we can read to-dos by adding the following to backend/tests/blueprints/test_todos.py:
async def test_get_todo(app: Quart) -> None:
test_client = app.test_client()
async with test_client.authenticated("1"): # type: ignore
response = await test_client.get("/todos/1/")
assert response.status_code == 200
assert (await response.get_json())["task"] == "Test Task"Continuing along the CRUD functionality, we can ensure that to-dos can be updated by adding the following to backend/tests/blueprints/test_todos.py:
async def test_put_todo(app: Quart) -> None:
test_client = app.test_client()
async with test_client.authenticated("1"): # type: ignore
response = await test_client.post(
"/todos/",
json={
"complete": False, "due": None, "task": "Test task"
},
)
todo_id = (await response.get_json())["id"]
response = await test_client.put(
f"/todos/{todo_id}/",
json={
"complete": False, "due": None, "task": "Updated"
},
)
assert (await response.get_json())["task"] == "Updated"
response = await test_client.get(f"/todos/{todo_id}/")
assert (await response.get_json())["task"] == "Updated"Finally, we can ensure that to-dos can be deleted by adding the following to backend/tests/blueprints/test_todos.py:
async def test_delete_todo(app: Quart) -> None:
test_client = app.test_client()
async with test_client.authenticated("1"): # type: ignore
response = await test_client.post(
"/todos/",
json={
"complete": False, "due": None, "task": "Test task"
},
)
todo_id = (await response.get_json())["id"]
await test_client.delete(f"/todos/{todo_id}/")
response = await test_client.get(f"/todos/{todo_id}/")
assert response.status_code == 404With these tests, we have all the functionality we need to manage to-dos.
Summary
In this chapter, we’ve defined how we are storing the data in the database and then built an API to manage sessions, members, and to-dos. This includes all the functionality our app will need via an easy-to-understand RESTful API.
While the to-do functionality is unlikely to be directly useful to your app, the CRUD functionality is a pattern you should use. In addition, the member and session APIs could be used directly in your app. Finally, you’ve hopefully gained an understanding of what makes a good RESTful API that you can apply and use elsewhere.
In the next chapter, we’ll create a styled frontend, including validated data entry in React, that we can use with this API or any other.
Further reading
We’ve built a fairly simple RESTful API in this chapter. As your API’s complexity increases, I’d recommend following the best practices at https://www.vinaysahni.com/best-practices-for-a-pragmatic-restful-api.