Neo4j and FastAPI Tutorial

FastAPI is a fantastic library for building APIs with Python. From the documentation:

FastAPI is a modern, fast (high-performance), web framework for building APIs with Python 3.6+ based on standard Python type hints.

Because both FastAPI and Neontology utilise Pydantic models, we can combine them to quickly build a Python based Neo4j API.

We'll need a few things before we begin:

pip install fastapi
pip install "uvicorn[standard]"

You'll also need a live Neo4j instance to connect to.

Defining Models

We're going to build a limited API for managing teams and team members.

First we need to define models for the different types of node we want to store.

# main.py
from typing import ClassVar, Optional, List

from fastapi import FastAPI, HTTPException
from neontology import BaseNode, BaseRelationship, init_neontology

class TeamNode(BaseNode):
    __primaryproperty__: ClassVar[str] = "teamname"
    __primarylabel__: ClassVar[str] = "Team"
    teamname: str
    slogan: str = "Better than the rest!"


class TeamMemberNode(BaseNode):
    __primaryproperty__: ClassVar[str] = "nickname"
    __primarylabel__: ClassVar[str] = "TeamMember"
    nickname: str

Adding a relationship

Team members should 'belong to' a team, so lets define that relationship.

class BelongsTo(BaseRelationship):
    __relationshiptype__: ClassVar[str] = "BELONGS_TO"

    source: TeamMemberNode
    target: TeamNode

Our first root

Now we need to initialise FastAPI and Neontology, and let's create a root route to try things:

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    init_neontology(
        init_neontology(
        neo4j_uri="NEO4J URI HERE",
        neo4j_username="NEO4J USERNAME HERE",
        neo4j_password="NEO4J PASSWORD HERE"
        )
    )

@app.get("/")
def read_root():
    return {"foo": "bar"}

What we have so far

With all this defined in main.py, we can check it's running with:

uvicorn main:app --reload

Now if you go to http://127.0.0.1:8000/ you should simply get {"foo":"bar"}.

But for a much more comprehensive view, you can checkout http://127.0.0.1:8000/docs or http://127.0.0.1:8000/redoc.

Adding teams

Now lets start actually interacting with our Neo4j graph.

We'll define a POST rout for adding a new team.

Because FastAPI and Neontology both use Pydantic models, we just need to give our node type as a type hint on the route so that FastAPI knows that data it receives should map to the TeamNode type.

Then we use Neontology's create() method to create that team in the database.

@app.post("/teams/")
async def create_team(team: TeamNode):

    team.create()

    return team

Now visit http://127.0.0.1:8000/docs again where you should see our POST entry for /teams/. Explore the information and then hit the Try it out button to post some information and create a team.

Getting teams

Here we'll add some more routes to get the teams that have been created.

/teams/ will provide a list of all the created teams.

/teams/<teamname> will let us access information about a specific team.

@app.get("/teams/")
async def get_teams() -> List[TeamNode]:

    return TeamNode.match_nodes()


@app.get("/teams/{pp}")
async def get_team(pp: str) -> Optional[TeamNode]:

    return TeamNode.match(pp)

You can now browse the API to create teams and get info about them.

Finishing Up

We can then add some more routes which will let us create team members and assign them to teams.

In full, this becomes:

# main.py
from typing import ClassVar, Optional, List

from fastapi import FastAPI, HTTPException
from neontology import BaseNode, BaseRelationship, init_neontology


class TeamNode(BaseNode):
    __primaryproperty__: ClassVar[str] = "teamname"
    __primarylabel__: ClassVar[str] = "Team"
    teamname: str
    slogan: str = "Better than the rest!"


class TeamMemberNode(BaseNode):
    __primaryproperty__: ClassVar[str] = "nickname"
    __primarylabel__: ClassVar[str] = "TeamMember"
    nickname: str


class BelongsTo(BaseRelationship):
    __relationshiptype__: ClassVar[str] = "BELONGS_TO"

    source: TeamMemberNode
    target: TeamNode


app = FastAPI()


@app.on_event("startup")
async def startup_event():
    init_neontology()


@app.get("/")
def read_root():
    return {"foo": "bar"}


@app.post("/teams/")
async def create_team(team: TeamNode):

    team.create()

    return team


@app.get("/teams/")
async def get_teams() -> List[TeamNode]:

    return TeamNode.match_nodes()


@app.get("/teams/{pp}")
async def get_team(pp: str) -> Optional[TeamNode]:

    return TeamNode.match(pp)


@app.post("/team-members/")
async def create_team_member(member: TeamMemberNode, team_name: str):

    team = TeamNode.match(team_name)

    if team is None:
        raise HTTPException(status_code=404, detail="Team doesn't exist")

    member.create()

    rel = BelongsTo(source=member, target=team)
    rel.merge()

    return member


@app.get("/team-members/")
async def get_team_members() -> List[TeamMemberNode]:

    return TeamMemberNode.match_nodes()


@app.get("/team-members/{pp}")
async def get_team_member(pp: str) -> Optional[TeamMemberNode]:

    return TeamMemberNode.match(pp)