Building a REST API with EdgeDB and FastAPI​

Create a project directory​

To get started, create a directory for your project and change into it.

$ 

mkdir fastapi-crud

$ 

cd fastapi-crud

Install the dependencies​

Create a Python 3.10 virtual environment, activate it, and install the dependencies with this command (in Linux/macOS; see the following note for help with Windows):

$ 

python -m venv myvenv

$ 

source myvenv/bin/activate

$ 

pip install edgedb fastapi 'httpx[cli]' uvicorn

Initialize the database​

Now, let’s initialize an EdgeDB project. From the project’s root directory:

$ 

edgedb project init

No `edgedb.toml` found in `<project-path>` or above
Do you want to initialize a new project? [Y/n]
> Y
Specify the name of EdgeDB instance to use with this project [default:
fastapi_crud]:
> fastapi_crud
Checking EdgeDB versions...
Specify the version of EdgeDB to use with this project [default: 2.7]:
> 2.7

Once you’ve answered the prompts, a new EdgeDB instance called fastapi_crud will be created and started. If you see Project initialized, you’re ready.

Connect to the database​

Let’s test that we can connect to the newly started instance. To do so, run:

$ 

edgedb

You should see this prompt indicating you are now connected to your new database instance:

EdgeDB 2.x (repl 2.x)
Type \help for help, \quit to quit.
edgedb>

You can start writing queries here. Since this database is empty, that won’t get you very far, so let’s start designing our data model instead.

Users API​

We want this app to be type safe, end to end. To achieve this, instead of hard-coding string queries into the app, we’ll use code generation to generate typesafe functions from queries we write in .edgeql files. These files are simple text files containing the queries we want our app to be able to run.

The code generator will search through our project for all files with the .edgeql extension and generate those functions for us as individual Python modules. When you installed the EdgeDB client (via pip install edgedb), the code generator was installed alongside it, so you’re already ready to go. We just need to write those queries!

We’ll write queries for one endpoint at a time to start so you can see how the pieces fit together. To keep things organized, create a new directory inside app called queries. Create a new file in app/queries named get_users.edgeql and open it in your editor. Write the query into this file. It’s the same one we would have written inline in our Python code as shown in the code block above:

select User {name, created_at};

We need one more query to finish off this endpoint. Create another file inside app/queries named get_user_by_name.edgeql and open it in your editor. Add this query:

select User {name, created_at}
filter User.name = <str>$name

Save that file and get ready to kick off the magic that is code generation! 🪄

$ 

edgedb-py

Found EdgeDB project: <project-path>
Processing <project-path>/app/queries/get_user_by_name.edgeql
Processing <project-path>/app/queries/get_users.edgeql
Generating <project-path>/app/queries/get_user_by_name.py
Generating <project-path>/app/queries/get_users.py

The code generator creates one module per query file by default and places them at the same path as the query files.

With code generated, we’re ready to write an endpoint. Let’s create the GET /users endpoint so that we can request the User objects saved in the database. Create a new file app/users.py, open it in your editor, and add the following code:

# app/users.py
from __future__ import annotations

import datetime
from http import HTTPStatus
from typing import List

import edgedb
from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel

from .queries import get_user_by_name_async_edgeql as get_user_by_name_qry
from .queries import get_users_async_edgeql as get_users_qry

router = APIRouter()
client = edgedb.create_async_client()


class RequestData(BaseModel):
    name: str


@router.get("/users")
async def get_users(
    name: str = Query(None, max_length=50)
) -> List[get_users_qry.GetUsersResult] | get_user_by_name_qry.GetUserByNameResult:

    if not name:
        users = await get_users_qry.get_users(client)
        return users
    else:
        user = await get_user_by_name_qry.get_user_by_name(client, name=name)
        return user

We’ve imported the generated code and aliased it (using as <new-name>) to make the module names we use in our code a bit neater.

The APIRouter instance does the actual work of exposing the API. We also create an async EdgeDB client instance to communicate with the database.

By default, this API will return a list of all users, but you can also filter the user objects by name. We have the RequestData class to handle the data an API consumer will need to send in case they want to get only a single user. The types we’re using in the return annotation have been generated by the EdgeDB code generation based on the queries we wrote and our database’s schema.

Note that we’re also calling the appropriate generated function based on whether or not the API consumer passes an argument for name.

