DEV Community: Igor Benav

FastAPI Mistakes That Kill Your Performance

Igor Benav — Sat, 13 Sep 2025 04:36:24 +0000

Most FastAPI performance problems aren't caused by FastAPI itself. They're caused by architectural issues - N+1 database queries, missing indexes, poor caching strategies. I covered these bigger problems in my previous post about Python's speed, and fixing those will give you 10-100x performance improvements.

But let's say you've already optimized your architecture. Your database queries are efficient, you're caching properly, and you're using async operations correctly. There are still FastAPI-specific optimizations that can give you meaningful performance gains - usually 20-50% improvements that add up.

Here's the thing: these optimizations won't save a badly designed system, but they can make a well-designed system significantly faster. Think of them as the final polish on an already efficient architecture.

TL;DR:

Install uvloop and httptools for faster event loops and HTTP parsing
Use ORJSONResponse for 20-50% faster JSON serialization
Choose async vs sync functions based on workload - I/O operations should be async, CPU-intensive work can be sync
Optimize Pydantic models with performance-focused configuration
Cache expensive dependencies to avoid repeated computations
Stream large responses to reduce memory usage by 80-90%
Increase thread pool size only if you must use sync operations
Use background tasks so users don't wait for non-critical operations
Run multiple workers in production to utilize all CPU cores
Enable GZip compression for large responses
Use FastAPI CLI production mode instead of development settings
Implement Pure ASGI middleware instead of BaseHTTPMiddleware
Avoid duplicate validation between request parsing and response models

Not Installing uvloop and httptools

An event loop is the heart of async programming - it's a continuous process that monitors and dispatches events and tasks. Think of it like a traffic controller at a busy intersection, deciding which car goes next and managing the flow of traffic. In Python's case, the event loop manages all your async operations: when to start database queries, when to process HTTP requests, when to handle responses from external APIs.

Python's default event loop (asyncio) is written in Python itself and does extensive safety checking. Every time it schedules a task or handles I/O, it's running interpreted Python code with all the associated overhead.

The uvloop package replaces Python's event loop with one written in Cython (compiled to C) and based on libuv - the same battle-tested C library that powers Node.js. Instead of interpreted Python managing your async operations, you get optimized C code handling the scheduling and I/O management.

The performance difference is huge under high concurrency. With 10 concurrent requests, you might not notice much difference, but with 1000+ concurrent requests (quite common in production), uvloop can handle 2-4x more throughput because the overhead of managing all those concurrent operations is reduced.

Beyond this, every HTTP request needs parsing - extracting headers, decoding the body, handling encoding. Python's default HTTP parser (h11) prioritizes correctness and handles edge cases gracefully, but it's pure Python code. httptools uses a C-based HTTP parser that's 40% faster but less forgiving of malformed requests - perfect for production where you control the client requests.

To apply these two changes is as simple as:

uv add uvloop httptools

The beauty is that Uvicorn automatically detects and uses them if installed - no code changes needed.

Note: uvloop doesn't work on Windows.

Using Python's Default JSON Encoder

Every time your FastAPI endpoint returns data, Python needs to convert your objects (dictionaries, lists, custom classes) into a JSON string that can be sent over HTTP. This process involves walking through your entire data structure, inspecting each value to determine its type, and converting it to the appropriate JSON representation.

Python's built-in json module is thorough but slow. For every single value, it asks: "Is this a string? A number? A boolean? A nested object?" It handles every edge case Python can throw at it, includes extensive type checking, and manages encoding issues. All this safety comes at a performance cost.

ORJSON is written in Rust with aggressive performance optimizations. It uses SIMD (Single Instruction, Multiple Data) instructions to process multiple values simultaneously, has specialized fast paths for common data types, and does minimal error checking.

from fastapi import FastAPI
from fastapi.responses import ORJSONResponse

app = FastAPI(default_response_class=ORJSONResponse)

@app.get("/data")
async def get_data():
    return {"message": "hello", "items": [1, 2, 3]}

The improvement is most dramatic for large responses with deeply nested data structures, APIs returning arrays with hundreds or thousands of items, responses with many numbers or dates. For small responses (under 1KB), the difference is minimal but there's no downside to using ORJSON.

Mixing Async and Sync Functions Incorrectly

This is critical and commonly misunderstood. The choice between async def and def fundamentally changes how FastAPI handles your endpoint, and getting it wrong can destroy performance.

How FastAPI handles each approach:

async def functions run in the main event loop alongside other requests
def functions are offloaded to a separate thread pool (limited to 40 threads by default)

When to use async def:

I/O-bound operations: database calls, HTTP requests, file operations
Operations that call other async functions
Light computational work: JSON parsing, data validation, simple transformations

When to use regular def:

CPU-intensive operations: image processing, heavy calculations, data analysis
Calling libraries that aren't async-compatible and do significant work
Operations that would benefit from running on a separate CPU core

import httpx

# GOOD - async for I/O operations
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
    return response.json()

# GOOD - sync for CPU-intensive work
@app.post("/process-data")
def process_data(data: bytes):
    # CPU-intensive processing
    result = heavy_computation(data)
    return {"result": result}

# BAD - blocking the event loop with CPU work
@app.post("/process-data-bad")
async def process_data_bad(data: bytes):
    # This blocks the entire event loop!
    result = heavy_computation(data)
    return {"result": result}

Using async database drivers like asyncpg (PostgreSQL) or aiomysql (MySQL) with async endpoints can provide 3-5x better throughput under concurrent load. This is because your entire request pipeline becomes truly asynchronous - no blocking operations.

With 40 default threads, if you have 41 users hitting sync endpoints simultaneously, one user waits for a thread to become available. With 1000 concurrent users hitting sync endpoints, 960 are queued waiting for threads - creating massive delays.

Overusing Pydantic Models Throughout Your Application

Pydantic is excellent for data validation at API boundaries, but using it everywhere in your application creates significant performance overhead that many developers don't realize.

The problem is subtle: Pydantic models look like regular Python classes, so it's tempting to use them as your primary data structures throughout your application. This creates what's known as "serialization/deserialization debt" - you're paying validation and conversion costs even when you don't need validation:

Pydantic object creation is 6.5x slower than Python dataclasses
Memory usage is 2.5x higher due to validation metadata storage
JSON operations are 1.5x slower across serialization and deserialization

This overhead compounds quickly. If you're creating thousands of objects during request processing, using Pydantic models internally can add significant latency.

Use Pydantic only at service boundaries:

from pydantic import BaseModel
from dataclasses import dataclass

# Good - Pydantic for API validation
class UserRequest(BaseModel):
    name: str
    email: str
    age: int

# Good - Dataclass for internal processing
@dataclass
class UserInternal:
    name: str
    email: str
    age: int
    created_at: datetime
    processed: bool = False

@app.post("/users")
async def create_user(user_request: UserRequest):
    # Pydantic validates the incoming request

    # Convert to internal dataclass for processing
    user = UserInternal(
        name=user_request.name,
        email=user_request.email,
        age=user_request.age,
        created_at=datetime.utcnow()
    )

    # Do internal processing with lightweight dataclass
    process_user(user)

    return {"id": user.id}

# Bad - Using Pydantic everywhere
class UserPydantic(BaseModel):
    name: str
    email: str
    age: int
    created_at: datetime
    processed: bool = False

def process_user(user: UserPydantic):
    # Every object creation/access pays validation overhead
    updated_user = UserPydantic(
        **user.dict(),
        processed=True
    )
    return updated_user

The rule of thumb: validate once at the boundary, then use lightweight structures internally. This gives you the safety of validation where it matters without paying the performance cost throughout your application.

Not Optimizing Pydantic Models for Performance

Pydantic models are validated on every request, so optimizing them provides immediate performance benefits for high-traffic endpoints.

from pydantic import BaseModel, Field
from typing import Optional

