Python
Putting it together
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 wrapper
You 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.
Exercise 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) define async def wrapper(*args, **kwargs): whose body first does print("called: " + func.__name__) then return await func(*args, **kwargs); then return wrapper - Redefine
async def fetch_tasks(): with @log_call (the DB-querying version)
The last line should already be 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()
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)
Test yourself
-
Why does the log_call decorator use functools.wraps?
- To force it to be async
- To preserve the original function's identity
- To make it faster
- To add comments
Show answer
To preserve the original function's identity — wraps copies attributes like __name__ and __doc__.
-
Complete the wraps call
import functools
def log_call(func):
@functools.____(func)
async def wrapper(*args, **kwargs):
return await func(*args, **kwargs)
return wrapper
Show answer
import functools
def log_call(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
return await func(*args, **kwargs)
return wrapper
-
Match the integration concepts
- log_call
- Writes call log
- functools.wraps
- Preserves identity
- @log_call
- Decorator application
- wrapper
- Wrapping function
Next lesson: Advanced section exam
