Migration from FastAPI

Migrating from FastAPI to Catzilla is designed to be seamless. This guide shows you exactly how to migrate your existing FastAPI applications to get better performance with minimal code changes.

Why Migrate to Catzilla?

Performance Gains

  • Significantly faster request handling than FastAPI

  • C-accelerated routing with O(log n) lookup

  • Zero-allocation middleware system

  • Optimized memory usage with jemalloc

Enhanced Features

  • True async/sync hybrid support (FastAPI is async-only)

  • Advanced dependency injection with multiple scopes

  • Built-in multi-layer caching

  • Background task system with monitoring

  • Superior file handling and streaming

Better Developer Experience

  • Faster startup times

  • Better error messages

  • Enhanced debugging tools

  • Production-ready out of the box

Quick Migration Checklist

Note

Migration Time: 5-15 minutes for most applications

Most FastAPI applications can be migrated by changing just the import statements!

  1. Update imports - Change fastapi to catzilla

  2. Update app initialization - FastAPI() → Catzilla()

  3. Update response imports - Use Catzilla’s response classes

  4. Test your application - Everything should work immediately

  5. Optimize for performance - Add async handlers where beneficial

Step-by-Step Migration

1. Install Catzilla

pip install catzilla

2. Update Imports

Before (FastAPI):

from fastapi import FastAPI, Depends, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

After (Catzilla):

from catzilla import Catzilla, Depends, JSONResponse
from catzilla import BaseModel, Field
# Note: Use JSONResponse with status codes instead of HTTPException

3. Update App Initialization

Before (FastAPI):

from fastapi import FastAPI

app = FastAPI(
    title="My API",
    description="My FastAPI application",
    version="1.0.0"
)

After (Catzilla):

from catzilla import Catzilla

app = Catzilla(
    title="My API",
    description="My Catzilla application",
    version="1.0.0",
    production=False,    # Enable dev features
    show_banner=True     # Show startup banner
)

4. Update Response Imports

Before (FastAPI):

from fastapi.responses import JSONResponse, HTMLResponse, FileResponse

After (Catzilla):

from catzilla import JSONResponse, HTMLResponse
# Note: FileResponse not available in current Catzilla v0.2.0
# Use Response with appropriate headers for file serving

Migration Examples

Basic CRUD API

Here’s a complete migration example of a typical FastAPI CRUD application:

FastAPI Version:

from fastapi import FastAPI, HTTPException, Depends
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class User(BaseModel):
    name: str
    email: str
    age: Optional[int] = None

users_db = {}
user_id_counter = 1

@app.get("/users", response_model=List[User])
async def get_users():
    return list(users_db.values())

@app.post("/users", response_model=User)
async def create_user(user: User):
    global user_id_counter
    user_data = {
        "id": user_id_counter,
        "name": user.name,
        "email": user.email,
        "age": user.age
    }
    users_db[user_id_counter] = user_data
    user_id_counter += 1
    return user_data

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    return users_db[user_id]

Catzilla Version (Direct Migration):

from catzilla import Catzilla, Depends, JSONResponse
from catzilla import BaseModel
from typing import List, Optional

app = Catzilla()

class User(BaseModel):
    name: str
    email: str
    age: Optional[int] = None

users_db = {}
user_id_counter = 1

@app.get("/users")
async def get_users(request):
    return JSONResponse(list(users_db.values()))

@app.post("/users")
async def create_user(request, user: User):
    global user_id_counter
    user_data = {
        "id": user_id_counter,
        "name": user.name,
        "email": user.email,
        "age": user.age
    }
    users_db[user_id_counter] = user_data
    user_id_counter += 1
    return JSONResponse(user_data, status_code=201)

@app.get("/users/{user_id}")
async def get_user(request, user_id: int):
    if user_id not in users_db:
        return JSONResponse({"error": "User not found"}, status_code=404)
    return JSONResponse(users_db[user_id])

