Python Performance Tips in Backend Development

# Basic profiling with cProfile
import cProfile
import pstats
from pstats import SortKey

def analyze_performance():
    profiler = cProfile.Profile()
    profiler.enable()
    
    # Your application code here
    result = expensive_operation()
    
    profiler.disable()
    stats = pstats.Stats(profiler)
    stats.sort_stats(SortKey.TIME)
    stats.print_stats(10)  # Top 10 time-consuming functions
    
    return result

# Memory profiling with tracemalloc
import tracemalloc

def memory_profile():
    tracemalloc.start()
    
    # Your code here
    data = process_large_dataset()
    
    current, peak = tracemalloc.get_traced_memory()
    print(f"Current memory usage: {current / 1024 / 1024:.1f} MB")
    print(f"Peak memory usage: {peak / 1024 / 1024:.1f} MB")
    tracemalloc.stop()

# Performance comparison: List vs Set vs Dict
import time

# Slow: O(n) lookup in list
permissions_list = ['read', 'write', 'delete', 'admin'] * 1000
start = time.time()
for _ in range(10000):
    'admin' in permissions_list
list_time = time.time() - start

# Fast: O(1) lookup in set
permissions_set = set(permissions_list)
start = time.time()
for _ in range(10000):
    'admin' in permissions_set
set_time = time.time() - start

print(f"List lookup: {list_time:.4f}s")
print(f"Set lookup: {set_time:.4f}s")
print(f"Set is {list_time/set_time:.1f}x faster!")

# Use collections.deque for frequent insertions/deletions at both ends
from collections import deque

# Efficient queue operations
task_queue = deque()
task_queue.appendleft("high_priority_task")  # O(1)
task_queue.append("normal_task")             # O(1)
next_task = task_queue.popleft()             # O(1)

import asyncio
import aiohttp
import time

# Synchronous approach (slow)
def sync_fetch_data(urls):
    results = []
    for url in urls:
        # Simulate API call
        time.sleep(0.1)  # 100ms delay
        results.append(f"Data from {url}")
    return results

# Asynchronous approach (fast)
async def async_fetch_data(session, url):
    # Simulate async API call
    await asyncio.sleep(0.1)
    return f"Data from {url}"

async def fetch_all_data(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [async_fetch_data(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    return results

# Performance comparison
urls = [f"https://api.example.com/{i}" for i in range(10)]

# Sync: 10 * 0.1s = 1.0s
start = time.time()
sync_results = sync_fetch_data(urls)
sync_time = time.time() - start

# Async: ~0.1s (concurrent execution)
start = time.time()
async_results = asyncio.run(fetch_all_data(urls))
async_time = time.time() - start

print(f"Sync time: {sync_time:.2f}s")
print(f"Async time: {async_time:.2f}s")
print(f"Speedup: {sync_time/async_time:.1f}x faster!")

from functools import lru_cache
import redis
import json
from typing import Optional

# In-memory caching with LRU
@lru_cache(maxsize=128)
def expensive_computation(n: int) -> int:
    """Cache results of expensive computations"""
    print(f"Computing for {n}...")  # Only prints on cache miss
    return sum(i * i for i in range(n))

# Redis caching for distributed systems
class RedisCache:
    def __init__(self, host='localhost', port=6379, db=0):
        self.redis_client = redis.Redis(host=host, port=port, db=db)
    
    def get(self, key: str) -> Optional[dict]:
        """Get cached value"""
        cached = self.redis_client.get(key)
        return json.loads(cached) if cached else None
    
    def set(self, key: str, value: dict, ttl: int = 3600):
        """Set cached value with TTL"""
        self.redis_client.setex(
            key, 
            ttl, 
            json.dumps(value, default=str)
        )
    
    def invalidate(self, pattern: str):
        """Invalidate cache by pattern"""
        keys = self.redis_client.keys(pattern)
        if keys:
            self.redis_client.delete(*keys)

# Usage example
cache = RedisCache()

def get_user_profile(user_id: int) -> dict:
    cache_key = f"user_profile:{user_id}"
    
    # Try cache first
    cached_profile = cache.get(cache_key)
    if cached_profile:
        return cached_profile
    
    # Fetch from database
    profile = fetch_user_from_db(user_id)
    
    # Cache for 1 hour
    cache.set(cache_key, profile, ttl=3600)
    return profile

def update_user_profile(user_id: int, data: dict):
    # Update database
    update_user_in_db(user_id, data)
    
    # Invalidate related caches
    cache.invalidate(f"user_profile:{user_id}")
    cache.invalidate(f"user_posts:{user_id}*")

# settings.py - Database optimization
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'your_db',
        'USER': 'your_user',
        'PASSWORD': 'your_password',
        'HOST': 'localhost',
        'PORT': '5432',
        'CONN_MAX_AGE': 600,  # Connection pooling
        'OPTIONS': {
            'MAX_CONNS': 20,
            'MIN_CONNS': 5,
            'sslmode': 'require',
        },
    }
}

# Enable query logging in development
LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
        },
    },
    'loggers': {
        'django.db.backends': {
            'handlers': ['console'],
            'level': 'DEBUG',
            'propagate': False,
        },
    },
}