class OptimizedModel(BaseModel):
    model_config = {
        "validate_assignment": False,
        "str_strip_whitespace": True,
        "validate_default": False
    }

    name: str = Field(min_length=1, max_length=100)
    email: str = Field(pattern=r'^[^@]+@[^@]+\.[^@]+$')
    price: float = Field(gt=0)

# For query parameters, use models instead of individual parameters
class SearchParams(BaseModel):
    q: str
    limit: int = Field(10, ge=1, le=100)
    offset: int = Field(0, ge=0)
    sort: Optional[str] = None

@app.get("/search")
async def search(params: SearchParams = Depends()):
    return await search_database(params)

These config options reduce validation overhead for frequently-created models. Using models for query parameters is faster than individual Query() parameters because Pydantic can optimize the validation pipeline for the entire model at once.

Not Caching Expensive Dependencies

FastAPI automatically caches dependency results within a single request, but you can also create dependencies that persist across requests.

If multiple parts of your endpoint need the same dependency with identical parameters, FastAPI creates it once and reuses it within that request. For expensive resources that don't change often, you can create singleton dependencies that live for your entire application lifetime.

from functools import lru_cache
from fastapi import Depends

@lru_cache()
def get_settings():
    return {"api_key": "secret", "timeout": 30}

@app.get("/data")
async def get_data(settings = Depends(get_settings)):
    return {"config": settings}

A singleton ensures only one instance of a resource exists during your application's lifetime. In FastAPI, @lru_cache() without parameters creates singleton-like behavior - the function runs once, and all subsequent calls return the cached result.

Not Streaming Large Responses

For large datasets, loading everything into memory and returning it at once can exhaust server memory and create terrible user experience.

from fastapi.responses import StreamingResponse
import json

@app.get("/large-data")
async def stream_data():
    def generate():
        yield '{"items": ['
        for i in range(10000):
            if i > 0:
                yield ','
            yield json.dumps({"id": i, "name": f"item_{i}"})
        yield ']}'

    return StreamingResponse(generate(), media_type="application/json")

Instead of loading 10,000 records into memory (potentially 100MB+), streaming processes them one by one, keeping memory usage under 1MB. This is a 99% reduction in memory usage.

Using Small Thread Pool for Sync Operations

When you use a regular def function in FastAPI, it doesn't run in the main event loop. Instead, it runs in a thread pool - a collection of worker threads that handle synchronous operations. By default, FastAPI provides only 40 threads.

If 41 users hit a sync endpoint simultaneously, one waits for a thread. With 1000 concurrent users, 960 are stuck waiting. This creates massive response time degradation.

If you must use sync operations (legacy database drivers, CPU-bound tasks, third-party libraries without async support), you need more threads.

import anyio
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    limiter = anyio.to_thread.current_default_thread_limiter()
    limiter.total_tokens = 100
    yield

app = FastAPI(lifespan=lifespan)

@app.post("/process-image")
def process_image(image_data: bytes):
    return expensive_image_processing(image_data)

Convert to async when possible. An async endpoint using asyncio.to_thread() can handle thousands of concurrent requests without thread pool limitations.

Making Users Wait for Background Work

Background tasks let you queue work that runs after the HTTP response is sent. Users get their response immediately while non-essential operations happen in the background.

Instead of a 3-second response (1s user creation + 2s email sending), users get a 1-second response while the email sends behind the scenes.

from fastapi import BackgroundTasks

def send_email(email: str):
    # Email sending logic
    print(f"Sending email to {email}")

@app.post("/users")
async def create_user(email: str, background_tasks: BackgroundTasks):
    # Save user
    user_id = 123

    # Send email in background
    background_tasks.add_task(send_email, email)

    return {"user_id": user_id}

Some good use cases are include email notifications, file processing. For heavy operations (video processing, large imports), use Celery or ARQ instead.

Running on a Single Worker in Production

A worker is an independent process running your FastAPI application. By default, you get one worker using one CPU core, even on servers with 8+ cores.

Your FastAPI process runs on one CPU core. On a 4-core server, you're using 25% of available processing power. Under load, that single core becomes a bottleneck. Each worker is an independent FastAPI process. With 4 workers on 4 cores, you can handle 4x more concurrent requests.

# Single worker (default) - uses one CPU core
uvicorn main:app --host 0.0.0.0 --port 8000

# Multiple workers - uses multiple CPU cores
uvicorn main:app --workers 4 --host 0.0.0.0 --port 8000

# Better: Use Gunicorn to manage Uvicorn workers
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

# Modern approach: FastAPI CLI
fastapi run main.py --workers 4

Gunicorn adds process management, automatic worker restarts on crashes, graceful shutdowns, better resource monitoring.

To choose worker count you may follow this process:

Start with one worker per CPU core
Monitor under realistic load
I/O-heavy apps may benefit from more workers than cores
CPU-heavy apps can suffer from too many workers (context switching overhead)

Not Compressing Large Responses

GZip finds repeated patterns in your JSON and replaces them with shorter references. JSON is highly compressible because it contains repeated keys, similar values, and predictable structure.

A 500KB API response might compress to 80KB (84% smaller), turning a 2-second download on slow connections into 0.3 seconds.

from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=1000)

@app.get("/large-data")
async def get_large_data():
    return {"data": [{"id": i, "name": f"Item {i}"} for i in range(100)]}

Compressing tiny responses actually makes them larger due to compression overhead. The 1000-byte threshold ensures only responses that benefit get compressed. GZip helps most with large JSON arrays or objects, text-heavy API responses, repeated data structures, it does not help with already compressed data (images, videos), very small responses or binary data.

Using Development Settings in Production

Use fastapi run which automatically optimizes for production environments.

# Development mode - verbose logging, auto-reload, localhost only
fastapi dev main.py

# Production mode - optimized logging, all interfaces, no reload
fastapi run main.py

# Production with multiple workers
fastapi run main.py --workers 4

fastapi run optimizes:

Logging: Reduces verbose development logs
Host binding: Listens on 0.0.0.0 (all interfaces) vs 127.0.0.1 (localhost)
Auto-reload: Disabled (expensive and unnecessary in production)
Security headers: Better defaults

Using Slow BaseHTTPMiddleware

BaseHTTPMiddleware has performance overhead due to how it wraps requests and responses. For high-traffic applications, Pure ASGI middleware provides 40% better performance.

import time
from starlette.types import ASGIApp, Scope, Receive, Send

# Slower: BaseHTTPMiddleware approach
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

# Faster: Pure ASGI middleware
class ProcessTimeMiddleware:
    def __init__(self, app: ASGIApp):
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        start_time = time.time()

        async def send_with_time(message):
            if message["type"] == "http.response.start":
                process_time = time.time() - start_time
                headers = list(message.get("headers", []))
                headers.append([b"x-process-time", str(process_time).encode()])
                message["headers"] = headers
            await send(message)

        await self.app(scope, receive, send_with_time)

app.add_middleware(ProcessTimeMiddleware)

Pure ASGI middleware is more complex but avoids BaseHTTPMiddleware's request/response wrapping overhead.

Validating Data Multiple Times

If you're using type hints or response_model, there's no need to manually create Pydantic models in your endpoint - FastAPI handles this automatically and doing both causes double validation.

from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str

# BAD - double validation
@app.get("/users/{user_id}")
async def get_user_bad(user_id: int) -> User:
    user_data = {"id": user_id, "name": "John"}
    user = User(**user_data)  # First validation
    return user  # Second validation by FastAPI

# GOOD - single validation
@app.get("/users/{user_id}")
async def get_user_good(user_id: int) -> User:
    user_data = {"id": user_id, "name": "John"}
    return user_data  # Only one validation by FastAPI

If you have a type hint or response_model, return raw data (dicts, database objects) and let FastAPI handle model creation. Double validation can add 20-50% overhead to response processing.

Monitor Your Optimizations