This nearly gets us there but not quite. We have one potential outcome not accounted for: a query for a user by name that returns no results. In that case, we’ll want to return a 404 (not found).

To fix it, we’ll check in the else case whether we got anything back from the single user query. If not, we’ll go ahead and raise an exception. This will send the 404 (not found) response to the user.

# app/users.py
...
if not name:
    users = await get_users_qry.get_users(client)
    return users
else:
    user = await get_user_by_name_qry.get_user_by_name(client, name=name)
    if not user:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail={"error": f"Username '{name}' does not exist."},
        )
    return user
...

To summarize, in the get_users function, we use our generated code to perform asynchronous queries via the edgedb client. Then we return the query results. Afterward, the JSON serialization part is taken care of by FastAPI.

Before we can use this endpoint, we need to expose it to the server. We’ll do that in the main.py module. Create app/main.py and open it in your editor. Here’s the content of the module:

# app/main.py
from __future__ import annotations

from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware

from app import users

fast_api = FastAPI()

# Set all CORS enabled origins.
fast_api.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


fast_api.include_router(users.router)

Here, we import everything we need, including our own users module containing the router and endpoint logic for the users API. We instantiate the API, give it a permissive CORS configuration, and give it the users router.

To test the endpoint, go to the project root and run:

$ 

uvicorn app.main:fast_api --port 5001 --reload

This will start a uvicorn server and you’ll be able to start making requests against it. Earlier, we installed the HTTPx client library to make HTTP requests programmatically. It also comes with a neat command-line tool that we’ll use to test our API.

While the uvicorn server is running, bring up a new console. Activate your virtual environment by running source myenv/bin/activate and run:

$ 

httpx -m GET http://localhost:5001/users

You’ll see the following output on the console:

HTTP/1.1 200 OK
date: Sat, 16 Apr 2022 22:58:11 GMT
server: uvicorn
content-length: 2
content-type: application/json

[]

If you see this result, that means the API is working! It’s not especially useful though. Our request yields an empty list because the database is currently empty. Let’s create the POST /users endpoint in app/users.py to start saving users in the database. Before we do that though, let’s go ahead and create the new query we’ll need.

Create and open app/queries/create_user.edgeql and fill it with this query:

select (insert User {
    name := <str>$name
}) {
    name,
    created_at
};

Save the file and run edgedb-py to generate the new function. Now, we’re ready to open app/users.py again and add the POST endpoint. First, import the generated function for the new query:

# app/users.py
...
from .queries import create_user_async_edgeql as create_user_qry
from .queries import get_user_by_name_async_edgeql as get_user_by_name_qry
from .queries import get_users_async_edgeql as get_users_qry
...

Then write the endpoint to call that function:

# app/users.py
...
@router.post("/users", status_code=HTTPStatus.CREATED)
async def post_user(user: RequestData) -> create_user_qry.CreateUserResult:

    try:
        created_user = await create_user_qry.create_user(client, name=user.name)
    except edgedb.errors.ConstraintViolationError:
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST,
            detail={"error": f"Username '{user.name}' already exists."},
        )
    return created_user

In the above snippet, we ingest data with the shape dictated by the RequestData model and return a payload of the query results. The try...except block gracefully handles the situation where the API consumer might try to create multiple users with the same name. A successful request will yield the status code HTTP 201 (created) along with the new user’s id, name, and created_at as JSON.

To test it out, make a request as follows:

$ 
  

httpx -m POST http://localhost:5001/users \
      --json '{"name" : "Jonathan Harker"}'

The output should look similar to this:

HTTP/1.1 201 Created
...
{
  "id": "53771f56-6f57-11ed-8729-572f5fba7ddc",
  "name": "Jonathan Harker",
  "created_at": "2022-04-16T23:09:30.929664+00:00"
}

If you try to make the same request again, it’ll throw an HTTP 400 (bad request) error:

HTTP/1.1 400 Bad Request
...
{
"detail": {
  "error": "Username 'Jonathan Harker' already exists."
  }
}

Before we move on to the next step, create 2 more users called Count Dracula and Mina Murray. Once you’ve done that, we can move on to the next step of building the PUT /users endpoint to update existing user data.

We’ll start again with the query. Create a new file in app/queries named update_user.edgeql. Open it in your editor and enter this query:

