KAHIBARO
Discord Login Register ☰
☰

Performance Optimization

Table of Contents

Why Performance Optimization Matters

In a production backend, performance is not just about “making it fast.” It affects:

For your final project, performance optimization means you:

  1. Measure where time and resources are spent.
  2. Choose the right strategy for each bottleneck.
  3. Make changes that are safe, testable, and observable.

This chapter focuses on practical techniques you can apply to your project, not on theoretical micro-optimizations.

Key rule: Always measure before and after you optimize. Never guess where the bottleneck is, and never change code for performance without checking that it actually helps.

Step 1: Define Performance Goals

Before changing anything, decide what “good” means. Examples for an API:

You do not need perfect numbers, but you do need targets so you can see if you are improving.

Example target table for an e‑commerce backend:

MetricEndpointTarget
P95 latencyGET /products≤ 150 ms
P95 latencyPOST /orders≤ 400 ms
ThroughputOverall≥ 150 RPS
Error rateAll 5xx responses< 1%
DB response timeAverage query time≤ 20 ms

Step 2: Measure and Profile

Basic Timing in Code

If you do not have a full metrics stack yet, you can still get useful timing information.

Example: FastAPI middleware to log request duration:

python 
import time
from fastapi import FastAPI, Request
import logging
logger = logging.getLogger(__name__)
app = FastAPI()
@app.middleware("http")
async def log_request_time(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = (start - time.perf_counter()) * -1000  # ms
    logger.info(
        "method=%s path=%s status=%s duration_ms=%.2f",
        request.method,
        request.url.path,
        response.status_code,
        duration,
    )
    return response

This gives you per-request latency in logs. You can use it to see which endpoints are slowest.

Profiling CPU-bound Code

If a specific function seems slow, you can profile it.

Example profiling script:

python 
import cProfile
import pstats
from myapp.some_module import slow_function
if __name__ == "__main__":
    profiler = cProfile.Profile()
    profiler.enable()
    slow_function()
    profiler.disable()
    stats = pstats.Stats(profiler).sort_stats("cumulative")
    stats.print_stats(20)  # show top 20 functions

Look at:

Using Load Testing Tools

You must test the application under load, not just locally by hand. Common tools:

Simple wrk example:

bash 
wrk -t4 -c100 -d30s http://localhost:8000/products

You will get latency percentiles and RPS. Compare these numbers before and after changes.

Important: Always load test non-production environments with similar configuration. Never stress your real users and production database when experimenting.

Step 3: Identify the Bottleneck

Performance problems usually fall into one of these categories:

CategorySymptomsTypical Fixes
DB boundSlow queries, DB CPU high, many I/O waitsIndexes, query optimization, caching, batching
I/O boundWaiting on external APIs, file or networkAsync I/O, caching, queues, parallel requests
CPU boundHigh CPU usage, tight loops, heavy computeAlgorithm changes, offloading to background workers
Concurrency / lock boundMany requests, low throughputConnection pooling, fewer locks, better architecture

Use logs and metrics:

Do not optimize everything. Start with the top 1 or 2 bottlenecks.

Database Performance

In a typical backend, the database is often the first real bottleneck.

Use Efficient Queries

Common problems:

Example of N+1 problem:

python 
# Inefficient: one query for users, and one per user for orders
users = session.query(User).all()
for user in users:
    print(user.name, len(user.orders))  # triggers a query per user

Better, use eager loading:

python 
from sqlalchemy.orm import selectinload
users = (
    session.query(User)
    .options(selectinload(User.orders))
    .all()
)
for user in users:
    print(user.name, len(user.orders))  # no extra queries

Use Indexes Wisely

Indexes speed up WHERE, ORDER BY, and JOIN conditions.

For example, a slow query:

sql 
SELECT * FROM orders WHERE user_id = 123 ORDER BY created_at DESC LIMIT 20;

Add an index to support this pattern:

sql 
CREATE INDEX idx_orders_user_created_at
ON orders (user_id, created_at DESC);

You can inspect query plans in PostgreSQL:

sql 
EXPLAIN ANALYZE
SELECT * FROM orders WHERE user_id = 123 ORDER BY created_at DESC LIMIT 20;

Look for:

Rule: Always confirm index usefulness with EXPLAIN ANALYZE. An index that is never used just slows down writes and takes disk.

Limit Data Returned

Returning a lot of rows or large payloads slows everything.

Techniques:

Example:

sql 
SELECT id, name, price
FROM products
WHERE category_id = 5
ORDER BY created_at DESC
LIMIT 20 OFFSET 0;

In SQLAlchemy:

python 
products = (
    session.query(Product.id, Product.name, Product.price)
    .filter(Product.category_id == category_id)
    .order_by(Product.created_at.desc())
    .limit(limit)
    .offset(offset)
    .all()
)

Using Caching Effectively

Caching can give huge gains with minimal code changes, especially for read-heavy endpoints.

Cache Where It Makes Sense

Ideal candidates:

Types of caches in your project:

LevelExampleUse case
Application cacheIn-memory dict in a processVery fast, but not shared
RedisKey-value data across instancesShared cache for multiple replicas
HTTP cacheClient or proxy cache (CDN)Public data, static responses

Simple Redis Cache Pattern

Example: cache product list by category.

python 
import json
from typing import List
import aioredis
from fastapi import Depends
redis = aioredis.from_url("redis://redis:6379", decode_responses=True)
CACHE_TTL_SECONDS = 60
async def get_products_by_category(category_id: int) -> List[dict]:
    cache_key = f"category:{category_id}:products"
    cached = await redis.get(cache_key)
    if cached is not None:
        return json.loads(cached)
    # Fallback to DB
    products = (
        session.query(Product)
        .filter(Product.category_id == category_id)
        .order_by(Product.created_at.desc())
        .all()
    )
    data = [p.to_dict() for p in products]
    await redis.set(cache_key, json.dumps(data), ex=CACHE_TTL_SECONDS)
    return data

Cache Invalidation

The hard part of caching is invalidation.

Examples:

Example invalidation on update:

python 
async def update_product(product_id: int, data: ProductUpdate):
    product = session.get(Product, product_id)
    # ... update fields ...
    session.commit()
    # Invalidate cache
    product_key = f"product:{product_id}"
    category_key = f"category:{product.category_id}:products"
    await redis.delete(product_key, category_key)
    return product

Rule: If you are not 100% sure that cached data is fresh enough, prefer correctness over caching. Do not serve stale or wrong data only to be faster.

Asynchronous I/O and Concurrency

Backends often spend time waiting on I/O, such as:

You can use async endpoints to handle more requests without blocking.

Async Endpoints for I/O Bound Work

Example: calling an external payment API.

python 
import httpx
from fastapi import APIRouter
router = APIRouter()
@router.post("/payments")
async def create_payment(order_id: int):
    async with httpx.AsyncClient(timeout=5.0) as client:
        response = await client.post(
            "https://payment.example.com/charge",
            json={"order_id": order_id},
        )
    response.raise_for_status()
    return response.json()

This does not block the server worker while waiting for the payment service.

Do Not Use Async for CPU-Bound Work

Long, CPU-heavy tasks should run in background workers, not directly inside API requests.

If you must run some CPU work synchronously, consider returning early and processing it with a queue, which you learned about in the background workers chapter.

Reducing Work in the Request Path

A simple, powerful strategy is to move non-critical work out of the request path.

Good candidates:

Use Background Jobs

Example: create order now, send email later.

python 
from fastapi import APIRouter, BackgroundTasks
router = APIRouter()
def send_order_confirmation_email(order_id: int, email: str):
    # slow function, external SMTP or API calls
    ...
@router.post("/orders")
async def create_order(order_data: OrderCreate, background_tasks: BackgroundTasks):
    order = create_order_in_db(order_data)
    background_tasks.add_task(
        send_order_confirmation_email,
        order_id=order.id,
        email=order.customer_email,
    )
    # Response is fast, email is sent in background
    return {"id": order.id, "status": "created"}

For more complex workloads, use Celery or similar workers, already covered earlier.

Efficient Serialization and Responses

Serialization and response building can also be a source of latency, especially when you return large objects.

Return Only What Clients Need

Example, instead of this:

python 
class Product(BaseModel):
    id: int
    name: str
    description: str
    price: float
    created_at: datetime
    updated_at: datetime
    internal_notes: str

For a public listing endpoint, define a smaller response model:

python 
class ProductListItem(BaseModel):
    id: int
    name: str
    price: float

Use that in your endpoint:

python 
@router.get("/products", response_model=List[ProductListItem])
async def list_products(...):
    ...

Less data means:

Avoid Unnecessary Work in Response Building

Be careful with operations like:

If possible, let the database handle sorting and filtering, and construct your response in a straightforward way.

Connection Pooling and Resource Limits

When your backend scales to multiple replicas, connection usage becomes critical.

Database Connection Pooling

Using SQLAlchemy or async drivers, you should:

Example SQLAlchemy configuration:

python 
engine = create_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=5,
    pool_timeout=30,
)

