FastAPI layer
- Python course
- Advanced
- Lesson 82/83
- 6 min
Write REST endpoints with FastAPI.
FastAPI layer
The fourth layer adds an HTTP route with FastAPI. The @app.get("/tasks") decorator maps the read_tasks function to the /tasks route. The function queries the database with asyncio.run(fetch_tasks()).
from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
@app.get("/tasks")
def read_tasks():
return asyncio.run(fetch_tasks())TestClient queries the app like an HTTP server; client.get("/tasks").text gives the response body. The response is converted from a Python list of dicts into structured data automatically.
Querying with TestClient
TestClient(app) queries the FastAPI app in memory. client.get("/tasks") returns a response object; .text gives the body, .status_code the status.
client = TestClient(app)
resp = client.get("/tasks")
print(resp.text)
print(resp.status_code) # 200The response body is the Python list converted into structured data. This lets you test the whole flow without running a server; it mimics a real HTTP request.
Practice Add the FastAPI route
The starting code contains the full previous solution (async + SQLAlchemy). Add the following:
from fastapi import FastAPIfrom fastapi.testclient import TestClientapp = FastAPI()- With
@app.get("/tasks")definedef read_tasks():whose body doesreturn asyncio.run(fetch_tasks()) client = TestClient(app)- Print the response body with
print(client.get("/tasks").text)
Starter code
import asyncio
async def fetch_tasks():
return [{"id": 1, "title": "Görev 1"}, {"id": 2, "title": "Görev 2"}]
async def fetch_all():
return await asyncio.gather(fetch_tasks(), fetch_tasks())
from sqlalchemy import create_engine, Column, Integer, String, StaticPool
from sqlalchemy.orm import declarative_base, sessionmaker
Base = declarative_base()
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True)
title = Column(String)
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
session.add(Task(id=1, title="Görev 1"))
session.add(Task(id=2, title="Görev 2"))
session.commit()
async def fetch_tasks():
return [{"id": t.id, "title": t.title} for t in session.query(Task).all()]
# bring in FastAPI from fastapi
# bring in TestClient from fastapi.testclient
# create the app by calling the FastAPI class
# bind read_tasks to the '/tasks' route via app.get, returning asyncio.run(fetch_tasks())
# query '/tasks' with TestClient and print the bodySample solution
import asyncio
async def fetch_tasks():
return [{"id": 1, "title": "Görev 1"}, {"id": 2, "title": "Görev 2"}]
async def fetch_all():
return await asyncio.gather(fetch_tasks(), fetch_tasks())
from sqlalchemy import create_engine, Column, Integer, String, StaticPool
from sqlalchemy.orm import declarative_base, sessionmaker
Base = declarative_base()
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True)
title = Column(String)
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
session.add(Task(id=1, title="Görev 1"))
session.add(Task(id=2, title="Görev 2"))
session.commit()
async def fetch_tasks():
return [{"id": t.id, "title": t.title} for t in session.query(Task).all()]
from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
@app.get("/tasks")
def read_tasks():
return asyncio.run(fetch_tasks())
client = TestClient(app)
print(client.get("/tasks").text)Expected output
[{"id":1,"title":"Görev 1"},{"id":2,"title":"Görev 2"}]
Questions and answers
Review the lesson questions together with their answers. To test yourself, solve the lesson interactively.
1 What does the @app.get("/tasks") decorator do?
- Defines a decorator
- Creates the database
- Starts the asyncio loop
- Maps read_tasks to the /tasks route Correct answer
Explanation: @app.get('/tasks') maps read_tasks to the /tasks route.
2 Complete the TestClient call
from fastapi.testclient import TestClient
client = ____(app)from fastapi.testclient import TestClient
client = TestClient(app)Explanation: TestClient(app) creates a test client.
3 Match the FastAPI layer concepts
| Term | Meaning |
|---|---|
FastAPI() |
Creates the app |
@app.get |
Defines a route |
TestClient |
Gives a client |
.text |
Gives the body |