select (
    update User filter .name = <str>$current_name
        set {name := <str>$new_name}
) {name, created_at};

Save the file and generate again using edgedb-py. Now, we’ll import that and add the endpoint over in app/users.py.

# app/users.py
...
from .queries import create_user_async_edgeql as create_user_qry
from .queries import get_user_by_name_async_edgeql as get_user_by_name_qry
from .queries import get_users_async_edgeql as get_users_qry
from .queries import update_user_async_edgeql as update_user_qry
...
@router.put("/users")
async def put_user(
    user: RequestData, current_name: str
) -> update_user_qry.UpdateUserResult:
    try:
        updated_user = await update_user_qry.update_user(
            client,
            new_name=user.name,
            current_name=current_name,
        )
    except edgedb.errors.ConstraintViolationError:
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST,
            detail={"error": f"Username '{user.name}' already exists."},
        )

    if not updated_user:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail={"error": f"User '{current_name}' was not found."},
        )
    return updated_user

Not much new happening here. We wrote our query with a current_name parameter for finding the user to be updated. The user argument will give us the changes to make to that user, which in this case can only be the name since that’s the only property a user has. We pull the name out of user and pass it as our new_name argument to the generated function. The endpoint calls the generated function passing the client and those two values, and the user is updated.

We’ve accounted for the possibility of a user trying to change a user’s name to a new name that conflicts with a different user. That will return a 400 (bad request) error. We’ve also accounted for the possibility of a user trying to update a user that doesn’t exist, which will return a 404 (not found).

Let’s save everything and test this out.

$ 
  
  

httpx -m PUT http://localhost:5001/users \
      -p 'current_name' 'Jonathan Harker' \
      --json '{"name" : "Dr. Van Helsing"}'

This will return:

HTTP/1.1 200 OK
...
[
  {
    "id": "53771f56-6f57-11ed-8729-572f5fba7ddc",
    "name": "Dr. Van Helsing",
    "created_at": "2022-04-16T23:09:30.929664+00:00"
  }
]

If you try to change the name of a user to match that of an existing user, the endpoint will throw an HTTP 400 (bad request) error:

$ 
  
  

httpx -m PUT http://localhost:5001/users \
      -p 'current_name' 'Count Dracula' \
      --json '{"name" : "Dr. Van Helsing"}'

This returns:

HTTP/1.1 400 Bad Request
...
{
  "detail": {
    "error": "Username 'Dr. Van Helsing' already exists."
  }
}

Since we’ve verified that endpoint is working, let’s move on to the DELETE /users endpoint. It’ll allow us to query the name of the targeted object to delete it.

Start by creating app/queries/delete_user.edgeql and filling it with this query:

select (
    delete User filter .name = <str>$name
) {name, created_at};

Generate the new function by again running edgedb-py. Then re-open app/users.py. This endpoint’s code will look similar to the endpoints we’ve already written:

# app/users.py
...
from .queries import create_user_async_edgeql as create_user_qry
from .queries import delete_user_async_edgeql as delete_user_qry
from .queries import get_user_by_name_async_edgeql as get_user_by_name_qry
from .queries import get_users_async_edgeql as get_users_qry
from .queries import update_user_async_edgeql as update_user_qry
...
@router.delete("/users")
async def delete_user(name: str) -> delete_user_qry.DeleteUserResult:
    try:
        deleted_user = await delete_user_qry.delete_user(
            client,
            name=name,
        )
    except edgedb.errors.ConstraintViolationError:
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST,
            detail={"error": "User attached to an event. Cannot delete."},
        )

    if not deleted_user:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail={"error": f"User '{name}' was not found."},
        )
    return deleted_user

This endpoint will simply delete the requested user if the user isn’t attached to any event. If the targeted object is attached to an event, the API will throw an HTTP 400 (bad request) error and refuse to delete the object. To test it out by deleting Count Dracula, on your console, run:

$ 
  

httpx -m DELETE http://localhost:5001/users \
      -p 'name' 'Count Dracula'

If it worked, you should see this result:

HTTP/1.1 200 OK
...
[
  {
    "id": "e6837562-6f55-11ed-8744-ff1b295ed864",
    "name": "Count Dracula",
    "created_at": "2022-04-16T23:23:56.630101+00:00"
  }
]

