Python
FastAPI layer
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) # 200
The 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.
Exercise 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") define def read_tasks(): whose body does return asyncio.run(fetch_tasks()) client = TestClient(app)- Print the response body with
print(client.get("/tasks").text)
Show example 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)
Test yourself
-
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
Show answer
Maps read_tasks to the /tasks route — @app.get('/tasks') maps read_tasks to the /tasks route.
-
Complete the TestClient call
from fastapi.testclient import TestClient
client = ____(app)
Show answer
from fastapi.testclient import TestClient
client = TestClient(app)
-
Match the FastAPI layer concepts
- FastAPI()
- Creates the app
- @app.get
- Defines a route
- TestClient
- Gives a client
- .text
- Gives the body
Next lesson: Putting it together