Don't guess - measure the impact of your optimizations:

import time
import logging

logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time

    logger.info(f"{request.method} {request.url.path} - {process_time:.4f}s")
    return response

Focus on optimizing endpoints that handle the most traffic or take the longest to process.

The Reality Check

These optimizations provide meaningful performance improvements - typically 20-50% for well-architected applications. But they won't save you from fundamental design problems.

If your API is slow because of N+1 database queries, missing indexes, or poor caching, fix those first. They'll give you 10-100x improvements that dwarf any FastAPI-specific optimizations.

Architecture problems will kill your performance. FastAPI optimizations will enhance your performance.

Get the big stuff right first, then polish with these techniques.

Want a production-ready FastAPI setup with all these optimizations built-in? Check out FastroAI - it includes everything from this post plus authentication, database integration, and deployment configuration.

References:

Originally published at the FastroAI Blog.

Yes, Python is Slow, but it doesn’t matter for AI SaaS

Igor Benav — Wed, 10 Sep 2025 18:22:25 +0000

Python gets criticized for being slow. Benchmarks show Rust and C++ running circles around it. Go handles thousands more requests per second. The critics aren't wrong about raw performance numbers (ignoring the fact that languages can't actually be slow or fast).

But for most applications, especially AI SaaS, there's a lot of context being left out. The other day a CEO that I know was telling me that he wanted to go with Rust for his startup. What does the startup do? A bunch of OpenAI requests, like most of the new ones.

Software people like to say they're different, but the earlier you learn this, the best: people talk about trends. At first, code should be performant, then it had to be easy to write, then it had to be performant again… this misses the point.

Software engineering is not about writing code, it's about analyzing tradeoffs and making the best possible decisions based on all data, and - like Knuth said - premature optimization is the root of all evil (97% of the times). Let's actually make this decision (Python or a faster language like Rust) like a software engineer should do, analyzing the tradeoffs.

The Structured Process

What even is premature optimization? We could think of it in two ways:

