ffsshhttiikk What I do
- Profile and identify bottlenecks
- Optimize database queries
- Implement caching strategies
- Optimize algorithmic complexity
- Handle memory efficiently
- Reduce network overhead
- Parallelize operations
- Monitor performance metrics
When to use me
When optimizing application performance or debugging slow code.
Performance Profiling
import cProfile
import pstats
import memory_profiler
import time
from functools import wraps
import line_profiler
def profile_function(profile_file: str = "profile.prof"):
"""Decorator to profile function execution."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
profiler = cProfile.Profile()
profiler.enable()
result = func(*args, **kwargs)
profiler.disable()
profiler.dump_stats(profile_file)
# Print stats
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(20)
return result
return wrapper
return decorator
@profile_function()
def slow_function():
"""Example slow function."""
data = []
for i in range(10000):
data.append(i * 2)
return sum(data)
# Memory profiling
@memory_profiler.profile
def memory_intensive():
"""Profile memory usage."""
large_list = [i for i in range(1000000)]
return sum(large_list)
# Line-by-line profiling
def profile_lines():
profiler = line_profiler.LineProfiler()
profiler.add_function(slow_function)
profiler.enable()
slow_function()
profiler.disable()
profiler.print_stats()
Database Query Optimization
# N+1 Query Problem
# BAD - N+1 queries
def get_all_users_with_posts():
users = db.query(User).all() # 1 query
result = []
for user in users: # N queries!
posts = db.query(Post).filter_by(user_id=user.id).all()
result.append({'user': user, 'posts': posts})
return result
# GOOD - eager loading
def get_all_users_with_posts():
# Single query with JOIN
users = (
db.query(User)
.options(joinedload(User.posts))
.all()
)
return [{'user': user, 'posts': user.posts} for user in users]
# GOOD - select in load
def get_all_users_with_posts():
users = db.query(User).all()
# Batch load posts
user_ids = [u.id for u in users]
posts = db.query(Post).filter(Post.user_id.in_(user_ids)).all()
posts_by_user = {}
for post in posts:
posts_by_user.setdefault(post.user_id, []).append(post)
return [
{'user': user, 'posts': posts_by_user.get(user.id, [])}
for user in users
]
# Index optimization
class OptimizedQuery:
@staticmethod
def create_optimized_indexes():
# Composite index for common queries
indexes = [
# For filtering and sorting
"CREATE INDEX idx_posts_status_published ON posts(status) WHERE status = 'published'",
"CREATE INDEX idx_users_email ON users(email)",
# Composite index
"CREATE INDEX idx_orders_user_date ON orders(user_id, created_at DESC)",
# Covering index (includes all queried columns)
"CREATE INDEX idx_products_cover ON products(category_id, price) INCLUDE (name, description)",
]
for idx in indexes:
db.execute(idx)
Caching Strategies
from functools import lru_cache
from cachetools import TTLCache, LRUCache
# Function-level caching
@lru_cache(maxsize=128)
def expensive_computation(n: int) -> int:
"""Cache expensive function results."""
result = sum(range(n))
return result
# Time-based caching
class TTLCache:
def __init__(self, ttl_seconds: int = 300, maxsize: int = 1000):
self.cache = TTLCache(maxsize=maxsize, ttl=ttl_seconds)
def get(self, key: str):
return self.cache.get(key)
def set(self, key: str, value):
self.cache[key] = value
# Query result caching
class QueryCache:
def __init__(self, cache: TTLCache):
self.cache = cache
def get_or_set(
self,
query_key: str,
query_func: callable,
ttl: int = 300
):
"""Cache query results with TTL."""
if query_key in self.cache:
return self.cache[query_key]
result = query_func()
self.cache[query_key] = result
return result
# Invalidation strategies
class CacheInvalidator:
def __init__(self, cache):
self.cache = cache
self.subscriptions = {}
def subscribe(self, key_pattern: str, callback: callable):
"""Subscribe to cache invalidation events."""
self.subscriptions[key_pattern] = callback
def invalidate(self, key: str):
"""Invalidate matching cache entries."""
self.cache.pop(key, None)
for pattern, callback in self.subscriptions.items():
if self._matches(key, pattern):
callback(key)
def _matches(self, key: str, pattern: str) -> bool:
"""Check if key matches pattern."""
import fnmatch
return fnmatch.fnmatch(key, pattern)
Async Optimization
import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import List
class AsyncBatchProcessor:
"""Process items in batches for efficiency."""
def __init__(self, batch_size: int = 100):
self.batch_size = batch_size
async def process_items(self, items: List[dict]) -> List[dict]:
"""Process items in batches."""
results = []
for i in range(0, len(items), self.batch_size):
batch = items[i:i + self.batch_size]
batch_results = await self._process_batch(batch)
results.extend(batch_results)
return results
async def _process_batch(self, batch: List[dict]) -> List[dict]:
"""Process single batch."""
return await asyncio.gather(
*[self._process_item(item) for item in batch]
)
async def _process_item(self, item: dict) -> dict:
"""Process single item."""
# Simulated processing
return item
# Parallel execution with thread pool
class ParallelProcessor:
"""Execute CPU-bound tasks in parallel."""
def __init__(self, max_workers: int = 4):
self.executor = ThreadPoolExecutor(max_workers=max_workers)
def process_cpu_intensive(self, items: List[int]) -> List[int]:
"""Process items in parallel."""
loop = asyncio.new_event_loop()
tasks = [
loop.run_in_executor(self.executor, self._cpu_task, item)
for item in items
]
return loop.run_until_complete(asyncio.gather(*tasks))
def _cpu_task(self, n: int) -> int:
"""CPU intensive task."""
return sum(range(n))
Memory Optimization
import gc
from typing import Generator
import sys
class MemoryOptimizer:
"""Optimize memory usage."""
@staticmethod
def get_memory_usage():
"""Get current memory usage in MB."""
import psutil
process = psutil.Process()
return process.memory_info().rss / 1024 / 1024
@staticmethod
def force_garbage_collection():
"""Force garbage collection."""
gc.collect()
@staticmethod
def disable_garbage_collection():
"""Disable GC for performance-critical sections."""
gc.disable()
@staticmethod
def enable_garbage_collection():
"""Re-enable garbage collection."""
gc.enable()
# Generator for lazy evaluation
def process_large_file(file_path: str) -> Generator[dict, None, None]:
"""Process file line by line without loading entire file."""
with open(file_path, 'r') as f:
for line in f:
yield parse_line(line)
# Chunked processing
def process_in_chunks(data: list, chunk_size: int = 1000):
"""Yield chunks from data for memory efficiency."""
for i in range(0, len(data), chunk_size):
yield data[i:i + chunk_size]
# Use slots for classes
class OptimizedClass:
"""Use __slots__ to reduce memory overhead."""
__slots__ = ['name', 'value', 'timestamp']
def __init__(self, name: str, value: int):
self.name = name
self.value = value
self.timestamp = None
# Data compression for transmission
import zlib
def compress_data(data: bytes) -> bytes:
"""Compress data for transmission."""
return zlib.compress(data, level=6)
def decompress_data(data: bytes) -> bytes:
"""Decompress data."""
return zlib.decompress(data)
Best Practices
1. Measure before optimizing
- Profile to find actual bottlenecks
- Don't guess performance issues
2. Use appropriate data structures
- Choose O(1) vs O(n) operations
- Use sets for membership testing
3. Batch operations
- Reduce round trips
- Use bulk operations
4. Lazy loading
- Defer expensive operations
- Use generators
5. Caching
- Cache expensive computations
- Use appropriate TTL
6. Connection pooling
- Reuse database connections
- Reuse HTTP connections
7. Async I/O
- Non-blocking for I/O-bound
- Parallel for CPU-bound
8. Monitor in production
- Track performance metrics
- Alert on degradation