You must size the pool based on:

Example calculation:

If PostgreSQL max_connections = 200, and you run 5 app instances, you might give each instance:

$$
\text{max\_pool\_connections\_total} = \text{instances} \times \text{pool\_size}
$$

You want:

$$
\text{max\_pool\_connections\_total} \le \text{max\_connections} - \text{reserved\_connections}
$$

Rule: If you see “too many connections” errors or high DB CPU, do not just increase max connections. First, check and tune your pools and queries.

HTTP Client Pooling

For external services, reuse HTTP clients where possible:

python 
import httpx
client = httpx.AsyncClient(timeout=5.0)
@router.get("/external")
async def call_external():
    resp = await client.get("https://api.example.com/data")
    return resp.json()

Creating a new client per request wastes time, sockets, and memory.

Using HTTP Caching and Compression

HTTP Caching Headers

When responses are cacheable, you should set headers like:

Example FastAPI response with headers:

python 
from fastapi import Response
@router.get("/public-products")
async def public_products():
    data = get_public_products()
    body = json.dumps(data)
    headers = {"Cache-Control": "public, max-age=60"}
    return Response(content=body, media_type="application/json", headers=headers)

Clients or proxies can then reuse responses instead of hitting your server again.

Compression

Most clients support gzip or similar. Use your web server or ASGI server to compress responses.