Improving what works well (and, therefore, doesn't need to be improved)
Trying to improve without proper profiling, that is, knowing exactly how good you're doing

The first one is relevant, and you should always take it into account. Time is finite, and the opportunity cost of improving something that works instead of fixing something that doesn't (or shipping new features) is way too high in a startup to be ignored.

Here we'll talk more about the second one though, since I want to show how to actually make a decision using the whole context.

Let's start with profiling, that is, measuring how good we're doing and in what specific metrics we need to improve. For an API, usually this would mean measuring response time, throughput (successful requests per second), latency and some other metrics, but we don't have an actual API yet, we're deciding what to use, so we need to do something about it.

Measuring what we can't measure

We need to use a proxy variable for this, that is, measure something that is not directly what we need to measure, but that we assume (for some good reason) that is closely related to what we actually need, and therefore will give us a good approximate result.

Since our application mostly makes requests to AI services, we can start by assuming that improving this flow will yield the best immediate result, so let's start by understanding what constitutes a process that involves a request to an AI service.

Request Steps

Thinking in a simplified way, what happens inside our API when someone does something (like generating text) is:

Our API receives an http request (via the network)
Python orchestrates this and what needs to be done
We query the database for something (maybe multiple things even) like the user who is making the request, their permissions, usage
We make a request to OpenAI or any other provider for text generation (and we need to wait for this)
We store results in the database so the user can access it
We return a response to the client (another network request)

Ok, now that we know the steps, we can actually think how long each of these would take in a typical request. By doing this, we can properly understand what optimization would make the most difference (even better if it's for the smallest possible effort).

Let's time each of these steps (approximated) and see where our performance bottleneck is:

Our API receives an http request: This is network I/O (Input, Output). Typically 1-20ms depending on the client's connection and geographic distance. Nothing we can do here - the data has to travel through the network.
Python orchestrates the request: This is CPU bound (our processor is actually doing something, not just waiting), but minimal. Parsing JSON, validating inputs, basic business logic. Maybe 1-5ms total for typical AI SaaS operations if done correctly.
Database queries: Classic I/O bound operation. Even with proper indexing, you're looking at 10-200ms per query depending on query complexity and database location. Multiple queries compound this.
OpenAI API call: The big one. Text generation takes 500ms to 3 seconds depending on model and prompt length. Image generation? 10-30 seconds. This is pure waiting time.
Store results back: Another database write, 10-100ms typically.
Return response: Network I/O again, similar to receiving the request.

Let me be concrete here. In a typical AI request:

Total Python execution time: ~5ms
Total I/O waiting time: 1000-5000ms (or a lot more depending on the AI generation)

Python execution represents 0.1-0.5% of total request time. Even if we magically made Python infinitely fast, users wouldn't even notice the difference. The language we pick to write our backend isn't even close to being the bottleneck here, so what is?

The Real Bottleneck: I/O vs CPU Operations

The speed problem in AI applications isn't your Python code - it's waiting for everything else. This is the key insight: most AI SaaS applications are I/O bound, not CPU bound.

I/O Bound Operations wait for input/output to complete. Network requests, database queries, file operations. The bottleneck isn't computation speed - it's waiting for data.

CPU Bound Operations are limited by computational power. Heavy math, data processing, complex algorithms. These actually benefit from faster languages.

But here's the thing - if your AI SaaS is making OpenAI requests, you're overwhelmingly I/O bound. The language choice becomes irrelevant when 99% of your time is spent waiting for external services. Even if you manage to make your orchestration code 10x faster, you'd save maybe 4ms out of a 2000ms request. That's 0.2% improvement.

Meanwhile, if you cache database queries properly, use connection pooling, optimize your database queries, you'd be improving what actually makes a difference. Plus, these optimizations are language-agnostic, easily providing 10-100x more impact than just switching to Rust.

Ok, but you might be thinking "even if it doesn't matter if python is slow for the orchestration, it's still handling one request at a time". That leads us to the next thing, async.

Async Changes Everything

Modern Python isn't the blocking, single-threaded Python of 2010.

With async/await and frameworks like FastAPI, Python handles I/O bound workloads brilliantly. While one request waits for OpenAI to respond, your Python server processes hundreds of other requests.

@app.get("/generate")
async def generate_content(prompt: str, db: AsyncSession = Depends(get_db)):
    # While this waits for the database...
    user_data = await db.execute(select(User).where(...))

    # ...and this waits for OpenAI...
    response = await openai_client.chat.completions.create(...)

    # ...Python handles other requests concurrently
    await db.execute(insert(Generation).values(...))

    return response

This concurrency model is perfect for AI SaaS. You're not CPU bound anyway, so async Python easily handles thousands of concurrent requests. Your bottleneck becomes API rate limits, database connections, or memory usage - not language speed.

When Python Speed Actually Matters

Even though your case probably isn't one of these, there are cases where Python's lack of speed is actually a constraint:

Running ML models locally: If you're doing inference in Python rather than calling APIs, computation speed matters. But most AI SaaS companies use external APIs (OpenAI, Anthropic, etc.), not local models.
Heavy data preprocessing: Transforming massive datasets before feeding them to models. But again, most startups don't have massive datasets initially, and when they do, they use specialized tools like Spark or Dask.
Real-time applications: High-frequency trading, gaming servers, real-time video processing. These are CPU bound and benefit from faster languages. But if you're building ChatGPT for lawyers, this isn't you.
Mathematical computation: Complex algorithms that don't rely on external services. Signal processing, scientific computing, etc. Again, not typical for AI SaaS startups.

The thing is, these represent edge cases for most AI SaaS applications. If your business model is "call OpenAI API and add some business logic around it" (which describes most AI startups), you're I/O bound by definition.

If your business gets to a point where Python isn't enough, you can even migrate just the parts that need to be performant - most libraries for performance sensitive domains in python are already written in another language like Rust or C++ and just called by python.

What Actually Kills Your Performance

For 99% (heck, even a lot more) of AI SaaS, the performance killers are completely different:

N+1 database queries: Making separate database calls in loops instead of batch operations. I've seen codebases where getting 100 users takes 101 database queries instead of 1. That's not a language problem - that's a classical architecture problem.

# This kills performance - 100 database queries
for user_id in user_ids:
    user = await db.get(User, user_id)

# This is fast - 1 database query
users = await db.execute(select(User).where(User.id.in_(user_ids)))

Missing database indexes: Forcing full table scans on queries that should be instant. Adding an index can turn a 5-second query into a 5-millisecond query.

No caching: Repeatedly calling the same endpoints with the same parameters. Cache that response and serve it instantly next time.

Synchronous operations: Blocking your entire request handler while waiting for I/O. Use async/await properly.

Poor API usage: Making 5 separate API calls when you could batch them into 1.

These problems cause 10x to 100x performance degradation. Fix the architecture problems first. They'll give you way more bang for your buck.

The Real Tradeoffs

Now let's think about this problem like actual software engineers.

Python advantages for AI SaaS:

Ship features fast: Python's ecosystem is unmatched for AI and integrations
Find developers easily: Way larger talent pool than Rust or Golang
Iterate quickly: Write, test, deploy cycles are faster
Proven ecosystem: FastAPI, SQLAlchemy, OpenAI SDK - everything just works

Rust advantages:

Raw performance: 2-5x faster for CPU-bound operations
Memory safety: Harder to write buggy code (once you get past the learning curve)
Lower resource usage: Matters if you're processing millions of requests

The real question: What's your bottleneck?

If you're doing heavy computation, languages like Rust might make sense. But if you're making API calls to OpenAI (like most AI startups), your bottleneck is database and network I/O, not CPU performance.

Optimizing Python won't help when 99% of your request time is waiting for external services. And rewriting everything in Rust will take months while your competitors ship new features.

The CEO I mentioned earlier? Ended up picking python and is getting traction with the solution in the hands of actual customers. That's the real cost of premature optimization: opportunity cost.

Resource Usage

Something Python is genuinely inefficient about is resource usage. Python objects have more overhead, and the garbage collector keeps references around longer than manual memory management would.

We can see this from Lovable's example - they went from 200 server instances down to just 10 when they migrated from Python to Go. That's a 20x reduction in infrastructure costs. Their deployment times dropped from 15 minutes to 3 minutes, and average request times improved by 12%.

But here's the key context: Lovable was handling 50+ concurrent HTTP requests per chat request with heavy parallelism. That's not a typical AI SaaS workload - most startups are making sequential API calls to OpenAI, not orchestrating dozens of parallel operations. For most AI SaaS applications, resource efficiency doesn't matter much until you're big. Really big.

Your OpenAI API calls don't care if your web server uses 100MB or 500MB of RAM. Your database doesn't run faster because your API server has lower memory usage. If you're doing 1000 requests per day, the difference between running 1 server instance vs 2 is maybe $50/month - not exactly a startup killer.

What actually constrains resources in typical AI SaaS:

API rate limits: Hit OpenAI's rate limit and your users wait regardless of how efficient your memory usage is.
Database connections: Running out of connection pool slots will kill performance. This has nothing to do with your programming language.
Developer bandwidth: The most expensive resource in a startup. Spending 3 months optimizing memory usage while competitors ship features is the real waste.
AI tokens: You're paying for OpenAI/Claude tokens, not server RAM.

The main takeaways from Lovable's migration: they got to almost a $2 billion valuation with Python before having to move. They knew exactly what they needed to optimize and why - specific concurrency patterns that were causing real operational pain. They measured concrete improvements: deployment speed, request performance, infrastructure costs.

That's engineering-driven optimization based on actual constraints, not premature optimization based on theoretical performance concerns.

Most AI SaaS startups will never hit the scale where Python's resource usage becomes the bottleneck. And if you do hit that scale, congratulations - you have a successful business and can afford to hire engineers to optimize the parts that actually matter.

The Bottom Line

Python's slowness and higher resource usage are real. But for AI SaaS applications that spend 99% of their time waiting for external APIs and databases, these limitations don't matter.

The path to success isn't picking the "fastest" language - it's understanding your actual constraints and optimizing what matters. Build your MVP in Python. Measure where the real bottlenecks are. Fix those first. When you hit Python's actual limits (not theoretical ones), then consider alternatives.

The CEO I mentioned? His startup shipped their MVP in Python and is now getting customer traction. Meanwhile, his competitors who chose Rust for "performance reasons" without actually needing it might still be rewriting HTTP handlers.

Software engineering is about tradeoffs, not trends. Make decisions based on your actual constraints, not benchmark blog posts. Your users don't care if your backend runs on Python or Rust. They care if your product solves their problems.

Want to skip months of setup and ship your AI SaaS today? Check out FastroAI - a production-ready FastAPI template with authentication, payments, AI integration, and everything else you need to launch. Stop building infrastructure, start building solutions.

Originally posted at FastroAI.

How to Create Successful Open Source Projects

Igor Benav — Thu, 09 Jan 2025 00:00:58 +0000

FastCRUD — 1 year, 1k stars, 130k downloads later

Introduction

A little over a year ago, I started programming FastCRUD — Robust CRUD Operations and Flexible Endpoint Creation for FastAPI, my open source project.

It wasn’t the first open source project I created, nor was it my first library, but certainly some interesting changes happened in my thinking about how to approach projects since then. The idea here is to talk a bit about these reflections and the lessons I’ve learned, and perhaps get more of you interested in open source.

The general idea is to discuss how to create a successful open source project.

TL;DR:

Make something useful for yourself
Don’t start with “what to do,” start with “why do it”
Make the simplest possible thing that solves the problem
Documentation, documentation, documentation
Get some social proof
Promote your project
Take criticism
Maintain your project
Know how to ask for help

Context

I used to work primarily with data, going from Data Analyst to Analytics Engineer and Data Engineer, even deploying Machine Learning models, but I confess I really missed programming.

The truth is that when working with data, you don’t program as much, and when you do, it’s not that interesting. It’s mostly data pipelines, transformations (in SQL or Python), eventually some predictive modeling or machine learning deployment, but the programming part of these problems is kind of already solved. What really generates value in these things — unless you’re in huge companies where infrastructure matters a lot — is much more about understanding the business, optimizing queries, knowing how to answer questions (and which questions people are really asking), and eventually this becomes somewhat repetitive.

Missing my programming days from college and my (failed) startups, I decided to start creating open source projects to satisfy this craving.

What Makes a Successful Project?

Well, to start talking about how to succeed in a project, it’s important to define what success is. The truth is that this varies quite a bit, but after some time I moved my definition of success from “people seeing, using, starring” to “it’s useful to me.”

The only thing you can control about the project is whether it will be useful to you, not whether others will like it or whether it will be useful to them. If these things happen, great, but the reality is that after a while, you value these vanity metrics less.

Of course, this changes depending on your goal. If you’re doing the project to improve your portfolio, you’ll want to use relevant technologies for that and want it to have more visibility. Either way, take the maxim “will this be useful to me?” to heart. In the worst-case scenario, you haven’t wasted your time (beyond the learning).

Perhaps you’ll discover that for your specific problem, it’s better to contribute to an existing project than to make your own.

How to Come Up with Project Ideas?

Here we have another maxim: don’t start with “what to do,” start with “why do it.” There’s no way to make something useful for you (and, maybe, for others) without some domain knowledge. The most important thing is to know what problems exist in the area.

Practical example:

I needed to improve on deploying machine learning models for work, so I decided to learn FastAPI
After using FastAPI for a while, I saw that many things didn’t have a standard way of doing them, so I decided to create a template for myself to facilitate creating APIs with the technologies I liked
I discovered there was no admin panel for FastAPI the way I liked, so I decided to create one (for me to use)
To create this panel, I needed an easy way to generate CRUD and endpoints from SQLAlchemy models and Pydantic schemas, so I created FastCRUD to make my work easier

Notice how every project here had a natural motivation before? Some problem I needed to solve for myself (or for the company I worked for). Precisely because of this, even if no one had seen or used the projects, it would still have been a success for me.

As a bonus, if these projects solve a problem for me, the chance they solve problems for other people is greater (as was the case).

The general idea here is to start by thinking about solving a problem, not creating a project. Maybe you don’t know what that problem is yet, so before that, go study, explore, experience it firsthand.

How to Develop?

Now that you know what to do (and more importantly, why do it), you will actually start developing. The maxim of this section is very simple: make the simplest possible thing that solves the problem.

Start small, remember that your initial goal is to solve the problem, not to make the prettiest thing possible. Once you have an initial solution, you can add things and refactor (when relevant).

Practical example:

I wanted an easier way to filter for my admin panel, so I added django-style ORM filters to FastCRUD
People started asking for SQLModel support, so I added it
I was tired of reimplementing pagination every time, so I created utils for this in FastCRUD

An interesting thing is that at this point, people started finding it useful and using it, so they identified problems, started asking for things, and even contributing by opening pull requests.

It’s important to talk about how people found out about my project, which brings us to the next part.

Here things start to get more political. Although our initial definition of a successful open source project was something useful to ourselves, the truth is that the more people contributing, the better.

You probably won’t find all the bugs alone, won’t think of all the improvements on your own, and will also learn more slowly without other people stressing your idea.

What I learned (after getting beaten up a lot) is:

Documentation, documentation, documentation. A useful README about what the project is and how to use it is the minimum. The more you make it easier for others to understand things, the more likely they are to help. Get inspired by documentation you like, write, write, write.
Get about 10–20 stars before promoting. This is sad, but if you don’t ask for initial stars from friends, no one will look at your project. The idea isn’t even to gain stars after that, but people won’t look and take it seriously without some social validation. If you don’t have many friends to try out your project, it’s okay to promote without this.
Promote. It might seem obvious, but people are often afraid of this. The sooner you lose the fear of showing things (which certainly won’t be perfect), the better. Post here, on LinkedIn, on Reddit (without emojis, they hate them) and expect criticism, praise, tips. Don’t be annoying, don’t keep promoting all the time.
Take criticism. People will speak ill, it’s normal. Evaluate what is useful to you and improve, ignore what isn’t useful.

Originally published at https://www.benav.io on January 9, 2025.

Building a Local AI Task Planner with ClientAI and Ollama

Igor Benav — Tue, 17 Dec 2024 03:37:19 +0000

In this tutorial, we'll build an AI-powered task planner using ClientAI and Ollama. Our planner will break down goals into actionable tasks, create realistic timelines, and manage resources — all of this running in your own machine.

Our task planner will be capable of:

Breaking down goals into specific, actionable tasks
Creating realistic timelines with error handling
Managing and allocating resources effectively
Providing structured, formatted plans

For ClientAI's docs see here and for Github Repo, here.

Setting Up Our Environment

First, create a new directory for your project:

mkdir local_task_planner
cd local_task_planner

Install ClientAI with Ollama support:

pip install clientai[ollama]

Make sure you have Ollama installed on your system. You can get it from Ollama's website.

Create our main Python file:

touch task_planner.py

Let's start with our core imports:

from datetime import datetime, timedelta
from typing import Dict, List
import logging

from clientai import ClientAI
from clientai.agent import create_agent, tool
from clientai.ollama import OllamaManager

logger = logging.getLogger(__name__)

Each component plays a crucial role:

datetime: Helps us manage task timelines and scheduling
ClientAI: Provides our AI framework
OllamaManager: Manages our local AI model
Various utility modules for type hints and logging

Building the Task Planner Core

First, let's create our TaskPlanner class that will manage the AI interaction:

class TaskPlanner:
    """A local task planning system using Ollama."""

    def __init__(self):
        """Initialize the task planner with Ollama."""
        self.manager = OllamaManager()
        self.client = None
        self.planner = None

    def start(self):
        """Start the Ollama server and initialize the client."""
        self.manager.start()
        self.client = ClientAI("ollama", host="http://localhost:11434")

        self.planner = create_agent(
            client=self.client,
            role="task planner",
            system_prompt="""You are a practical task planner. Break down goals into
            specific, actionable tasks with realistic time estimates and resource needs.
            Use the tools provided to validate timelines and format plans properly.""",
            model="llama3",
            step="think",
            tools=[validate_timeline, format_plan],
            tool_confidence=0.8,
            stream=True,
        )

This class serves as our foundation. It manages the Ollama server lifecycle, creates and configures our AI client and sets up our planning agent with specific capabilities.

Creating Our Planning Tools

Now let's build the tools our AI will use. First, the timeline validator:

@tool(name="validate_timeline")
def validate_timeline(tasks: Dict[str, int]) -> Dict[str, dict]:
    """
    Validate time estimates and create a realistic timeline.

    Args:
        tasks: Dictionary of task names and estimated hours

    Returns:
        Dictionary with start dates and deadlines
    """
    try:
        current_date = datetime.now()
        timeline = {}
        accumulated_hours = 0

        for task, hours in tasks.items():
            try:
                hours_int = int(float(str(hours)))

                if hours_int <= 0:
                    logger.warning(f"Skipping task {task}: Invalid hours value {hours}")
                    continue

                days_needed = hours_int / 6
                start_date = current_date + timedelta(hours=accumulated_hours)
                end_date = start_date + timedelta(days=days_needed)

                timeline[task] = {
                    "start": start_date.strftime("%Y-%m-%d"),
                    "end": end_date.strftime("%Y-%m-%d"),
                    "hours": hours_int,
                }

                accumulated_hours += hours_int

            except (ValueError, TypeError) as e:
                logger.warning(f"Skipping task {task}: Invalid hours value {hours} - {e}")
                continue

        return timeline
    except Exception as e:
        logger.error(f"Error validating timeline: {str(e)}")
        return {}

This validator converts time estimates to working days, handles invalid inputs gracefully, creates realistic sequential scheduling and provides detailed logging for debugging.

Next, let's create our plan formatter:

@tool(name="format_plan")
def format_plan(
    tasks: List[str],
    timeline: Dict[str, dict],
    resources: List[str]
) -> str:
    """
    Format the plan in a clear, structured way.

    Args:
        tasks: List of tasks
        timeline: Timeline from validate_timeline
        resources: List of required resources

    Returns:
        Formatted plan as a string
    """
    try:
        plan = "== Project Plan ==\n\n"

        plan += "Tasks and Timeline:\n"
        for i, task in enumerate(tasks, 1):
            if task in timeline:
                t = timeline[task]
                plan += f"\n{i}. {task}\n"
                plan += f"   Start: {t['start']}\n"
                plan += f"   End: {t['end']}\n"
                plan += f"   Estimated Hours: {t['hours']}\n"

        plan += "\nRequired Resources:\n"
        for resource in resources:
            plan += f"- {resource}\n"

        return plan
    except Exception as e:
        logger.error(f"Error formatting plan: {str(e)}")
        return "Error: Unable to format plan"

Here we want to create a consistent, readable output with proper task numbering and organized timeline.

Building the Interface

Let's create a user-friendly interface for our planner:

def get_plan(self, goal: str) -> str:
    """
    Generate a plan for the given goal.

    Args:
        goal: The goal to plan for

    Returns:
        A formatted plan string
    """
    if not self.planner:
        raise RuntimeError("Planner not initialized. Call start() first.")

    return self.planner.run(goal)

def main():
    planner = TaskPlanner()

    try:
        print("Task Planner (Local AI)")
        print("Enter your goal, and I'll create a practical, timeline-based plan.")
        print("Type 'quit' to exit.")

        planner.start()

        while True:
            print("\n" + "=" * 50 + "\n")
            goal = input("Enter your goal: ")

            if goal.lower() == "quit":
                break

            try:
                plan = planner.get_plan(goal)
                print("\nYour Plan:\n")
                for chunk in plan:
                    print(chunk, end="", flush=True)
            except Exception as e:
                print(f"Error: {str(e)}")

    finally:
        planner.stop()

if __name__ == "__main__":
    main()

Our interface provides:

Clear user instructions
Real-time plan generation with streaming
Proper error handling
Clean shutdown management

Example Usage

Here's what you'll see when you run the planner:

Task Planner (Local AI)
Enter your goal, and I'll create a practical, timeline-based plan.
Type 'quit' to exit.

==================================================

Enter your goal: Create a personal portfolio website

Your Plan:

== Project Plan ==

Tasks and Timeline:
1. Requirements Analysis and Planning
   Start: 2024-12-08
   End: 2024-12-09
   Estimated Hours: 6

2. Design and Wireframing
   Start: 2024-12-09
   End: 2024-12-11
   Estimated Hours: 12

3. Content Creation
   Start: 2024-12-11
   End: 2024-12-12
   Estimated Hours: 8

4. Development
   Start: 2024-12-12
   End: 2024-12-15
   Estimated Hours: 20

Required Resources:
- Design software (e.g., Figma)
- Text editor or IDE
- Web hosting service
- Version control system

Future Improvements

Consider these enhancements for your own task planner:

Add dependency tracking between tasks
Include cost calculations for resources
Save plans to files or project management tools
Track progress against the original plan
Add validation for resource availability
Implement parallel task scheduling
Add support for recurring tasks
Include priority levels for tasks

To see more about ClientAI, go to the docs.

Connect With Me

If you have any questions about this tutorial or want to share your improvements to the task planner, feel free to reach out:

GitHub: igorbenav
X/Twitter: @igorbenav
LinkedIn: Igor

Building a Local AI Code Reviewer with ClientAI and Ollama - Part 2

Igor Benav — Tue, 17 Dec 2024 02:03:31 +0000

In Part 1, we built the core analysis tools for our code reviewer. Now we'll create an AI assistant that can use these tools effectively. We'll go through each component step by step, explaining how everything works together.

For ClientAI's docs see here and for Github Repo, here.

Series Index

Part 1: Introduction, Setup, Tool Creation
Part 2: Building the Assistant and Command Line Interface (you are here)

Registering Our Tools with ClientAI

First, we need to make our tools available to the AI system. Here's how we register them:

def create_review_tools() -> List[ToolConfig]:
    """Create the tool configurations for code review."""
    return [
        ToolConfig(
            tool=analyze_python_code,
            name="code_analyzer",
            description=(
                "Analyze Python code structure and complexity. "
                "Expects a 'code' parameter with the Python code as a string."
            ),
            scopes=["observe"],
        ),
        ToolConfig(
            tool=check_style_issues,
            name="style_checker",
            description=(
                "Check Python code style issues. "
                "Expects a 'code' parameter with the Python code as a string."
            ),
            scopes=["observe"],
        ),
        ToolConfig(
            tool=generate_docstring,
            name="docstring_generator",
            description=(
                "Generate docstring suggestions for Python code. "
                "Expects a 'code' parameter with the Python code as a string."
            ),
            scopes=["act"],
        ),
    ]

Let's break down what's happening here:

Each tool is wrapped in a ToolConfig object that tells ClientAI:
- tool: The actual function to call
- name: A unique identifier for the tool
- description: What the tool does and what parameters it expects
- scopes: When the tool can be used ("observe" for analysis, "act" for generation)
We classify our tools into two categories:
- "observe" tools (code_analyzer and style_checker) gather information
- "act" tools (docstring_generator) produce new content

Building the AI Assistant Class

Now let's create our AI assistant. We'll design it to work in steps, mimicking how a human code reviewer would think:

class CodeReviewAssistant(Agent):
    """An agent that performs comprehensive Python code review."""

    @observe(
        name="analyze_structure",
        description="Analyze code structure and style",
        stream=True,
    )
    def analyze_structure(self, code: str) -> str:
        """Analyze the code structure, complexity, and style issues."""
        self.context.state["code_to_analyze"] = code
        return """
        Please analyze this Python code structure and style:

        The code to analyze has been provided in the context as 'code_to_analyze'.
        Use the code_analyzer and style_checker tools to evaluate:
        1. Code complexity and structure metrics
        2. Style compliance issues
        3. Function and class organization
        4. Import usage patterns
        """

This first method is crucial:

The @observe decorator marks this as an observation step
stream=True enables real-time output
We store the code in the context to access it in later steps
The return string is a prompt that guides the AI in using our tools

Next, we add the improvement suggestion step:

    @think(
        name="suggest_improvements",
        description="Suggest code improvements based on analysis",
        stream=True,
    )
    def suggest_improvements(self, analysis_result: str) -> str:
        """Generate improvement suggestions based on the analysis results."""
        current_code = self.context.state.get("current_code", "")
        return f"""
        Based on the code analysis of:

        ```
{% endraw %}
python
        {current_code}
{% raw %}

        ```

        And the analysis results:
        {analysis_result}

        Please suggest specific improvements for:
        1. Reducing complexity where identified
        2. Fixing style issues
        3. Improving code organization
        4. Optimizing import usage
        5. Enhancing readability
        6. Enhancing explicitness
        """

This method:

Uses @think to indicate this is a reasoning step
Takes the analysis results as input
Retrieves the original code from context
Creates a structured prompt for improvement suggestions

The Command-Line Interface

Now let's create a user-friendly interface. We'll break this down into parts:

def main():
    # 1. Set up logging
    logger = logging.getLogger(__name__)

    # 2. Configure Ollama server
    config = OllamaServerConfig(
        host="127.0.0.1",  # Local machine
        port=11434,        # Default Ollama port
        gpu_layers=35,     # Adjust based on your GPU
        cpu_threads=8,     # Adjust based on your CPU
    )

This first part sets up error logging, configures the Ollama server with sensible defaults and allows customization of GPU and CPU usage.

Next, we create the AI client and assistant:

    # Use context manager for Ollama server
    with OllamaManager(config) as manager:
        # Initialize ClientAI with Ollama
        client = ClientAI(
            "ollama", 
            host=f"http://{config.host}:{config.port}"
        )

        # Create code review assistant with tools
        assistant = CodeReviewAssistant(
            client=client,
            default_model="llama3",
            tools=create_review_tools(),
            tool_confidence=0.8,  # How confident the AI should be before using tools
            max_tools_per_step=2, # Maximum tools to use per step
        )

Key points about this setup:

The context manager (with) ensures proper server cleanup
We connect to the local Ollama instance
The assistant is configured with:
- Our custom tools
- A confidence threshold for tool usage
- A limit on tools per step to prevent overuse

Finally, we create the interactive loop:

        print("Code Review Assistant (Local AI)")
        print("Enter Python code to review, or 'quit' to exit.")
        print("End input with '###' on a new line.")

        while True:
            try:
                print("\n" + "=" * 50 + "\n")
                print("Enter code:")

                # Collect code input
                code_lines = []
                while True:
                    line = input()
                    if line == "###":
                        break
                    code_lines.append(line)

                code = "\n".join(code_lines)
                if code.lower() == "quit":
                    break

                # Process the code
                result = assistant.run(code, stream=True)

                # Handle both streaming and non-streaming results
                if isinstance(result, str):
                    print(result)
                else:
                    for chunk in result:
                        print(chunk, end="", flush=True)
                print("\n")

            except Exception as e:
                logger.error(f"Unexpected error: {e}")
                print("\nAn unexpected error occurred. Please try again.")

This interface:

Collects multiline code input until seeing "###"
Handles both streaming and non-streaming output
Provides clean error handling
Allows easy exit with "quit"

And let's make it a script we're able to run:

if __name__ == "__main__":
    main()

Using the Assistant

Let's see how the assistant handles real code. Let's run it:

python code_analyzer.py

Here's an example with issues to find:

def calculate_total(values,tax_rate):
    Total = 0
    for Val in values:
        if Val > 0:
            if tax_rate > 0:
                Total += Val + (Val * tax_rate)
            else:
                Total += Val
    return Total

The assistant will analyze multiple aspects:

Structural Issues (nested if statements increasing complexity, missing type hints, no input validation)
Style Problems (inconsistent variable naming, missing spaces after commas, missing docstring)

Extension Ideas

Here are some ways to enhance the assistant:

Additional Analysis Tools
Enhanced Style Checking
Documentation Improvements
Auto-fixing Features

Each of these can be added by creating a new tool function, wrapping it in appropriate JSON formatting, adding it to the create_review_tools() function and then updating the assistant's prompts to use the new tool.

To see more about ClientAI, go to the docs.

Connect with Me

If you have any questions, want to discuss tech-related topics, or share your feedback, feel free to reach out to me on social media:

GitHub: igorbenav
X/Twitter: @igorbenav
LinkedIn: Igor

Building a Local AI Code Reviewer with ClientAI and Ollama

Igor Benav — Tue, 17 Dec 2024 01:59:10 +0000

Ever wanted your own AI-powered code reviewer that runs entirely on your local machine? In this two-part tutorial, we'll build exactly that using ClientAI and Ollama.

Our assistant will analyze Python code structure, identify potential issues, and suggest improvements — all while keeping your code private and secure.

For ClientAI's docs see here and for Github Repo, here.

Series Index

Part 1: Introduction, Setup, Tool Creation (you are here)
Part 2: Building the Assistant and Command Line Interface

Project Overview

Our code analysis assistant will be capable of:

Analyzing code structure and complexity
Identifying style issues and potential problems
Generating documentation suggestions
Providing actionable improvement recommendations

All of this will run locally on your machine, giving you the power of AI-assisted code review while maintaining complete privacy of your code.

Setting Up Our Environment

First, create a new directory for your project:

mkdir local_task_planner
cd local_task_planner

Install ClientAI with Ollama support:

pip install clientai[ollama]

Make sure you have Ollama installed on your system. You can get it from Ollama's website.

Now let's create the file we'll write the code into:

touch code_analyzer.py

And start with our core imports:

import ast
import json
import logging
import re
from dataclasses import dataclass
from typing import List
from clientai import ClientAI
from clientai.agent import (
    Agent,
    ToolConfig,
    act,
    observe,
    run,
    synthesize,
    think,
)
from clientai.ollama import OllamaManager, OllamaServerConfig

Each of these components plays a crucial role:

ast: Helps us understand Python code by parsing it into a tree structure
ClientAI: Provides our AI framework
Various utility modules for data handling and pattern matching

Structuring Our Analysis Results

When analyzing code, we need a clean way to organize our findings. Here's how we'll structure our results:

@dataclass
class CodeAnalysisResult:
    """Results from code analysis."""
    complexity: int
    functions: List[str]
    classes: List[str]
    imports: List[str]
    issues: List[str]

Think of this as our report card for code analysis:

Complexity score indicates how intricate the code is
Lists of functions and classes help us understand code structure
Imports show external dependencies
Issues track any problems we discover

Building the Core Analysis Engine

Now for the actual core — let's build our code analysis engine:

def analyze_python_code_original(code: str) -> CodeAnalysisResult:
    """Analyze Python code structure and complexity."""
    try:
        tree = ast.parse(code)
        functions = []
        classes = []
        imports = []
        complexity = 0
        for node in ast.walk(tree):
            if isinstance(node, ast.FunctionDef):
                functions.append(node.name)
                complexity += sum(
                    1
                    for _ in ast.walk(node)
                    if isinstance(_, (ast.If, ast.For, ast.While))
                )
            elif isinstance(node, ast.ClassDef):
                classes.append(node.name)
            elif isinstance(node, (ast.Import, ast.ImportFrom)):
                for name in node.names:
                    imports.append(name.name)
        return CodeAnalysisResult(
            complexity=complexity,
            functions=functions,
            classes=classes,
            imports=imports,
            issues=[],
        )
    except Exception as e:
        return CodeAnalysisResult(
            complexity=0, functions=[], classes=[], imports=[], issues=[str(e)]
        )

This function is like our code detective. It:

Parses code into a tree structure
Walks through the tree looking for functions, classes, and imports
Calculates complexity by counting control structures
Returns a comprehensive analysis result

Implementing Style Checking

Good code isn't just about working correctly — it should be readable and maintainable. Here's our style checker:

def check_style_issues_original(code: str) -> List[str]:
    """Check for Python code style issues."""
    issues = []
    for i, line in enumerate(code.split("\n"), 1):
        if len(line.strip()) > 88:
            issues.append(f"Line {i} exceeds 88 characters")
    function_pattern = r"def\s+([a-zA-Z_][a-zA-Z0-9_]*)\s*\("
    for match in re.finditer(function_pattern, code):
        name = match.group(1)
        if not name.islower():
            issues.append(f"Function '{name}' should use snake_case")
    return issues

Our style checker focuses on two key aspects:

Line length — ensuring code stays readable
Function naming conventions — enforcing Python's preferred snake_case style

Documentation Helper

Documentation is crucial for maintainable code. Here's our documentation generator:

def generate_docstring(code: str) -> str:
    """Generate docstring for Python code."""
    try:
        tree = ast.parse(code)
        for node in ast.walk(tree):
            if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
                args = []
                if isinstance(node, ast.FunctionDef):
                    args = [a.arg for a in node.args.args]
                return f"""
                Suggested docstring for {node.name}:
                Args:
                {chr(4).join(f"{arg}: Description of {arg}" for arg in args)}
                Returns:
                    Description of return value
                Examples:
                    ```
{% endraw %}
python
                    # Example usage of {node.name}
{% raw %}

                    ```
                """
        return "No functions or classes found to document."
    except Exception as e:
        return f"Error generating docstring: {str(e)}"

This helper:

Identifies functions and classes
Extracts parameter information
Generates documentation templates
Includes placeholders for examples

Making Our Tools AI-Ready

To prepare our tools for integration with the AI system, we need to wrap them in JSON-friendly formats:

def analyze_python_code(code: str) -> str:
    """Wrap analyze_python_code_original to return JSON string."""
    if not code:
        return json.dumps({"error": "No code provided"})
    result = analyze_python_code_original(code)
    return json.dumps({
        "complexity": result.complexity,
        "functions": result.functions,
        "classes": result.classes,
        "imports": result.imports,
        "issues": result.issues,
    })

def check_style_issues(code: str) -> str:
    """Wrap check_style_issues_original to return JSON string."""
    if not code:
        return json.dumps({"error": "No code provided"})
    issues = check_style_issues_original(code)
    return json.dumps({"issues": issues})

These wrappers add input validation, JSON serialization and error handling to make our assistant more error proof.

Coming Up in Part 2

In this post we set up our environment, structured our results, and built the functions we will use as tools for our Agent. In the next part, we'll actually create our AI assistant, register these tools, build a command-line interface and see this assistant in action.

Your next step is Part 2: Building the Assistant and Command Line Interface.

To see more about ClientAI, go to the docs.

Connect with Me

If you have any questions, want to discuss tech-related topics, or share your feedback, feel free to reach out to me on social media:

GitHub: igorbenav
X/Twitter: @igorbenav
LinkedIn: Igor

Prototyping FastAPI faster with FastCRUD

Igor Benav — Wed, 31 Jan 2024 16:18:56 +0000

You probably already know about FastAPI, but in case you don't - it's a modern python web framework built on top of Starlette and Pydantic with cool features like Async support and auto generated swagger documentation.

Creating a minimal FastAPI app is as simple as:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

FastAPI gained a lot of popularity by being fast compared to other python web frameworks, having modern features, and great documentation, but it is not a batteries included framework like Django so you need to use other libraries for stuff like database operations.

Let's create a more functional FastAPI application with database interaction using SQLModel. We start by actually creating a connection to the database:

FastAPI Basic Setup

# database.py

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)

async_session = sessionmaker(
  engine, 
  class_=AsyncSession, 
  expire_on_commit=False
)

And creating a function to actually get this database session

# database.py
...
async def async_get_db() -> AsyncSession:
    async_session = local_session

    async with async_session() as db:
        yield db
        await db.commit()

Now we can finally create our database models. I'll keep it simple, a user can create posts and that's it. Posts have a title and the actual text.

# models.py

from sqlmodel import Field, SQLModel, Relationship
from typing import Optional, List
from datetime import datetime

class User(SQLModel, table=True):
    __tablename__ = "user"

    id: Optional[int] = Field(default=None, primary_key=True, nullable=False, index=True)
    name: str = Field(max_length=30)
    username: str = Field(max_length=20, index=True, unique=True)

    posts: List["Post"] = Relationship(back_populates="creator")

class Post(SQLModel, table=True):
    __tablename__ = "post"

    id: Optional[int] = Field(default=None, primary_key=True, nullable=False, index=True)
    created_by_user_id: int = Field(default=None, foreign_key="user.id")
    title: str = Field(max_length=30)
    text: str = Field(max_length=63206)

    creator: User = Relationship(back_populates="posts")

Now let's create the schemas that we'll use to validate our input and output:

# user_schemas.py

from typing import List, Optional
from sqlmodel import Field, SQLModel

# Shared properties
class UserBase(SQLModel):
    name: Optional[str] = None
    username: Optional[str] = None

# Properties to receive via API on creation
class UserCreate(UserBase):
    pass  

# Properties to receive via API on update
class UserUpdate(UserBase):
    pass

# Properties to return to client
class UserRead(UserBase):
    id: Optional[int] = None
    posts: List["PostRead"] = []

You should similarly create the schemas for posts, but I'm skipping it since it's similar.

Now we'll finally create our FastAPI app:

# main.py

from .database.py import SQLModel
from fastapi import FastAPI

# We want to create the tables on start, so let's create this function
async def create_tables() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

app = FastAPI()

# now adding the event handler:
app.add_event_handler("startup", create_tables)

Creating Endpoints

We finally have our basic setup done, now let's create the user endpoints for our app, you'll write similar code for each model you add to the api.

# user_endpoints.py

import fastapi
from fastapi import HTTPException, Depends, status
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel import select
from typing import List, Optional

from .models import User
from .user_schemas import UserCreate, UserRead, UserUpdate
from .database.py import async_get_db

router = fastapi.APIRouter(tags=["Users"])

@app.post("/users/", response_model=UserRead)
async def create_user(user: UserCreate, session: AsyncSession = Depends(get_session)):
    db_user = User.from_orm(user)
    session.add(db_user)
    await session.commit()
    await session.refresh(db_user)
    return db_user

@app.get("/users/", response_model=List[UserRead])
async def read_users(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User))
    users = result.scalars().all()
    return users

