Building a GraphQL Endpoints with FastAPI and Strawberry: An Exciting Adventure with Test-Driven Development! 🚀
Learn how to build a GraphQL API to manage students and colleges using FastAPI and Strawberry.
GraphQL is an effective query language for APIs, allowing customers to query their desired records. FastAPI, alternatively, is a modern-day internet framework for building APIs in Python, recognized for its velocity and simplicity of use.
Strawberry is a Python library for GraphQL, which presents a safe manner for defining GraphQL schemas.
This article will show you how to build a GraphQL API to manage students and colleges using FastAPI and Strawberry. We'll use SQLAlchemy to define our records model, Strawberry to create a GraphQL schema, and FastAPI to set up API endpoints.
Along the way, we can use pytest to ensure our API works perfectly. and introduce a subscription to real-time updates.
Prerequisites
Before we begin, ensure you have the subsequent setup:
- Python 3.6+
- Pip (Python package manager)
Setup
1. Create a New Directory
Start by growing a new directory for our project and navigating it into:
mkdir student-api && cd student-api2. Set Up a Virtual Environment
Next, create a virtual environment and activate it:
python3 -m venv venv && source venv/bin/activate3. Install Required Packages
Install the necessary packages using pip:
pip install fastapi sqlalchemy strawberry-graphql pytest uvicorn[standard] httpxDefine the Data Models
We’ll start by defining our data models for students and colleges using SQLAlchemy. Create a models.py file and define the models as below:
# models.py
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class College(Base):
__tablename__ = "colleges"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
location = Column(String)
class Student(Base):
__tablename__ = "students"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
age = Column(Integer)
college_id = Column(Integer, ForeignKey("colleges.id"))Connect to the Database
Now, connect to the database and define a session to run queries. Create a database.py file and add the following code:
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base
DATABASE_URI = "sqlite:///./students.db"
engine = create_engine(DATABASE_URI)
Base.metadata.create_all(bind=engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
db = SessionLocal()
try:
return db
finally:
db.close()Define the GraphQL Schema
Define the GraphQL schema using Strawberry. Create a schema.py file and add the following code:
import asyncio
from typing import List, AsyncGenerator
import strawberry
from strawberry import Schema
from models import College, Student
from database import get_db
@strawberry.type
class CollegeType:
id: int
name: str
location: str
@strawberry.type
class StudentType:
id: int
name: str
age: int
college_id: int
@strawberry.type
class Query:
@strawberry.field
async def colleges(self) -> List[CollegeType]:
db=get_db()
colleges = db.query(College).all()
return [CollegeType(id=college.id, name=college.name, location=college.location) for college in colleges]
@strawberry.field
async def students(self) -> List[StudentType]:
db=get_db()
students = db.query(Student).all()
return [StudentType(id=student.id, name=student.name, age=student.age, college_id=student.college_id) for student in students]Implement GraphQL Mutations
After that, add mutations to create, update, and delete students and colleges. Update the schema.py file with the following code:
@strawberry.type
class Mutation:
@strawberry.mutation
async def create_college(self, name: str, location: str) -> CollegeType:
db=get_db()
college = College(name=name, location=location)
db.add(college)
db.commit()
db.refresh(college)
return CollegeType(id=college.id, name=college.name, location=college.location)
@strawberry.mutation
async def create_student(self, name: str, age: int, college_id: int) -> StudentType:
db=get_db()
college = db.query(College).filter(College.id == college_id).first()
if not college:
raise ValueError("College not found")
student = Student(name=name, age=age, college_id=college_id)
db.add(student)
db.commit()
db.refresh(student)
return StudentType(id=student.id, name=student.name, age=student.age, college_id=student.college_id)Create the FastAPI Application
Now, implement the FastAPI application and integrate it with Strawberry. Create a main.py file and add the following code:
from fastapi import FastAPI
from strawberry.asgi import GraphQL
from schema import schema
app = FastAPI()
@app.get("/")
async def index():
return {"message": "Welcome to the Student API"}
app.add_route("/graphql", GraphQL(schema))Test-Driven Development (TDD) Approach
We will use Test-Driven Development (TDD) methodologies to ensure our API functions correctly and bug-free. Write the following test cases in a test_student.py file within a tests directory:
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from models import Base, College, Student
from main import app
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
@pytest.fixture
async def async_session() -> AsyncSession:
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
async with engine.begin() as connection:
await connection.run_sync(Base.metadata.create_all)
async with async_session() as session:
async with session.begin():
yield session
await session.execute("DELETE FROM students")
await session.execute("DELETE FROM colleges")
@pytest.fixture
async def async_client(async_session: AsyncSession):
async with AsyncClient(app=app, base_url="http://testserver") as ac:
yield ac
@pytest.mark.asyncio
async def test_graphql_queries(async_session: AsyncSession, async_client: AsyncClient):
async with async_session() as session:
async with session.begin():
# Create some test data
college = College(name="Test College", location="Test Location")
session.add(college)
await session.commit()
# Test query
query = """
query {
colleges {
id
name
location
}
}
"""
response = await async_client.post("/graphql", json={"query": query})
assert response.status_code == 200
data = response.json()
assert "data" in data
assert "colleges" in data["data"]
assert len(data["data"]["colleges"]) == 1
@pytest.mark.asyncio
async def test_graphql_mutations(async_session: AsyncSession, async_client: AsyncClient):
# Test mutation
mutation = """
mutation {
create_college(name: "New College", location: "New Location") {
id
name
location
}
}
"""
response = await async_client.post("/graphql", json={"query": mutation})
assert response.status_code == 200
data = response.json()
assert "data" in data
assert "create_college" in data["data"]
assert data["data"]["create_college"]["name"] == "New College"
assert data["data"]["create_college"]["location"] == "New Location"
@pytest.mark.asyncio
async def test_graphql_subscription(async_session: AsyncSession, async_client: AsyncClient):
# Test subscription
subscription = """
subscription {
studentAdded(collegeId: 1) {
id
name
age
collegeId
}
}
"""
async with async_client.websocket_connect("/graphql") as ws:
await ws.send_json({"type": "connection_init"})
await ws.send_json({"id": "1", "type": "start", "payload": {"query": subscription}})
response = await ws.receive_json()
assert response["type"] == "data"
assert "studentAdded" in response["payload"]["data"]
assert response["payload"]["data"]["studentAdded"]["id"] is not None
Implement Subscriptions
1. Define the Subscription
Add the following code to schema.py:
@strawberry.type
class Subscription:
@strawberry.subscription
async def student_added(self, college_id: int) -> StudentType:
async for student in student_stream(college_id):
yield student2. Define the Stream Function
Add the following code to schema.py:
async def student_stream(college_id: int) -> AsyncGenerator[StudentType, None]:
while True:
await asyncio.sleep(5) # Simulate real-time updates
db=get_db()
student = db.query(Student).filter(Student.college_id == college_id).order_by(Student.id.desc()).first()
if student:
yield StudentType(id=student.id, name=student.name, age=student.age, college_id=student.college_id)3. Update the Schema
Update the schema definition in schema.py to include the Subscription:
schema = Schema(query=Query, mutation=Mutation, subscription=Subscription)Run GraphQL API Over Client
When you run the below command on the terminal:
uvicorn main:app --reloadThe app will enable localhost with port 8000
http://127.0.0.1:8000/graphqlYou can directly go to the web page and see the GraphQL webpage, where you can run GraphQL queries, mutations, and subscriptions.
Or else, use a third-party tool like Altair. Download it from here.
4. Test the Mutation
You can test the mutation using a GraphQL client by running the following mutation query:
mutation {
createStudent(name: "deepak", age: 18, collegeId: 5) {
id
name
age
collegeId
}
}It will create data in the database as given in the query.
5. Test the Query
You can test the Query using a GraphQL client, Query the things or columns, or do whatever is needed.
Query to the Collage by running the following query:
query {
colleges {
id
name
location
}
}It will provide a list of data that includes defined query columns like id, name, and location.
6. Test the Subscription
You can test the subscription using a GraphQL client that supports subscriptions, like GraphQL Altair.
Subscribe to the student_added subscription by running the following subscription query:
subscription {
studentAdded(collegeId: 1) {
id
name
age
}
}Whenever a new student is added to college with collegeId 1, you should receive real-time updates.
Conclusion
Congratulations on completing this journey of constructing a GraphQL API with FastAPI and Strawberry! 🎉 Now, it’s your turn to create your API. Share your ideas and projects in the comments below!
Keep coding, and may your APIs be swift and your queries precise!😉
For more updates on the latest tools and technologies, follow the Simform Engineering blog.
