Building a REST API with EdgeDB and FastAPI​

Install the dependencies​

To follow along, clone the repository and head over to the fastapi-crud directory.

$ 

git clone git@github.com:edgedb/edgedb-examples.git

$ 

cd edgedb-examples/fastapi-crud

Create a Python 3.10 virtual environment, activate it, and install the dependencies with this command. On Linux/macOS:

$ 

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

Initializing project...

Specify the name of EdgeDB instance to use with this project
[default: fastapi_crud]:
> fastapi_crud

Do you want to start instance automatically on login? [y/n]
> y
Checking EdgeDB versions...

Once you’ve answered the prompts, a new EdgeDB instance called fastapi_crud will be created and started.

Connect to the database​

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

$ 

edgedb

You should be connected to the database instance and able to see a prompt similar to this:

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

You can start writing queries here. However, the database is currently schemaless. Let’s start designing out data model.

User APIs​

Since the User type is simpler, we’ll start with that. Let’s create a GET /users endpoint so that we can see the User objects saved in the database. You can create the API with a couple of lines of code in FastAPI:

# fastapi-crud/app/users.py
from __future__ import annotations

import datetime
from http import HTTPStatus
from typing import Iterable

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

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


class RequestData(BaseModel):
    name: str


class ResponseData(BaseModel):
    name: str
    created_at: datetime.datetime


@router.get("/users")
async def get_users(
    name: str = Query(None, max_length=50)
    ) -> Iterable[ResponseData]:

    if not name:
        users = await client.query(
            "SELECT User {name, created_at};"
            )
    else:
        users = await client.query(
        """SELECT User {name, created_at}
            FILTER User.name = <str>$name""",
            name=name,
        )
    response = (
        ResponseData(
            name=user.name,
            created_at=user.created_at
        ) for user in users
    )
    return response

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 users, but you can also filter the objects by name.

In the get_users function, we perform asynchronous queries via the edgedb client and serialize the returned data with the ResponseData model. Then we aggregate the instances in a generator and return it. Afterward, the JSON serialization part is taken care of by FastAPI. This endpoint is exposed to the server in the main.py module. Here’s the content of the module:

# fastapi-crud/app/main.py
from __future__ import annotations

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

from app import events, 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(events.router)
fast_api.include_router(users.router)

To test the endpoint, go to the fastapi-crud directory and run:

$ 

uvicorn app.main:fast_api --port 5000 --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, on a new console, run:

$ 

httpx -m GET http://localhost:5000/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

[]

Our request yielded an empty list because the database is currently empty. Let’s create the POST /users endpoint to start saving users in the database. The POST endpoint can be built similarly:

# fastapi-crud/app/users.py
...
@router.post("/users", status_code=HTTPStatus.CREATED)
async def post_user(user: RequestData) -> ResponseData:
    try:
        (created_user,) = await client.query(
            """
            WITH
                new_user := (INSERT User {name := <str>$name})
            SELECT new_user {
                name,
                created_at
            };
            """,
            name=user.name,
        )
    except edgedb.errors.ConstraintViolationError:
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST,
            detail={
            "error": f"Username '{user.name}' already exists,"
            },
        )
    response = ResponseData(
        name=created_user.name,
        created_at=created_user.created_at,
    )
    return response

In the above snippet, we ingest data with the shape dictated by the RequestData model and return a payload with the shape defined in the ResponseData model. 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). To test it out, make a request as follows:

$ 
  

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

The output should look similar to this:

HTTP/1.1 201 Created
...
{
  "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 the user data. It can be built like this:

# fastapi-crud/app/users.py
...
@router.put("/users")
async def put_user(
    user: RequestData, filter_name: str
) -> Iterable[ResponseData]:
    try:
        updated_users = await client.query(
            """
            SELECT (
                UPDATE User FILTER .name=<str>$filter_name
                    SET {name:=<str>$name}
            ) {name, created_at};
            """,
            name=user.name,
            filter_name=filter_name,
        )
    except edgedb.errors.ConstraintViolationError:
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST,
            detail={
            "error": f"Username '{filter_name}' already exists."
            },
        )
    response = (
        ResponseData(
            name=user.name, created_at=user.created_at
        ) for user in updated_users
    )
    return response