@app.get("/users/{user_id}", response_model=UserRead)
async def read_user(user_id: int, session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.patch("/users/{user_id}", response_model=UserRead)
async def update_user(user_id: int, user: UserUpdate, session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User).where(User.id == user_id))
    db_user = result.scalar_one_or_none()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    user_data = user.dict(exclude_unset=True)
    for key, value in user_data.items():
        setattr(db_user, key, value)
    session.add(db_user)
    await session.commit()
    await session.refresh(db_user)
    return db_user

This works, but a lot of the time you are just writing the same code again and again. You have the session transactions, filters, updating fields, committing changes to the database…

Creating Endpoints with FastCRUD

Enter FastCRUD - a package for FastAPI, offering robust async CRUD operations and flexible endpoint creation utilities.
The same code above using FastCRUD would be written as:

# user_endpoints.py

import fastapi
from fastcrud import FastCRUD
from fastcrud.exceptions.http_exceptions import (
    DuplicateValueException, 
    NotFoundException
)

router = fastapi.APIRouter(tags=["Users"])

from .models import User
from .user_schemas import UserCreate, UserRead, UserUpdate
from .database.py import async_get_db

crud_user = FastCRUD(User)

@app.post("/users/", response_model=UserRead)
async def create_user(
    user: UserCreate, 
    db: AsyncSession = Depends(get_session)
):
    username_row = await crud_user.exists(db=db, username=user.username)
    if username_row:
        raise DuplicateValueException("Username not available")
    return await crud_user.create(db=db, object=user)