if __name__ == "__main__":
    app.listen(port=8000)

Catzilla Version (Optimized with Async/Sync Hybrid):

from catzilla import Catzilla, Path, JSONResponse
from catzilla import BaseModel, Field
from typing import List, Optional
import asyncio

app = Catzilla(production=False, show_banner=True)

class User(BaseModel):
    name: str = Field(min_length=2, max_length=50)
    email: str = Field(regex=r'^[^@]+@[^@]+\.[^@]+$')
    age: Optional[int] = Field(None, ge=0, le=120)

users_db = {}
user_id_counter = 1

# Sync handler for simple operations
@app.get("/users")
def get_users(request):
    return JSONResponse(list(users_db.values()))

# Async handler for database operations
@app.post("/users")
async def create_user(request, user: User):
    global user_id_counter

    # Simulate async database insert
    await asyncio.sleep(0.01)

    user_data = {
        "id": user_id_counter,
        "name": user.name,
        "email": user.email,
        "age": user.age
    }
    users_db[user_id_counter] = user_data
    user_id_counter += 1

    return JSONResponse(user_data, status_code=201)

# Sync handler with validation
@app.get("/users/{user_id}")
def get_user(request, user_id: int = Path(..., ge=1)):
    if user_id not in users_db:
        return JSONResponse(
            {"error": "User not found"},
            status_code=404
        )
    return JSONResponse(users_db[user_id])

if __name__ == "__main__":
    app.listen(port=8000)

Authentication & Dependencies

FastAPI Version:

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

app = FastAPI()
security = HTTPBearer()

def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    # Validate JWT token
    if not validate_token(credentials.credentials):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )
    return get_user_from_token(credentials.credentials)

@app.get("/protected")
async def protected_route(current_user = Depends(get_current_user)):
    return {"user": current_user}

Catzilla Version:

from catzilla import Catzilla, service, JSONResponse, Header, Depends
from catzilla.dependency_injection import set_default_container
from typing import Optional

app = Catzilla(enable_di=True)
set_default_container(app.di_container)

# Define the authentication service
@service("auth_service")
class AuthenticationService:
    def __init__(self):
        self.users_db = {
            "admin": {"id": 1, "username": "admin", "email": "admin@example.com"},
            "user1": {"id": 2, "username": "user1", "email": "user1@example.com"}
        }

    def authenticate_token(self, token: str) -> Optional[dict]:
        """Validate token and return user info"""
        # Simple mock authentication - in reality you'd validate JWT tokens
        if token == "valid_token":
            return self.users_db["admin"]
        elif token == "user_token":
            return self.users_db["user1"]
        return None

    def get_current_user(self, authorization: str) -> dict:
        """Extract and validate user from authorization header"""
        if not authorization.startswith("Bearer "):
            raise ValueError("Invalid authorization header format")

        token = authorization.replace("Bearer ", "")
        user = self.authenticate_token(token)

        if not user:
            raise ValueError("Invalid or expired token")

        return user

@app.get("/protected")
def protected_route(
    request,
    authorization: str = Header(..., description="Authorization header"),
    auth_service: AuthenticationService = Depends("auth_service")
):
    """Protected route that requires authentication"""
    current_user = auth_service.get_current_user(authorization)
    return JSONResponse({
        "message": f"Hello {current_user['username']}!",
        "user_info": current_user
    })

if __name__ == "__main__":
    app.listen(port=8000)

File Uploads

FastAPI Version:

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    content = await file.read()
    return {"filename": file.filename, "size": len(content)}

Catzilla Version:

from catzilla import Catzilla, JSONResponse, UploadFile, File, Form
from typing import List

app = Catzilla()

# Single file upload
@app.post("/upload")
async def upload_file(request, file: UploadFile = File(max_size="50MB")):
    # Catzilla provides optimized file handling
    content = await file.read()

    return JSONResponse({
        "filename": file.filename,
        "size": len(content),
        "content_type": file.content_type
    })

