Putting it together
- Python course
- Advanced
- Lesson 83/83
- 6 min
Combine all layers into a working mini framework.
Putting it together: decorator
The final layer adds call logging with a decorator. log_call is a decorator that wraps async functions; functools.wraps preserves the original function's identity.
import functools
def log_call(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
print("called: " + func.__name__)
return await func(*args, **kwargs)
return wrapperYou apply this decorator to fetch_tasks: @log_call. On each call, called: fetch_tasks is printed first, then the function's result is returned. This makes API calls traceable.
All layers together
Now all layers are in place:
- fetch_tasks: async skeleton + decorator.
- fetch_all: asyncio.gather.
- SQLAlchemy: Task model and query.
- FastAPI: /tasks route and TestClient.
@log_call
async def fetch_tasks():
return [{...} for t in session.query(Task).all()]This pattern is the backbone of real-world APIs: async I/O, persistent data, an HTTP route and observability. Each layer can be tested in isolation; adding the decorator does not break existing behavior.
Practice Add the log_call decorator
The starting code contains the full previous solution (async + SQLAlchemy + FastAPI). Add the following:
import functools- Define
def log_call(func):decorator; with@functools.wraps(func)defineasync def wrapper(*args, **kwargs):whose body first doesprint("called: " + func.__name__)thenreturn await func(*args, **kwargs); thenreturn wrapper - Redefine
async def fetch_tasks():with@log_call(the DB-querying version)
The last line should already be 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()
from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
# bring in functools
# define a decorator named log_call (async wrapper with functools.wraps)
# wrapper should first print the function name with a 'called:' prefix, then await func
# redefine fetch_tasks with @log_call, querying the DB
# 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()
import functools
def log_call(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
print("called: " + func.__name__)
return await func(*args, **kwargs)
return wrapper
@log_call
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
called: fetch_tasks
[{"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 Why does the log_call decorator use functools.wraps?
- To force it to be async
- To preserve the original function's identity Correct answer
- To make it faster
- To add comments
Explanation: wraps copies attributes like __name__ and __doc__.
2 Complete the wraps call
Question
import functools
def log_call(func):
@functools.____(func)
async def wrapper(*args, **kwargs):
return await func(*args, **kwargs)
return wrapper
Answer
import functools
def log_call(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
return await func(*args, **kwargs)
return wrapperExplanation: functools.wraps copies the original function's identity.
3 Match the integration concepts
| Term | Meaning |
|---|---|
log_call |
Writes call log |
functools.wraps |
Preserves identity |
@log_call |
Decorator application |
wrapper |
Wrapping function |