With that, you’ve written the entire users API! Now, we move onto the events API which is slightly more complex. (Nothing you can’t handle though. 😁)

Events API​

Let’s start with the POST /events endpoint, and then we’ll fetch the objects created via POST using the GET /events endpoint.

First, we need a query. Create a file app/queries/create_event.edgeql and drop this query into it:

with name := <str>$name,
    address := <str>$address,
    schedule := <str>$schedule,
    host_name := <str>$host_name

select (
    insert Event {
        name := name,
        address := address,
        schedule := <datetime>schedule,
        host := assert_single(
            (select detached User filter .name = host_name)
        )
    }
) {name, address, schedule, host: {name}};

Run edgedb-py to generate a function from that query.

Create a file in app named events.py and open it in your editor. It’s time to code up the endpoint to use that freshly generated query.

# app/events.py
from __future__ import annotations

from http import HTTPStatus
from typing import List

import edgedb
from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel

from .queries import create_event_async_edgeql as create_event_qry

router = APIRouter()
client = edgedb.create_async_client()


class RequestData(BaseModel):
    name: str
    address: str
    schedule: str
    host_name: str


@router.post("/events", status_code=HTTPStatus.CREATED)
async def post_event(event: RequestData) -> create_event_qry.CreateEventResult:
    try:
        created_event = await create_event_qry.create_event(
            client,
            name=event.name,
            address=event.address,
            schedule=event.schedule,
            host_name=event.host_name,
        )

    except edgedb.errors.InvalidValueError:
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST,
            detail={
                "error": "Invalid datetime format. "
                "Datetime string must look like this: "
                "'2010-12-27T23:59:59-07:00'",
            },
        )

    except edgedb.errors.ConstraintViolationError:
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST,
            detail=f"Event name '{event.name}' already exists,",
        )

    return created_event

Like the POST /users endpoint, the incoming and outgoing shape of the POST /events endpoint’s data are defined by the RequestData model and the generated CreateEventResult model respectively. The post_events function asynchronously inserts the data into the database and returns the fields defined in the select query we wrote earlier, along with the new event’s id.

The exception handling logic validates the shape of the incoming data. For example, just as before in the users API, the events API will complain if you try to create multiple events with the same name. Also, the field schedule accepts data as an ISO 8601 timestamp string. Values not adhering to that will incur an HTTP 400 (bad request) error.

It’s almost time to test, but before we can do that, we need to expose this new API in app/main.py. Open that file, and update the import on line 6 to also import events:

# app/main.py
...
from app import users, events
...

Drop down to the bottom of main.py and include the events router:

# app/main.py
...
fast_api.include_router(events.router)

Let’s try it out. Here’s how you’d create an event:

$ 
  

httpx -m POST http://localhost:5001/events \
      --json '{

                  "name":"Resuscitation",
                  "address":"Britain",
                  "schedule":"1889-07-27T23:59:59-07:00",
                  "host_name":"Mina Murray"
                }'

If everything worked, you’ll see output like this:

HTTP/1.1 200 OK
...
{
  "id": "0b1847f4-6f3d-11ed-9f27-6fcdf20ffe22",
  "name": "Resuscitation",
  "address": "Britain",
  "schedule": "1889-07-28T06:59:59+00:00",
  "host": {
    "name": "Mina Murray"
  }
}

To speed this up a bit, we’ll go ahead and write all the remaining queries in one shot. Then we can flip back to app/events.py and code up all the endpoints. Start by creating a file in app/queries named get_events.edgeql. This one is really straightforward:

select Event {name, address, schedule, host : {name}};

Save that one and create app/queries/get_event_by_name.edgeql with this query:

select Event {
    name, address, schedule,
    host : {name}
} filter .name = <str>$name;

Those two will handle queries for GET /events. Next, create app/queries/update_event.edgeql with this query:

with current_name := <str>$current_name,
    new_name := <str>$name,
    address := <str>$address,
    schedule := <str>$schedule,
    host_name := <str>$host_name

select (
    update Event filter .name = current_name
    set {
        name := new_name,
        address := address,
        schedule := <datetime>schedule,
        host := (select User filter .name = host_name)
    }
) {name, address, schedule, host: {name}};