# Multiple file upload
@app.post("/upload/multiple")
def upload_multiple_files(
    request,
    files: List[UploadFile] = File(max_files=10, max_size="50MB"),
    category: str = Form("other")
):
    results = []
    for file in files:
        results.append({
            "filename": file.filename,
            "size": file.size,
            "content_type": file.content_type
        })

    return JSONResponse({
        "uploaded_files": len(files),
        "category": category,
        "files": results
    })

# Sync version for simple file handling
@app.post("/upload-sync")
def upload_file_sync(request, file: UploadFile = File(max_size="50MB")):
    return JSONResponse({
        "filename": file.filename,
        "size": file.size,
        "upload_method": "sync"
    })

if __name__ == "__main__":
    app.listen(port=8000)

Key Differences & Improvements

Request Object

FastAPI: Request object is optional and imported separately Catzilla: Request object is always the first parameter

# FastAPI
from fastapi import Request

@app.get("/")
async def handler(request: Request):
    pass

# Catzilla - request is always first parameter
@app.get("/")
def handler(request):
    pass

Async/Sync Handling

FastAPI: All handlers must be async Catzilla: Mix async and sync handlers freely

# FastAPI - everything must be async
@app.get("/sync-task")
async def sync_task():
    # Even CPU-bound tasks must be async
    return compute_something()

# Catzilla - use the right tool for the job
@app.get("/sync-task")
def sync_task(request):
    # CPU-bound tasks can be sync
    return JSONResponse(compute_something())

@app.get("/async-task")
async def async_task(request):
    # I/O-bound tasks can be async
    data = await fetch_from_database()
    return JSONResponse(data)

Dependency Injection

FastAPI: Basic dependency injection Catzilla: Advanced DI with scopes and service management

# FastAPI - basic dependencies
def get_database():
    return DatabaseConnection()

@app.get("/")
async def handler(db = Depends(get_database)):
    pass

# Catzilla - advanced DI with scopes
from catzilla import Catzilla, service, Depends, JSONResponse
from catzilla.dependency_injection import set_default_container

app = Catzilla(enable_di=True)
set_default_container(app.di_container)

@service("database", scope="singleton")
class DatabaseConnection:
    def __init__(self):
        self.connection_id = "db_12345"
        print("Database connection created")

    def query(self, sql: str):
        return f"Result for: {sql}"

@app.get("/data")
def get_data(request, db: DatabaseConnection = Depends("database")):
    result = db.query("SELECT * FROM users")
    return JSONResponse({"data": result, "connection": db.connection_id})

if __name__ == "__main__":
    app.listen(port=8000)

Performance Optimizations

FastAPI: Manual optimization required Catzilla: Automatic optimizations built-in

# Catzilla automatically provides:
# - C-accelerated routing
# - Optimal async/sync execution
# - Memory-efficient request handling
# - Built-in caching
# - Connection pooling

Common Migration Issues & Solutions

1. Response Model Decorators

Issue: FastAPI’s response_model parameter doesn’t exist in Catzilla

Solution: Use explicit JSON serialization

# FastAPI
@app.get("/users", response_model=List[User])
async def get_users():
    return users

# Catzilla
@app.get("/users")
def get_users(request):
    return JSONResponse([{
        "name": user.name,
        "email": user.email,
        "age": user.age
    } for user in users])

2. Status Code Responses

Issue: FastAPI’s automatic status code handling

Solution: Use explicit status codes in JSONResponse

# FastAPI
@app.post("/users", status_code=201)
async def create_user(user: User):
    return user

# Catzilla
@app.post("/users")
def create_user(request, user: User):
    return JSONResponse({
        "name": user.name,
        "email": user.email,
        "age": user.age
    }, status_code=201)

3. Background Tasks

Issue: FastAPI’s BackgroundTasks

Solution: Use Catzilla’s advanced background task system

# FastAPI
from fastapi import BackgroundTasks