# models.py
from django.db import models

class Author(models.Model):
    name = models.CharField(max_length=100, db_index=True)
    email = models.EmailField(unique=True)
    created_at = models.DateTimeField(auto_now_add=True)
    
    class Meta:
        indexes = [
            models.Index(fields=['name', 'created_at']),
        ]

class Book(models.Model):
    title = models.CharField(max_length=200, db_index=True)
    author = models.ForeignKey(Author, on_delete=models.CASCADE)
    published_date = models.DateField()
    isbn = models.CharField(max_length=13, unique=True)
    
    class Meta:
        indexes = [
            models.Index(fields=['published_date']),
            models.Index(fields=['author', 'published_date']),
        ]

# BAD: N+1 queries (1 + N queries)
def get_books_slow():
    books = Book.objects.all()  # 1 query
    for book in books:
        print(f"{book.title} by {book.author.name}")  # N queries

# GOOD: select_related for ForeignKey (2 queries total)
def get_books_optimized():
    books = Book.objects.select_related('author').all()  # 1 query with JOIN
    for book in books:
        print(f"{book.title} by {book.author.name}")  # No additional queries

# GOOD: prefetch_related for reverse ForeignKey/ManyToMany
def get_authors_with_books():
    authors = Author.objects.prefetch_related('book_set').all()
    for author in authors:
        books = author.book_set.all()  # No additional queries
        print(f"{author.name}: {books.count()} books")

# ADVANCED: Custom prefetch with filtering
from django.db.models import Prefetch

def get_authors_with_recent_books():
    recent_books = Book.objects.filter(
        published_date__year__gte=2020
    ).select_related('author')
    
    authors = Author.objects.prefetch_related(
        Prefetch('book_set', queryset=recent_books)
    ).all()

# Use annotations for aggregations
from django.db.models import Count, Avg

def get_author_stats():
    return Author.objects.annotate(
        book_count=Count('book'),
        avg_year=Avg('book__published_date__year')
    ).filter(book_count__gt=0)

# Use only() and defer() for large models
def get_book_titles_only():
    return Book.objects.only('title', 'author__name').select_related('author')

def get_books_without_content():
    return Book.objects.defer('content', 'summary').all()

from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import selectinload
from sqlalchemy import select
import asyncio
import aioredis
from typing import List, Optional

app = FastAPI(title="High Performance API", version="1.0.0")

# Async database setup
async_engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/db",
    pool_size=20,
    max_overflow=0,
    pool_pre_ping=True,
    pool_recycle=3600,
)

# Dependency for database sessions
async def get_db() -> AsyncSession:
    async with AsyncSession(async_engine) as session:
        try:
            yield session
        finally:
            await session.close()