Here, we’ll isolate the intended object that we want to update by filtering the users with the filter_name parameter. For example, if you wanted to update the properties of Jonathan Harker, the value of the filter_name query parameter would be Jonathan Harker. The following command changes the name of Jonathan Harker to Dr. Van Helsing.

$ 
  
  

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

This will return:

HTTP/1.1 200 OK
...
[
  {
    "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:5000/users \
      -p 'filter_name' 'Count Dracula' \
      --json '{"name" : "Dr. Van Helsing"}'

This returns:

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

Another API that we’ll need to cover is the DELETE /users endpoint. It’ll allow us to query the name of the targeted object and delete that. The code looks similar to the ones you’ve already seen:

# fastapi-crud/app/users.py
...
@router.delete("/users")
async def delete_user(name: str) -> Iterable[ResponseData]:
    try:
        deleted_users = await client.query(
            """SELECT (
                DELETE User FILTER .name = <str>$name
            ) {name, created_at};
            """,
            name=name,
        )
    except edgedb.errors.ConstraintViolationError:
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST,
            detail={
                "error": "User attached to an event. "
                "Cannot delete."
            },
        )

    response = (
        ResponseData(
            name=deleted_user.name,
            created_at=deleted_user.created_at
        ) for deleted_user in deleted_users
    )

    return response

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 delete Count Dracula, on your console, run:

$ 
  

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

That’ll return:

HTTP/1.1 200 OK
...
[
  {
    "name": "Count Dracula",
    "created_at": "2022-04-16T23:23:56.630101+00:00"
  }
]

Event APIs​

The event APIs are built in a similar manner as the user APIs. Without sounding too repetitive, let’s look at how the POST /events endpoint is created and then we’ll introspect the objects created with this API via the GET /events endpoint.

Take a look at how the POST API is built:

# fastapi-crud/app/events.py

from __future__ import annotations

import datetime
from http import HTTPStatus
from typing import Iterable

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

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


class RequestData(BaseModel):
    name: str


class ResponseData(BaseModel):
    name: str
    created_at: datetime.datetime


@router.post("/events", status_code=HTTPStatus.CREATED)
async def post_event(event: RequestData) -> ResponseData:
    try:
        (created_event,) = await client.query(
        """
        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 := (SELECT User FILTER .name = host_name)
        }) {name, address, schedule, host: {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=f"Event name '{event.name}' already exists,",
        )

    return ResponseData(
        name=created_event.name,
        address=created_event.address,
        schedule=created_event.schedule,
        host=Host(
            name=created_event.host.name
        ) if created_event.host else None,
    )

Like the POST /users API, here, the incoming and outgoing shape of the data is defined by the RequestData and ResponseData``models respectively. The ``post_events function asynchronously inserts the data into the database and returns the fields defined in the SELECT statement. EdgeQL allows us to perform insertion and selection of data fields at the same time. The exception handling logic validates the shape of the incoming data. For example, just as before, this API will complain if you try to create multiple events with the same. Also, the field schedule accepts data as an ISO 8601 timestamp string. Failing to do so will incur an HTTP 400 (bad request) error.

Here’s how you’d create an event:

$ 
  

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

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

That’ll return:

HTTP/1.1 200 OK
...
{
  "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 list and filter the event objects. To locate the Resuscitation event, you’d use the filter_name parameter with the GET API as follows:

$ 
  

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

That’ll return:

HTTP/1.1 200 OK
...
{
  "name": "Resuscitation",
  "address": "Britain",
  "schedule": "1889-07-28T06:59:59+00:00",
  "host": {
    "name": "Mina Murray"
  }
}

Take a look at the app/events.py file to see how the PUT /events and DELETE /events endpoints are constructed.

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:5000/docs. You should see an API navigator like this:

The doc 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: