Advanced Python Programming: Mastering Concurrency, Metaprogramming, and Performance Optimization
Python's simplicity hides extraordinary depth. While beginners appreciate its readable syntax and vast ecosystem, experienced developers discover a language capable of sophisticated metaprogramming, high-performance concurrent systems, and elegant abstractions that rival any language in existence. This comprehensive guide explores the advanced techniques that separate Python experts from intermediate developers — the patterns, tools, and mental models that enable you to write Python that is not just correct, but genuinely exceptional.
We'll cover concurrency models in depth (threading, multiprocessing, asyncio, and their interplay), metaprogramming techniques (descriptors, metaclasses, decorators as class factories), performance optimization (profiling, NumPy vectorization, Cython, memory management), and the design patterns that make large Python codebases maintainable. Each section includes production-quality code examples drawn from real-world applications.
Python's Execution Model: Understanding the Foundation
Advanced Python programming begins with a clear mental model of how Python actually executes code. Most Python implementations use a Global Interpreter Lock (GIL) — a mutex that protects access to Python objects, preventing multiple threads from executing Python bytecodes simultaneously. This fundamental design decision has profound implications for concurrent programming.
The GIL: Implications and Workarounds
The GIL exists because CPython's memory management is not thread-safe. Reference counting — Python's primary garbage collection mechanism — requires atomic updates to reference counts, and the GIL provides this safety without per-object locking overhead. The practical consequence: CPU-bound Python code cannot benefit from multiple threads running simultaneously on multiple cores.
However, the GIL is released during I/O operations and certain C extension calls. This means threading is effective for I/O-bound workloads (network requests, disk operations, database queries) but not for CPU-bound workloads (numerical computation, image processing, cryptography).
import threading
import time
import requests
from concurrent.futures import ThreadPoolExecutor
# I/O bound: threading helps significantly
def fetch_url(url):
response = requests.get(url)
return len(response.content)
urls = [
'https://httpbin.org/get',
'https://httpbin.org/headers',
'https://httpbin.org/ip',
'https://httpbin.org/user-agent',
]
# Sequential: ~2 seconds (network latency stacks)
start = time.time()
results = [fetch_url(url) for url in urls]
print(f"Sequential: {time.time() - start:.2f}s")
# Threaded: ~0.5 seconds (I/O overlaps)
start = time.time()
with ThreadPoolExecutor(max_workers=4) as executor:
results = list(executor.map(fetch_url, urls))
print(f"Threaded: {time.time() - start:.2f}s")Bytecode and the CPython VM
Understanding Python bytecode helps optimize performance-critical code. The dis module disassembles Python bytecode, revealing the actual operations the interpreter performs:
import dis
def example(x, y):
result = x * y + x
return result
dis.dis(example)
# LOAD_FAST x
# LOAD_FAST y
# BINARY_MULTIPLY
# LOAD_FAST x
# BINARY_ADD
# STORE_FAST result
# LOAD_FAST result
# RETURN_VALUEEach bytecode instruction has overhead — the interpreter loop dispatch, operand stack management, and object method resolution. Hot paths in performance-critical code should minimize bytecode operations: local variable access is faster than global, attribute access caches with __slots__, and avoiding unnecessary object creation reduces allocation pressure.
Concurrency: Threading, Multiprocessing, and AsyncIO
Threading: Shared Memory Concurrency
Python's threading module provides OS threads with shared memory. Despite the GIL, threading is effective for I/O-bound workloads and provides genuine parallelism for I/O operations. Thread safety requires careful attention to shared mutable state.
import threading
from collections import deque
from contextlib import contextmanager
class ThreadSafeQueue:
"""Thread-safe queue with blocking operations."""
def __init__(self, maxsize=0):
self._queue = deque()
self._maxsize = maxsize
self._lock = threading.Lock()
self._not_empty = threading.Condition(self._lock)
self._not_full = threading.Condition(self._lock)
def put(self, item, block=True, timeout=None):
with self._not_full:
if self._maxsize > 0:
while len(self._queue) >= self._maxsize:
if not block:
raise Full()
self._not_full.wait(timeout)
self._queue.append(item)
self._not_empty.notify()
def get(self, block=True, timeout=None):
with self._not_empty:
while not self._queue:
if not block:
raise Empty()
self._not_empty.wait(timeout)
item = self._queue.popleft()
self._not_full.notify()
return item
# Thread-local storage for per-thread state
_thread_local = threading.local()
def get_connection():
"""Get or create a per-thread database connection."""
if not hasattr(_thread_local, 'connection'):
_thread_local.connection = create_db_connection()
return _thread_local.connectionMultiprocessing: True Parallelism
For CPU-bound workloads, multiprocessing bypasses the GIL by spawning separate processes. Each process has its own Python interpreter and memory space; communication occurs through IPC mechanisms (pipes, queues, shared memory).
import multiprocessing as mp
from multiprocessing import Pool, Manager, shared_memory
import numpy as np
import time
def compute_chunk(args):
"""CPU-intensive computation on a data chunk."""
data, start, end = args
result = np.zeros(end - start)
for i in range(end - start):
# Simulated intensive computation
result[i] = sum(data[start + i] ** k for k in range(1, 10))
return result
def parallel_compute(data, n_workers=None):
"""Distribute computation across multiple processes."""
n_workers = n_workers or mp.cpu_count()
chunk_size = len(data) // n_workers
chunks = [
(data, i * chunk_size, min((i + 1) * chunk_size, len(data)))
for i in range(n_workers)
]
with Pool(n_workers) as pool:
results = pool.map(compute_chunk, chunks)
return np.concatenate(results)
# Shared memory for zero-copy data sharing (Python 3.8+)
def worker_with_shared_memory(shm_name, shape, dtype, start, end, result_queue):
shm = shared_memory.SharedMemory(name=shm_name)
data = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
# Process slice without copying
chunk_result = np.mean(data[start:end])
result_queue.put((start, end, chunk_result))
shm.close()
if __name__ == '__main__':
# Create large array in shared memory
data = np.random.random(10_000_000).astype(np.float64)
shm = shared_memory.SharedMemory(create=True, size=data.nbytes)
shared_array = np.ndarray(data.shape, dtype=data.dtype, buffer=shm.buf)
shared_array[:] = data[:]
manager = Manager()
result_queue = manager.Queue()
n_workers = mp.cpu_count()
chunk_size = len(data) // n_workers
processes = []
for i in range(n_workers):
start = i * chunk_size
end = min((i + 1) * chunk_size, len(data))
p = mp.Process(
target=worker_with_shared_memory,
args=(shm.name, data.shape, data.dtype, start, end, result_queue)
)
processes.append(p)
p.start()
for p in processes:
p.join()
shm.close()
shm.unlink()AsyncIO: Cooperative Multitasking
AsyncIO uses an event loop and coroutines for cooperative multitasking — perfect for high-concurrency I/O-bound workloads where thousands of simultaneous connections are needed. A single thread handles thousands of concurrent connections, with the event loop switching between coroutines when they await I/O operations.
import asyncio
import aiohttp
import aiofiles
from typing import AsyncIterator, TypeVar
from contextlib import asynccontextmanager
T = TypeVar('T')
class AsyncRateLimiter:
"""Token bucket rate limiter for async code."""
def __init__(self, rate: float, capacity: int):
self.rate = rate # tokens per second
self.capacity = capacity
self.tokens = capacity
self.last_refill = asyncio.get_event_loop().time()
self._lock = asyncio.Lock()
async def acquire(self, tokens: int = 1):
async with self._lock:
now = asyncio.get_event_loop().time()
elapsed = now - self.last_refill
self.tokens = min(
self.capacity,
self.tokens + elapsed * self.rate
)
self.last_refill = now
if self.tokens < tokens:
wait_time = (tokens - self.tokens) / self.rate
await asyncio.sleep(wait_time)
self.tokens = 0
else:
self.tokens -= tokens
async def fetch_with_retry(
session: aiohttp.ClientSession,
url: str,
max_retries: int = 3,
backoff_base: float = 1.0
) -> dict:
"""Fetch URL with exponential backoff retry."""
for attempt in range(max_retries):
try:
async with session.get(url) as response:
response.raise_for_status()
return await response.json()
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
if attempt == max_retries - 1:
raise
wait = backoff_base * (2 ** attempt)
await asyncio.sleep(wait)
async def process_urls_concurrently(urls: list[str], max_concurrent: int = 10):
"""Fetch multiple URLs with bounded concurrency."""
semaphore = asyncio.Semaphore(max_concurrent)
rate_limiter = AsyncRateLimiter(rate=20, capacity=20)
async def bounded_fetch(session, url):
async with semaphore:
await rate_limiter.acquire()
return await fetch_with_retry(session, url)
timeout = aiohttp.ClientTimeout(total=30)
async with aiohttp.ClientSession(timeout=timeout) as session:
tasks = [bounded_fetch(session, url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
# Async generators for streaming data processing
async def stream_file_lines(path: str) -> AsyncIterator[str]:
async with aiofiles.open(path) as f:
async for line in f:
yield line.rstrip()
async def process_large_file(path: str):
async for line in stream_file_lines(path):
# Process line without loading entire file
await process_line(line)Mixing Concurrency Models
Real applications often combine concurrency models. A web server might use asyncio for request handling, thread pool for blocking library calls, and process pool for CPU-intensive work:
import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
async def handle_request(request_data: dict):
loop = asyncio.get_event_loop()
# Run blocking I/O in thread pool (doesn't block event loop)
with ThreadPoolExecutor() as thread_pool:
db_result = await loop.run_in_executor(
thread_pool,
blocking_db_query,
request_data['user_id']
)
# Run CPU-intensive work in process pool (bypasses GIL)
with ProcessPoolExecutor() as process_pool:
analysis_result = await loop.run_in_executor(
process_pool,
cpu_intensive_analysis,
db_result
)
return analysis_resultMetaprogramming: Python's Introspective Power
Metaprogramming — code that manipulates code — is where Python's dynamic nature truly shines. Descriptors, metaclasses, and advanced decorator patterns enable frameworks, ORMs, and APIs that feel like language extensions.
Descriptors: The Protocol Behind Python's Object Model
Descriptors are the mechanism behind properties, methods, class methods, and static methods. Any class implementing __get__, __set__, or __delete__ is a descriptor. Understanding descriptors unlocks the full power of Python's object system.
from typing import Any, Optional, Type
import weakref
class TypedAttribute:
"""Descriptor that enforces type checking."""
def __set_name__(self, owner, name):
self.name = name
self.private_name = f'_{name}'
def __init__(self, expected_type: type, default=None):
self.expected_type = expected_type
self.default = default
def __get__(self, obj, objtype=None):
if obj is None:
return self
return getattr(obj, self.private_name, self.default)
def __set__(self, obj, value):
if not isinstance(value, self.expected_type):
raise TypeError(
f'{self.name} must be {self.expected_type.__name__}, '
f'got {type(value).__name__}'
)
setattr(obj, self.private_name, value)
class CachedProperty:
"""Lazy property computed once and cached on the instance."""
def __init__(self, func):
self.func = func
self.attrname = None
self.__doc__ = func.__doc__
def __set_name__(self, owner, name):
self.attrname = name
def __get__(self, instance, owner=None):
if instance is None:
return self
cache = instance.__dict__
val = cache.get(self.attrname)
if val is None:
val = self.func(instance)
cache[self.attrname] = val
return val
class ObservableAttribute:
"""Descriptor that notifies observers on change."""
def __set_name__(self, owner, name):
self.name = name
self.private_name = f'_{name}'
# Register with owner class
if not hasattr(owner, '_observers'):
owner._observers = {}
owner._observers[name] = []
def __get__(self, obj, objtype=None):
if obj is None:
return self
return getattr(obj, self.private_name, None)
def __set__(self, obj, value):
old_value = getattr(obj, self.private_name, None)
setattr(obj, self.private_name, value)
if old_value != value:
for observer in obj.__class__._observers.get(self.name, []):
observer(obj, self.name, old_value, value)
class DataModel:
name = TypedAttribute(str, default='')
age = TypedAttribute(int, default=0)
email = ObservableAttribute()
@CachedProperty
def computed_hash(self):
# Expensive computation cached after first access
import hashlib
return hashlib.md5(f'{self.name}{self.email}'.encode()).hexdigest()Metaclasses: Classes That Create Classes
Metaclasses define how classes themselves behave. They're the mechanism behind ORMs (SQLAlchemy's declarative base), API frameworks, and automatic class registration patterns. While often described as complex, metaclasses become natural once you understand that class statements are syntactic sugar for metaclass calls.
from typing import Dict, Any, Type
class SingletonMeta(type):
"""Metaclass that enforces singleton pattern."""
_instances: Dict[type, Any] = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]
class RegistryMeta(type):
"""Metaclass that auto-registers subclasses."""
registry: Dict[str, type] = {}
def __new__(mcs, name, bases, namespace):
cls = super().__new__(mcs, name, bases, namespace)
# Don't register the base class itself
if bases:
mcs.registry[name] = cls
return cls
@classmethod
def get(mcs, name: str) -> Optional[type]:
return mcs.registry.get(name)
class Plugin(metaclass=RegistryMeta):
"""Base class; all subclasses automatically registered."""
def execute(self):
raise NotImplementedError
class CSVPlugin(Plugin):
def execute(self):
return "Processing CSV"
class JSONPlugin(Plugin):
def execute(self):
return "Processing JSON"
# Auto-registered: RegistryMeta.registry == {'CSVPlugin': CSVPlugin, 'JSONPlugin': JSONPlugin}
plugin = RegistryMeta.get('CSVPlugin')()
plugin.execute() # "Processing CSV"
class ABCMeta(type):
"""Simplified abstract base class metaclass."""
def __new__(mcs, name, bases, namespace):
abstract_methods = set()
# Collect abstract methods from bases
for base in bases:
if hasattr(base, '__abstractmethods__'):
abstract_methods |= base.__abstractmethods__
# Check which abstract methods are implemented
for key, value in namespace.items():
if callable(value) and not getattr(value, '__isabstractmethod__', False):
abstract_methods.discard(key)
cls = super().__new__(mcs, name, bases, namespace)
cls.__abstractmethods__ = frozenset(abstract_methods)
return clsAdvanced Decorator Patterns
Decorators are Python's most visible metaprogramming tool. Advanced patterns include class-based decorators with state, decorator factories with complex argument processing, and decorators that work on both functions and methods.
import functools
import time
import threading
from typing import Callable, TypeVar, ParamSpec
P = ParamSpec('P')
R = TypeVar('R')
class retry:
"""Decorator class with configurable retry logic."""
def __init__(
self,
exceptions: tuple = (Exception,),
max_attempts: int = 3,
delay: float = 1.0,
backoff: float = 2.0,
logger=None
):
self.exceptions = exceptions
self.max_attempts = max_attempts
self.delay = delay
self.backoff = backoff
self.logger = logger
def __call__(self, func: Callable[P, R]) -> Callable[P, R]:
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
delay = self.delay
for attempt in range(self.max_attempts):
try:
return func(*args, **kwargs)
except self.exceptions as e:
if attempt == self.max_attempts - 1:
raise
if self.logger:
self.logger.warning(
f'Attempt {attempt + 1} failed: {e}. '
f'Retrying in {delay}s...'
)
time.sleep(delay)
delay *= self.backoff
return wrapper
class memoize:
"""LRU cache decorator with size limit and TTL."""
def __init__(self, maxsize: int = 128, ttl: Optional[float] = None):
self.maxsize = maxsize
self.ttl = ttl
self.cache = {}
self.access_times = {}
self.creation_times = {}
self.lock = threading.RLock()
def __call__(self, func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
key = (args, tuple(sorted(kwargs.items())))
with self.lock:
now = time.monotonic()
if key in self.cache:
# Check TTL
if self.ttl and (now - self.creation_times[key]) > self.ttl:
del self.cache[key]
else:
self.access_times[key] = now
return self.cache[key]
# Evict if at capacity (LRU)
if len(self.cache) >= self.maxsize:
oldest_key = min(self.access_times, key=self.access_times.get)
del self.cache[oldest_key]
del self.access_times[oldest_key]
del self.creation_times[oldest_key]
result = func(*args, **kwargs)
self.cache[key] = result
self.access_times[key] = now
self.creation_times[key] = now
return result
wrapper.cache_clear = lambda: self.cache.clear()
wrapper.cache_info = lambda: {
'size': len(self.cache),
'maxsize': self.maxsize
}
return wrapper
# Context manager decorator
def contextual(func):
"""Make a generator function into a context manager."""
@functools.wraps(func)
def helper(*args, **kwargs):
return contextlib.contextmanager(func)(*args, **kwargs)
return helperPerformance Optimization: Making Python Fast
Profiling: Finding Real Bottlenecks
Never optimize without profiling. Python provides multiple profiling tools that reveal where time is actually spent, often in surprising places.
import cProfile
import pstats
import io
import line_profiler
import memory_profiler
# CPU profiling with cProfile
def profile_function(func, *args, **kwargs):
pr = cProfile.Profile()
pr.enable()
result = func(*args, **kwargs)
pr.disable()
stream = io.StringIO()
ps = pstats.Stats(pr, stream=stream).sort_stats('cumulative')
ps.print_stats(20) # Top 20 functions
print(stream.getvalue())
return result
# Line-by-line profiling with line_profiler
@profile # Add @profile decorator, run with kernprof -l -v script.py
def slow_function(data):
result = []
for item in data:
processed = item ** 2
result.append(processed)
return sum(result)
# Memory profiling
@memory_profiler.profile
def memory_intensive():
large_list = [i for i in range(1_000_000)]
filtered = [x for x in large_list if x % 2 == 0]
return filtered
# Timing micro-benchmarks with timeit
import timeit
# Compare list comprehension vs map vs for loop
setup = "data = list(range(10000))"
list_comp = timeit.timeit(
"[x**2 for x in data]",
setup=setup,
number=1000
)
map_func = timeit.timeit(
"list(map(lambda x: x**2, data))",
setup=setup,
number=1000
)
print(f"List comprehension: {list_comp:.3f}s")
print(f"Map: {map_func:.3f}s")NumPy Vectorization: Eliminating Python Loops
For numerical computation, replacing Python loops with NumPy vectorized operations can provide 10-100x speedups by executing optimized C code on entire arrays.
import numpy as np
import time
# Scalar Python implementation
def python_moving_average(data, window):
result = []
for i in range(len(data) - window + 1):
result.append(sum(data[i:i+window]) / window)
return result
# NumPy vectorized implementation
def numpy_moving_average(data, window):
cumsum = np.cumsum(np.insert(data, 0, 0))
return (cumsum[window:] - cumsum[:-window]) / window
data = list(range(100_000))
np_data = np.array(data, dtype=np.float64)
# Python: ~2 seconds; NumPy: ~0.001 seconds (2000x faster)
start = time.perf_counter()
python_result = python_moving_average(data, 100)
print(f"Python: {time.perf_counter() - start:.3f}s")
start = time.perf_counter()
numpy_result = numpy_moving_average(np_data, 100)
print(f"NumPy: {time.perf_counter() - start:.3f}s")
# Broadcasting: operations on arrays of different shapes
# Apply 3 different thresholds to a 1M element array
data = np.random.random(1_000_000)
thresholds = np.array([0.25, 0.5, 0.75])
# Without broadcasting: would need 3 separate operations or loop
# With broadcasting: single vectorized operation
# data[:, np.newaxis] shape: (1M, 1)
# thresholds shape: (3,)
# result shape: (1M, 3)
result = data[:, np.newaxis] > thresholds
counts = result.sum(axis=0) # Elements above each thresholdMemory Management and Object Optimization
import sys
from __slots__ import slots # illustration
# __slots__ reduces memory by ~40-60% for classes with many instances
class PointWithDict:
def __init__(self, x, y, z):
self.x = x
self.y = y
self.z = z
class PointWithSlots:
__slots__ = ('x', 'y', 'z')
def __init__(self, x, y, z):
self.x = x
self.y = y
self.z = z
# Memory comparison
p_dict = PointWithDict(1.0, 2.0, 3.0)
p_slots = PointWithSlots(1.0, 2.0, 3.0)
print(f"With dict: {sys.getsizeof(p_dict) + sys.getsizeof(p_dict.__dict__)} bytes")
print(f"With slots: {sys.getsizeof(p_slots)} bytes")
# Generators for memory-efficient data processing
def read_large_file(path):
"""Memory-efficient file reading with generators."""
with open(path) as f:
for line in f:
yield line.strip()
def process_csv_streaming(path):
"""Process CSV without loading entire file into memory."""
import csv
with open(path, newline='') as f:
reader = csv.DictReader(f)
for row in reader:
yield process_row(row)
# Memory views for zero-copy slicing
data = bytearray(b'Hello, World! ' * 1000)
view = memoryview(data)
# Slicing creates views, not copies
chunk = view[100:200] # No memory allocationCython and C Extensions: When Python Isn't Fast Enough
For maximum performance, Cython compiles Python-like code to C. Key Cython optimizations: static typing with cdef, disabling Python overhead with @cython.boundscheck(False), and using typed memoryviews for array access.
# example.pyx (Cython file)
# cython: language_level=3
import cython
import numpy as np
cimport numpy as cnp
@cython.boundscheck(False)
@cython.wraparound(False)
def fast_moving_average(cnp.ndarray[double, ndim=1] data, int window):
cdef int n = len(data) - window + 1
cdef cnp.ndarray[double, ndim=1] result = np.empty(n)
cdef double window_sum = 0.0
cdef int i
# Initial window
for i in range(window):
window_sum += data[i]
result[0] = window_sum / window
# Sliding window
for i in range(1, n):
window_sum += data[i + window - 1] - data[i - 1]
result[i] = window_sum / window
return result
# setup.py to compile:
# from setuptools import setup
# from Cython.Build import cythonize
# setup(ext_modules=cythonize("example.pyx"))Design Patterns in Python
Structural Patterns: Adapters, Proxies, and Composites
from abc import ABC, abstractmethod
from typing import Iterator
# Composite pattern: tree structures with uniform interface
class FileSystemItem(ABC):
@abstractmethod
def size(self) -> int: ...
@abstractmethod
def items(self) -> Iterator['FileSystemItem']: ...
class File(FileSystemItem):
def __init__(self, name: str, content: bytes):
self.name = name
self.content = content
def size(self) -> int:
return len(self.content)
def items(self) -> Iterator['FileSystemItem']:
return iter([])
class Directory(FileSystemItem):
def __init__(self, name: str):
self.name = name
self._children: list[FileSystemItem] = []
def add(self, item: FileSystemItem):
self._children.append(item)
def size(self) -> int:
return sum(child.size() for child in self._children)
def items(self) -> Iterator['FileSystemItem']:
return iter(self._children)
# Proxy pattern: transparent interception
class DatabaseProxy:
"""Proxy adding caching and logging to database calls."""
def __init__(self, real_db, cache, logger):
self._db = real_db
self._cache = cache
self._logger = logger
def query(self, sql: str, params=None):
cache_key = (sql, params)
if cache_key in self._cache:
self._logger.debug(f"Cache hit: {sql[:50]}")
return self._cache[cache_key]
self._logger.info(f"DB query: {sql[:50]}")
result = self._db.query(sql, params)
self._cache[cache_key] = result
return result
def __getattr__(self, name):
# Delegate all other attributes to real db
return getattr(self._db, name)Behavioral Patterns: Observers, Commands, and State Machines
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Callable
# State machine pattern
class OrderState(Enum):
PENDING = auto()
PROCESSING = auto()
SHIPPED = auto()
DELIVERED = auto()
CANCELLED = auto()
@dataclass
class Order:
id: str
state: OrderState = OrderState.PENDING
_transitions: dict = field(default_factory=dict, init=False, repr=False)
_callbacks: list = field(default_factory=list, init=False, repr=False)
def __post_init__(self):
self._transitions = {
OrderState.PENDING: [OrderState.PROCESSING, OrderState.CANCELLED],
OrderState.PROCESSING: [OrderState.SHIPPED, OrderState.CANCELLED],
OrderState.SHIPPED: [OrderState.DELIVERED],
OrderState.DELIVERED: [],
OrderState.CANCELLED: [],
}
def transition_to(self, new_state: OrderState):
if new_state not in self._transitions[self.state]:
raise ValueError(
f"Invalid transition: {self.state} -> {new_state}"
)
old_state = self.state
self.state = new_state
for callback in self._callbacks:
callback(self, old_state, new_state)
def on_transition(self, callback: Callable):
self._callbacks.append(callback)
return callback
# Command pattern with undo support
class Command(ABC):
@abstractmethod
def execute(self): ...
@abstractmethod
def undo(self): ...
class CommandHistory:
def __init__(self):
self._history = []
self._redo_stack = []
def execute(self, command: Command):
command.execute()
self._history.append(command)
self._redo_stack.clear()
def undo(self):
if not self._history:
return
command = self._history.pop()
command.undo()
self._redo_stack.append(command)
def redo(self):
if not self._redo_stack:
return
command = self._redo_stack.pop()
command.execute()
self._history.append(command)Testing Advanced Python Code
Property-Based Testing with Hypothesis
from hypothesis import given, strategies as st, assume
from hypothesis.stateful import RuleBasedStateMachine, rule, invariant
@given(st.lists(st.integers()))
def test_sort_idempotent(lst):
"""Sorting twice gives same result as sorting once."""
assert sorted(sorted(lst)) == sorted(lst)
@given(st.lists(st.floats(allow_nan=False, allow_infinity=False)))
def test_moving_average_length(data):
window = 5
assume(len(data) >= window)
result = numpy_moving_average(np.array(data), window)
assert len(result) == len(data) - window + 1
# Stateful testing: model a counter
class CounterMachine(RuleBasedStateMachine):
def __init__(self):
super().__init__()
self.model = 0
self.counter = Counter()
@rule(amount=st.integers(min_value=1, max_value=100))
def increment(self, amount):
self.model += amount
self.counter.increment(amount)
@invariant()
def counter_matches_model(self):
assert self.counter.value == self.modelMocking and Patching
from unittest.mock import Mock, patch, AsyncMock, MagicMock
import pytest
@pytest.fixture
def mock_db():
db = MagicMock()
db.query.return_value = [{'id': 1, 'name': 'test'}]
db.execute.return_value = None
return db
@pytest.mark.asyncio
async def test_async_service(mock_db):
"""Test async code with AsyncMock."""
service = DataService(db=mock_db)
with patch('mymodule.external_api', new_callable=AsyncMock) as mock_api:
mock_api.return_value = {'status': 'ok'}
result = await service.process_data(user_id=1)
mock_db.query.assert_called_once_with(
'SELECT * FROM users WHERE id = ?', (1,)
)
mock_api.assert_awaited_once()
assert result['status'] == 'processed'Python Packaging and Distribution
Modern Packaging with pyproject.toml
# pyproject.toml
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "mypackage"
version = "1.0.0"
description = "A high-performance data processing library"
readme = "README.md"
requires-python = ">=3.11"
license = {text = "MIT"}
dependencies = [
"numpy>=1.24",
"pydantic>=2.0",
"aiohttp>=3.9",
]
[project.optional-dependencies]
dev = ["pytest", "hypothesis", "mypy", "ruff"]
docs = ["sphinx", "sphinx-rtd-theme"]
[tool.hatch.envs.default]
dependencies = ["pytest", "pytest-asyncio", "hypothesis"]
[tool.mypy]
strict = true
python_version = "3.11"
[tool.ruff]
line-length = 88
select = ["E", "F", "I", "N", "W", "UP"]Type System: Advanced Typing
from typing import (
TypeVar, Generic, Protocol, runtime_checkable,
overload, Literal, TypedDict, NamedTuple,
get_type_hints, TYPE_CHECKING
)
T = TypeVar('T')
T_co = TypeVar('T_co', covariant=True)
# Protocol for structural subtyping (duck typing with types)
@runtime_checkable
class Comparable(Protocol):
def __lt__(self, other: 'Comparable') -> bool: ...
def __le__(self, other: 'Comparable') -> bool: ...
def find_min(items: list[T]) -> T:
"""Works for any type implementing Comparable."""
if not items:
raise ValueError("Empty sequence")
return min(items)
# Generic classes
class Stack(Generic[T]):
def __init__(self) -> None:
self._items: list[T] = []
def push(self, item: T) -> None:
self._items.append(item)
def pop(self) -> T:
if not self._items:
raise IndexError("Stack is empty")
return self._items.pop()
def peek(self) -> T:
if not self._items:
raise IndexError("Stack is empty")
return self._items[-1]
def __len__(self) -> int:
return len(self._items)
# Overloaded functions
@overload
def process(data: str) -> str: ...
@overload
def process(data: bytes) -> bytes: ...
@overload
def process(data: list) -> list: ...
def process(data):
if isinstance(data, str):
return data.upper()
elif isinstance(data, bytes):
return data.upper()
elif isinstance(data, list):
return [process(item) for item in data]Data Classes and Pydantic
from dataclasses import dataclass, field, KW_ONLY
from pydantic import BaseModel, validator, Field, root_validator
from datetime import datetime
from decimal import Decimal
# Advanced dataclasses
@dataclass(frozen=True, slots=True)
class Vector3D:
x: float
y: float
z: float
def __add__(self, other: 'Vector3D') -> 'Vector3D':
return Vector3D(self.x + other.x, self.y + other.y, self.z + other.z)
@property
def magnitude(self) -> float:
return (self.x**2 + self.y**2 + self.z**2) ** 0.5
def normalize(self) -> 'Vector3D':
mag = self.magnitude
return Vector3D(self.x/mag, self.y/mag, self.z/mag)
# Pydantic for data validation and serialization
class OrderItem(BaseModel):
product_id: str = Field(..., min_length=1, max_length=50)
quantity: int = Field(..., gt=0, le=1000)
unit_price: Decimal = Field(..., gt=0, decimal_places=2)
@validator('product_id')
def validate_product_id(cls, v):
if not v.startswith('PROD-'):
raise ValueError("Product ID must start with PROD-")
return v
@property
def total_price(self) -> Decimal:
return self.quantity * self.unit_price
class Order(BaseModel):
order_id: str
customer_id: str
items: list[OrderItem] = Field(..., min_items=1)
created_at: datetime = Field(default_factory=datetime.utcnow)
@root_validator
def validate_order(cls, values):
items = values.get('items', [])
total = sum(item.total_price for item in items)
if total > Decimal('10000'):
raise ValueError("Order total exceeds maximum allowed value")
return values
class Config:
json_encoders = {
datetime: lambda v: v.isoformat(),
Decimal: str
}Conclusion: The Path to Python Mastery
Advanced Python mastery is not a destination but a continuous journey. The language rewards deep exploration: every pattern you learn reveals new patterns beneath it, every optimization teaches you more about how computation works, and every design problem solved with Python's unique features deepens your appreciation for the language's design philosophy.
The key principles to carry forward: always profile before optimizing; choose the right concurrency model for your workload; use metaprogramming to eliminate repetition rather than to show off; write types even when they're optional (they document intent and catch bugs); and test with property-based tests and stateful machines, not just example-based tests.
Python's ecosystem continues to evolve rapidly. The addition of structural pattern matching (Python 3.10), exception groups (3.11), asyncio improvements, and performance enhancements in CPython mean that expertise requires ongoing engagement. The engineers who excel with Python are those who keep building real systems, reading CPython source code to understand behavior, and contributing to open source projects where their code will be scrutinized by other experts.
Master these advanced techniques, and Python becomes not just a scripting language but a genuinely powerful platform for building systems that are correct, performant, and a pleasure to maintain.
Comments
Post a Comment