@app.post("/send-email")
async def send_email(background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email_task, "user@example.com")

# Catzilla
app = Catzilla()

def send_email_task(email: str):
    """Background task function"""
    print(f"Sending email to {email}")
    # Email sending logic here

@app.post("/send-email")
def send_email(request):
    # Schedule background task using app.add_task()
    app.add_task(send_email_task, "user@example.com")
    return JSONResponse({"message": "Email scheduled"})

if __name__ == "__main__":
    app.listen(port=8000)

Migration Testing

After migration, test your application thoroughly:

# Install testing dependencies
pip install pytest httpx

# Run your existing FastAPI tests
# Most should work with minimal changes

# test_migration.py
import pytest
from httpx import AsyncClient
from your_app import app

@pytest.mark.asyncio
async def test_get_users():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/users")
        assert response.status_code == 200

Performance Verification

Verify the performance improvements after migration:

# Add this endpoint to your migrated app
@app.get("/performance-info")
def performance_info(request):
    from catzilla import get_performance_stats

    return JSONResponse({
        "framework": "Catzilla",
        "performance_vs_fastapi": "significantly faster",
        "router": "C-accelerated",
        "async_support": "hybrid",
        "stats": get_performance_stats()
    })

Benchmark your migrated application:

# Install benchmarking tools
pip install wrk

# Benchmark your FastAPI app
wrk -t12 -c400 -d30s http://localhost:8000/users

# Benchmark your Catzilla app
wrk -t12 -c400 -d30s http://localhost:8000/users

# Compare the results!

Advanced Migration Tips

  1. Gradual Migration - Migrate one route at a time - Use feature flags to switch between versions - Run both versions in parallel during testing

  2. Optimize After Migration - Convert CPU-bound endpoints to sync handlers - Use async handlers for I/O-bound operations - Implement caching where appropriate

  3. Take Advantage of New Features - Use dependency injection for better service management - Implement background tasks for long-running operations - Add monitoring and health checks

  4. Production Considerations - Update deployment scripts - Update monitoring and logging - Test under production load

Migration Checklist

□ Install Catzilla: pip install catzilla
□ Update imports: fastapi → catzilla
□ Update app initialization: FastAPI() → Catzilla()
□ Update response imports
□ Test basic functionality
□ Run existing test suite
□ Optimize async/sync handlers
□ Add performance monitoring
□ Benchmark performance improvements
□ Deploy to staging environment
□ Monitor production metrics
□ Celebrate significant performance improvement! 🚀

Key API Differences

App Startup

FastAPI can run with uvicorn, but Catzilla requires explicit app.listen():

# ❌ FastAPI pattern (Catzilla doesn't support uvicorn):
# uvicorn main:app --reload

# ✅ Catzilla pattern - always required:
if __name__ == "__main__":
    app.listen(port=8000)

Error Handling

FastAPI uses HTTPException, but Catzilla uses JSONResponse with status codes:

# ❌ FastAPI pattern (not available in Catzilla):
raise HTTPException(status_code=404, detail="User not found")

# ✅ Catzilla pattern:
return JSONResponse({"error": "User not found"}, status_code=404)

File Responses

FileResponse is not available in current Catzilla v0.2.0. Use Response with file content:

# ❌ Not available:
return FileResponse("file.pdf")

# ✅ Catzilla pattern:
with open("file.pdf", 'rb') as f:
    content = f.read()
return Response(
    body=content,
    content_type="application/pdf",
    headers={"Content-Length": str(len(content))}
)

Need Help?

If you encounter issues during migration:

  • 📖 Documentation: Check this guide and API reference

  • 🐛 Issues: Report migration issues on GitHub

  • 💬 Community: Ask questions in GitHub Discussions

  • 📧 Support: Contact the Catzilla team for enterprise support

Migration time: 5-15 minutes Performance gain: Significantly faster Code changes: Minimal

Welcome to the Catzilla family! 🚀