@app.get("/users/", response_model=List[UserRead])
async def read_users(
    db: AsyncSession = Depends(get_session), 
    page: int = 1, 
    items_per_page: int = 10
):
    return await crud_user.get_multi(
        db=db,
        offset=((page - 1) * items_per_page),
        limit=items_per_page,
        schema_to_select=UserRead
    )

@app.get("/users/{user_id}", response_model=UserRead)
async def read_user(
    user_id: int, 
    db: AsyncSession = Depends(get_session)
):
    db_user = await crud_user.get(
        db=db, 
        schema_to_select=UserRead, 
        id=user_id
    )
    if db_user is None:
        raise NotFoundException("User not found")

    return db_user

@app.patch("/users/{user_id}", response_model=UserRead)
async def update_user(
    user_id: int, 
    user: UserUpdate, 
    db: AsyncSession = Depends(get_session)
    ):
    db_user = await crud_users.get(
        db=db, 
        schema_to_select=UserRead, 
        username=username
    )
    if db_user is None:
        raise NotFoundException("User not found")

    return await crud_user.update(db=db, object=user, id=user_id)

When you instantiate FastCRUD with a Model, you automatically get a bunch of methods for robust database interaction. A complete list of the methods and how they work here, but the most common are:

Create
Get
Exists
Count
Get Multi
Update