This reduces bandwidth for mid-to-large responses but adds some CPU. For JSON APIs, compression is almost always beneficial.

Optimize Critical Paths First

Not all endpoints are equal. Focus on:

A typical improvement loop for a single endpoint:

  1. Measure baseline latency and RPS with load testing.
  2. Inspect logs and DB queries for that endpoint.
  3. Apply one improvement:
    • Add index.
    • Add small cache.
    • Remove unnecessary work.
  4. Load test again.
  5. Compare metrics.

Example results table:

ChangeP95 latencyRPSNotes
Baseline420 ms80N+1 queries, no cache
Add eager loading260 ms120Fewer DB roundtrips
Add Redis cache (60s)90 ms300DB load reduced significantly

Avoid Common Anti-patterns

Some patterns look simple but can destroy performance.

Doing Work in Loops That Could Be Batched

Slow:

python 
orders_summary = []
for order_id in order_ids:
    order = session.get(Order, order_id)
    orders_summary.append(order.to_summary())

Faster:

python 
orders = (
    session.query(Order)
    .filter(Order.id.in_(order_ids))
    .all()
)
orders_summary = [o.to_summary() for o in orders]

Fetching All Rows Without Need

Avoid SELECT * FROM big_table without limits.

At API level, never design an endpoint that “returns all entries” for large collections. Always require pagination.

Synchronous Blocking in Async Code

In async endpoints, do not call blocking functions directly, such as:

They block the event loop and reduce concurrency.

Validate Performance With Tests

You do not need full performance tests in your unit test suite, but you should:

You can also add simple timing asserts in integration tests for critical paths:

python 
import time
def test_products_endpoint_is_fast(client):
    start = time.perf_counter()
    response = client.get("/products?limit=20")
    duration = time.perf_counter() - start
    assert response.status_code == 200
    assert duration < 0.5  # 500 ms

These do not guarantee real-world performance, but they help catch regressions.

Putting It All Together for the Final Project

For your production-ready final project, a practical performance optimization plan might look like this:

  1. Set targets
    • For example, P95 latency ≤ 250 ms for main GET endpoints.
  2. Instrument your app
    • Add request timing middleware.
    • Log slow queries from PostgreSQL.
    • Enable basic metrics collection.
  3. Optimize database usage
    • Fix N+1 patterns.
    • Add missing indexes for most common filters and joins.
    • Implement pagination everywhere lists are returned.
  4. Introduce caching
    • Use Redis for expensive, frequently read endpoints.
    • Invalidate cache entries on writes.
  5. Move non-critical work to background
    • Emails, some external integrations, heavy processing.
  6. Tune connection pools
    • Configure reasonable DB pool sizes for the number of app instances.
    • Reuse HTTP clients.
  7. Run load tests
    • Before deployment and after significant changes.
    • Compare against your targets and adjust.
  8. Monitor in production
    • Track latency, throughput, error rate, and resource usage.
    • Set alerts when metrics exceed thresholds.

Final principle: Performance optimization is a continuous process, not a one-time task. Measure, improve, deploy, and observe, then repeat when usage patterns or requirements change.

With these practices, your final project will not only function correctly, it will also handle real-world traffic and grow with your users.

Views: 5

Comments

Please login to add a comment.

Don't have an account? Register now!

© 2025 Kahibaro
Visitors: 3097 / 24h
Impressum / Imprint
Datenschutz / Data Policy