That query will handle PUT requests. The last method left is DELETE. Create app/queries/delete_event.edgeql and put this query in it:

select (
    delete Event filter .name = <str>$name
) {name, address, schedule, host : {name}};

Run edgedb-py to generate the new functions. Open app/events.py so we can start getting these functions implemented in the API! We’ll start by coding GET. Import the newly generated queries and write the GET endpoint in events.py:

# app/events.py
...
from .queries import create_event_async_edgeql as create_event_qry
from .queries import delete_event_async_edgeql as delete_event_qry
from .queries import get_event_by_name_async_edgeql as get_event_by_name_qry
from .queries import get_events_async_edgeql as get_events_qry
from .queries import update_event_async_edgeql as update_event_qry
...
@router.get("/events")
async def get_events(
    name: str = Query(None, max_length=50)
) -> List[get_events_qry.GetEventsResult] | get_event_by_name_qry.GetEventByNameResult:
    if not name:
        events = await get_events_qry.get_events(client)
        return events
    else:
        event = await get_event_by_name_qry.get_event_by_name(client, name=name)
        if not event:
            raise HTTPException(
                status_code=HTTPStatus.NOT_FOUND,
                detail={"error": f"Event '{name}' does not exist."},
            )
        return event

Save that file and test it like this:

$ 

httpx -m GET http://localhost:5001/events

We should get back an array containing all our events (which, at the moment, is just the one):

HTTP/1.1 200 OK
...
[
    {
        "id": "0b1847f4-6f3d-11ed-9f27-6fcdf20ffe22",
        "name": "Resuscitation",
        "address": "Britain",
        "schedule": "1889-07-28T06:59:59+00:00",
        "host": {
            "name": "Mina Murray"
        }
    }
]

You can also use the GET /events endpoint to return a single event object by name. To locate the Resuscitation event, you’d use the name parameter with the GET API as follows:

$ 
  

httpx -m GET http://localhost:5001/events \
      -p 'name' 'Resuscitation'

That’ll return a result that looks like the response we just got without the name parameter, except that it’s a single object instead of an array.

HTTP/1.1 200 OK
...
{
  "id": "0b1847f4-6f3d-11ed-9f27-6fcdf20ffe22",
  "name": "Resuscitation",
  "address": "Britain",
  "schedule": "1889-07-28T06:59:59+00:00",
  "host": {
    "name": "Mina Murray"
  }
}

If we’d had multiple events, the response to our first test would have given us all of them.

Let’s finish off the events API with the PUT and DELETE endpoints. Open app/events.py and add this code:

# app/events.py
...
@router.put("/events")
async def put_event(
    event: RequestData, current_name: str
) -> update_event_qry.UpdateEventResult:
    try:
        updated_event = await update_event_qry.update_event(
            client,
            current_name=current_name,
            name=event.name,
            address=event.address,
            schedule=event.schedule,
            host_name=event.host_name,
        )

    except edgedb.errors.InvalidValueError:
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST,
            detail={
                "error": "Invalid datetime format. "
                "Datetime string must look like this: '2010-12-27T23:59:59-07:00'",
            },
        )

    except edgedb.errors.ConstraintViolationError:
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST,
            detail={"error": f"Event name '{event.name}' already exists."},
        )

    if not updated_event:
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
            detail={"error": f"Update event '{event.name}' failed."},
        )

    return updated_event


@router.delete("/events")
async def delete_event(name: str) -> delete_event_qry.DeleteEventResult:
    deleted_event = await delete_event_qry.delete_event(client, name=name)

    if not deleted_event:
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
            detail={"error": f"Delete event '{name}' failed."},
        )

    return deleted_event

The events API is now ready to handle updates and deletion. Let’s try out a cool alternative way to test these new endpoints.

Browse the endpoints using the native OpenAPI doc​

FastAPI automatically generates OpenAPI schema from the API endpoints and uses those to build the API docs. While the uvicorn server is running, go to your browser and head over to http://localhost:5001/docs. You should see an API navigator like this:

This documentation allows you to play with the APIs interactively. Let’s try to make a request to the PUT /events. Click on the API that you want to try and then click on the Try it out button. You can do it in the UI as follows:

Clicking the execute button will make the request and return the following payload:

You can do the same to test DELETE /events, just make sure you give it whatever name you set for the event in your previous test of the PUT method.