# Redis connection pool
redis_pool = None

@app.on_event("startup")
async def startup_event():
    global redis_pool
    redis_pool = aioredis.ConnectionPool.from_url(
        "redis://localhost", 
        max_connections=20
    )

# Optimized endpoint with concurrent operations
@app.get("/users/{user_id}/dashboard")
async def get_user_dashboard(
    user_id: int,
    db: AsyncSession = Depends(get_db)
):
    # Concurrent database queries
    user_task = asyncio.create_task(get_user_with_profile(db, user_id))
    posts_task = asyncio.create_task(get_user_posts(db, user_id))
    stats_task = asyncio.create_task(get_user_stats(db, user_id))
    
    # Concurrent external API calls
    weather_task = asyncio.create_task(get_weather_data())
    notifications_task = asyncio.create_task(get_notifications(user_id))
    
    # Wait for all operations to complete
    user, posts, stats, weather, notifications = await asyncio.gather(
        user_task, posts_task, stats_task, weather_task, notifications_task
    )
    
    return {
        "user": user,
        "posts": posts,
        "stats": stats,
        "weather": weather,
        "notifications": notifications
    }

async def get_user_with_profile(db: AsyncSession, user_id: int):
    stmt = select(User).options(selectinload(User.profile)).where(User.id == user_id)
    result = await db.execute(stmt)
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

async def get_user_posts(db: AsyncSession, user_id: int, limit: int = 10):
    stmt = (
        select(Post)
        .where(Post.user_id == user_id)
        .order_by(Post.created_at.desc())
        .limit(limit)
    )
    result = await db.execute(stmt)
    return result.scalars().all()

# Background tasks for non-blocking operations
@app.post("/users/{user_id}/send-email")
async def send_user_email(
    user_id: int,
    email_data: EmailSchema,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db)
):
    # Immediate response
    user = await get_user_with_profile(db, user_id)
    
    # Queue email sending in background
    background_tasks.add_task(
        send_email_async, 
        user.email, 
        email_data.subject, 
        email_data.body
    )
    
    return {"message": "Email queued for sending", "user_id": user_id}

async def send_email_async(email: str, subject: str, body: str):
    # Simulate email sending
    await asyncio.sleep(2)
    print(f"Email sent to {email}: {subject}")

from flask import Flask, request, jsonify
from flask_caching import Cache
from flask_sqlalchemy import SQLAlchemy
from werkzeug.exceptions import NotFound
import redis
import json
from functools import wraps

app = Flask(__name__)

# Configure caching
app.config['CACHE_TYPE'] = 'RedisCache'
app.config['CACHE_REDIS_URL'] = 'redis://localhost:6379/0'
app.config['CACHE_DEFAULT_TIMEOUT'] = 300

cache = Cache(app)
db = SQLAlchemy(app)

# Model with caching methods
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    
    @classmethod
    @cache.memoize(timeout=600)
    def get_by_id(cls, user_id):
        return cls.query.get(user_id)
    
    @classmethod
    def invalidate_cache(cls, user_id):
        cache.delete_memoized(cls.get_by_id, user_id)

# Custom caching decorator with cache keys
def cached_route(timeout=300, key_prefix=None):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            # Generate cache key
            if key_prefix:
                cache_key = f"{key_prefix}:{request.full_path}"
            else:
                cache_key = f"{f.__name__}:{request.full_path}"
            
            # Try cache first
            cached_result = cache.get(cache_key)
            if cached_result:
                return cached_result
            
            # Execute function and cache result
            result = f(*args, **kwargs)
            cache.set(cache_key, result, timeout=timeout)
            return result
        return decorated_function
    return decorator

# Cached routes
@app.route('/users/<int:user_id>')
@cache.cached(timeout=300, key_prefix='user_profile')
def get_user(user_id):
    user = User.get_by_id(user_id)
    if not user:
        raise NotFound()
    
    return jsonify({
        'id': user.id,
        'username': user.username,
        'email': user.email
    })