Automatic Endpoints

Sometimes, even this is not quick enough. When you are just prototyping or at a hackathon, you really want to do this as fast as you can. That's why FastCRUD provides crud_router, a way to instantly have your endpoints ready for prototyping:

# user_endpoints.py

from .models import User
from .user_schemas import UserCreate, UserRead, UserUpdate
from .database.py import async_get_db

from fastcrud import FastCRUD, crud_router

# CRUD router setup
user_router = crud_router(
    session=async_session,
    model=User,
    crud=FastCRUD(User),
    create_schema=UserCreate,
    update_schema=UserUpdate,
    tags=["Users"]
)

Let's add this router to our application:

# main.py

# now also importing our user_router
from .user_endpoints import user_router
from .database.py import SQLModel

from fastapi import FastAPI

async def create_tables() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

app = FastAPI()
app.add_event_handler("startup", create_tables)

# including the user_router
app.include_router(user_router)

And that's it. As simple as that, you may go to /docs and you'll see the created CRUD endpoints. If you have a more custom usage, just check the docs.

Full disclaimer: I'm the creator and maintainer of FastCRUD, I am sharing it because it's a solution for a problem of mine that I think a lot of people could benefit from.

FAQ

Q1: Where Can I read more about FastCRUD?
Just go to the docs: https://igorbenav.github.io/fastcrud/

Q2: Supported ORMs
Currently, FastCRUD works with SQLAlchemy 2.0 or greater and SQLModel 0.14 or greater.

Q3: Where can I find support or contribute to FastCRUD?
If you have any question or issue, just head to the Github repo and open an issue.

Connect With Me

If you have any questions, want to discuss tech-related stuff or simply share your feedback, feel free to reach me on social media

Github: igorbenav
X: igorbenav
Linkedin: Igor