The Proxy Pattern is a structural design pattern that provides a surrogate or placeholder for another object to control access to it. The proxy acts as an intermediary between the client and the real object, allowing you to perform additional operations before or after the request is forwarded to the real object.
Core Concept
The pattern involves creating a proxy class that implements the same interface as the real object. This proxy can control access, add caching, provide lazy initialization, or implement security checks without the client knowing they’re working with a proxy instead of the real object.
Structure
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Client │ ────> │ Proxy │ ─────> │ RealSubject │
└─────────────┘ └─────────────┘ └─────────────┘
│ │
└─────────────────────────┘
implements
┌──────────────┐
│ Subject │
│ (Interface) │
└──────────────┘
Key Components
- Subject: Interface that defines the common operations for both RealSubject and Proxy
- RealSubject: The actual object that performs the real work
- Proxy: Controls access to the RealSubject and may perform additional operations
- Client: Uses the Subject interface to work with objects
Common Use Cases
- Access Control: Control who can access the real object and under what conditions.
- Caching: Store results of expensive operations to avoid repeated computations.
- Lazy Initialization: Defer the creation of expensive objects until they’re actually needed.
- Logging and Monitoring: Track method calls and performance metrics.
- Remote Proxy: Represent objects that exist in different address spaces (network, different processes).
Types of Proxies
Proxy Types
├── Virtual Proxy (Lazy Loading)
├── Protection Proxy (Access Control)
├── Remote Proxy (Network Communication)
├── Cache Proxy (Performance Optimization)
└── Smart Reference (Additional Functionality)
Benefits
- Controlled Access: Manage how and when the real object is accessed
- Performance Optimization: Add caching, lazy loading, or connection pooling
- Security: Implement authentication and authorization
- Transparency: Client code remains unchanged when using proxies
- Separation of Concerns: Keep business logic separate from cross-cutting concerns
Drawbacks
- Complexity: Adds another layer of abstraction
- Performance Overhead: May introduce latency for simple operations
- Memory Usage: Proxy objects consume additional memory
- Debugging Difficulty: Stack traces become more complex with proxy layers
Implementation Considerations
When implementing the Proxy Pattern, consider:
- Interface Design: Ensure the proxy implements the same interface as the real object
- Proxy Lifetime: Decide when to create and destroy proxy instances
- Thread Safety: Handle concurrent access if the application is multi-threaded
- Error Handling: Properly propagate exceptions from the real object
- Performance Impact: Balance the benefits against the overhead introduced
The Proxy Pattern is particularly useful when you need to add functionality to existing objects without modifying their code, or when you want to control expensive operations like network calls or database queries.This example demonstrates the basic Proxy Pattern implementation. The proxy provides lazy initialization, access control, logging, and monitoring without the client knowing it’s working with a proxy instead of the real object.This example demonstrates a caching proxy that significantly improves performance by storing results of expensive operations. The proxy handles cache invalidation, statistics tracking, and provides the same interface as the real service.
from abc import ABC, abstractmethod
from typing import Optional
import time
class Subject(ABC):
"""
Abstract base class defining the interface for both RealSubject and Proxy.
This ensures both classes have the same method signatures.
"""
@abstractmethod
def request(self, data: str) -> str:
"""Process a request with the given data."""
pass
class RealSubject(Subject):
"""
The actual object that performs the real work.
This represents an expensive operation that we want to control access to.
"""
def __init__(self, name: str) -> None:
self.name = name
print(f"RealSubject '{name}' initialized (expensive operation)")
def request(self, data: str) -> str:
"""
Simulate an expensive operation like database query or network call.
"""
print(f"RealSubject.request() processing: {data}")
# Simulate processing time
time.sleep(0.1)
return f"Processed by {self.name}: {data.upper()}"
class Proxy(Subject):
"""
Proxy that controls access to the RealSubject.
Implements the same interface as RealSubject.
"""
def __init__(self, subject_name: str) -> None:
self.subject_name = subject_name
self._real_subject: Optional[RealSubject] = None
self._access_count = 0
def request(self, data: str) -> str:
"""
Control access to the real subject and add additional functionality.
"""
# Access control - reject empty requests
if not data.strip():
raise ValueError("Request data cannot be empty")
# Lazy initialization - create real subject only when needed
if self._real_subject is None:
print("Proxy: Creating RealSubject (lazy initialization)")
self._real_subject = RealSubject(self.subject_name)
# Logging and monitoring
self._access_count += 1
print(f"Proxy: Access #{self._access_count} to RealSubject")
# Pre-processing
start_time = time.time()
# Delegate to real subject
result = self._real_subject.request(data)
# Post-processing
end_time = time.time()
print(f"Proxy: Request completed in {end_time - start_time:.3f}s")
return result
def get_access_count(self) -> int:
"""Additional functionality provided by proxy."""
return self._access_count
def demonstrate_proxy_pattern():
"""
Demonstrate the Proxy Pattern with a simple example.
"""
print("=== Proxy Pattern Demonstration ===\n")
# Client works with proxy, not knowing it's not the real object
print("1. Creating proxy (RealSubject not created yet)")
proxy = Proxy("DatabaseService")
print(f" Access count: {proxy.get_access_count()}")
print()
# First request triggers lazy initialization
print("2. First request (triggers RealSubject creation)")
try:
result1 = proxy.request("hello world")
print(f" Result: {result1}")
except ValueError as e:
print(f" Error: {e}")
print(f" Access count: {proxy.get_access_count()}")
print()
# Second request uses existing RealSubject
print("3. Second request (uses existing RealSubject)")
result2 = proxy.request("python programming")
print(f" Result: {result2}")
print(f" Access count: {proxy.get_access_count()}")
print()
# Error handling demonstration
print("4. Invalid request (access control)")
try:
proxy.request("")
except ValueError as e:
print(f" Error: {e}")
print(f" Access count: {proxy.get_access_count()}")
if __name__ == "__main__":
demonstrate_proxy_pattern()
This example demonstrates a caching proxy that significantly improves performance by storing results of expensive operations. The proxy handles cache invalidation, statistics tracking, and provides the same interface as the real service.
from abc import ABC, abstractmethod
from typing import Dict, Optional, Any
import time
import hashlib
class DataService(ABC):
"""
Abstract interface for data services.
Both real service and proxy implement this interface.
"""
@abstractmethod
def fetch_data(self, query: str) -> Dict[str, Any]:
"""Fetch data based on the query."""
pass
@abstractmethod
def get_user_profile(self, user_id: int) -> Dict[str, Any]:
"""Get user profile information."""
pass
class DatabaseService(DataService):
"""
Real subject that performs expensive database operations.
This simulates actual database queries with network latency.
"""
def __init__(self) -> None:
self.query_count = 0
print("DatabaseService: Connected to database")
def fetch_data(self, query: str) -> Dict[str, Any]:
"""
Simulate expensive database query operation.
"""
self.query_count += 1
print(f"DatabaseService: Executing query #{self.query_count}: {query}")
# Simulate network latency and processing time
time.sleep(0.5)
# Simulate database response
return {
"query": query,
"results": [f"result_{i}" for i in range(3)],
"timestamp": time.time(),
"query_id": self.query_count
}
def get_user_profile(self, user_id: int) -> Dict[str, Any]:
"""
Simulate user profile retrieval from database.
"""
self.query_count += 1
print(f"DatabaseService: Fetching user profile #{self.query_count} for user {user_id}")
# Simulate processing time
time.sleep(0.3)
return {
"user_id": user_id,
"name": f"User_{user_id}",
"email": f"user{user_id}@example.com",
"created_at": "2024-01-01",
"last_login": time.time()
}
class CachingProxy(DataService):
"""
Proxy that implements caching to improve performance.
Stores results in memory to avoid repeated expensive operations.
"""
def __init__(self, real_service: DataService, cache_ttl: int = 300) -> None:
self.real_service = real_service
self.cache_ttl = cache_ttl # Time-to-live in seconds
self._cache: Dict[str, Dict[str, Any]] = {}
self.cache_hits = 0
self.cache_misses = 0
print("CachingProxy: Initialized with caching layer")
def _generate_cache_key(self, method: str, *args) -> str:
"""
Generate a unique cache key for the method and arguments.
"""
key_data = f"{method}:{':'.join(map(str, args))}"
return hashlib.md5(key_data.encode()).hexdigest()
def _is_cache_valid(self, cache_entry: Dict[str, Any]) -> bool:
"""
Check if cache entry is still valid based on TTL.
"""
return time.time() - cache_entry["cached_at"] < self.cache_ttl
def _get_from_cache(self, cache_key: str) -> Optional[Dict[str, Any]]:
"""
Retrieve data from cache if valid, otherwise return None.
"""
if cache_key in self._cache:
cache_entry = self._cache[cache_key]
if self._is_cache_valid(cache_entry):
self.cache_hits += 1
print(f"CachingProxy: Cache HIT for key {cache_key[:8]}...")
return cache_entry["data"]
else:
# Remove expired entry
del self._cache[cache_key]
print(f"CachingProxy: Cache EXPIRED for key {cache_key[:8]}...")
self.cache_misses += 1
print(f"CachingProxy: Cache MISS for key {cache_key[:8]}...")
return None
def _store_in_cache(self, cache_key: str, data: Dict[str, Any]) -> None:
"""
Store data in cache with timestamp.
"""
self._cache[cache_key] = {
"data": data,
"cached_at": time.time()
}
print(f"CachingProxy: Stored in cache with key {cache_key[:8]}...")
def fetch_data(self, query: str) -> Dict[str, Any]:
"""
Fetch data with caching support.
"""
cache_key = self._generate_cache_key("fetch_data", query)
# Try to get from cache first
cached_result = self._get_from_cache(cache_key)
if cached_result is not None:
return cached_result
# Cache miss - fetch from real service
result = self.real_service.fetch_data(query)
# Store in cache for future requests
self._store_in_cache(cache_key, result)
return result
def get_user_profile(self, user_id: int) -> Dict[str, Any]:
"""
Get user profile with caching support.
"""
cache_key = self._generate_cache_key("get_user_profile", user_id)
# Try to get from cache first
cached_result = self._get_from_cache(cache_key)
if cached_result is not None:
return cached_result
# Cache miss - fetch from real service
result = self.real_service.get_user_profile(user_id)
# Store in cache for future requests
self._store_in_cache(cache_key, result)
return result
def get_cache_stats(self) -> Dict[str, Any]:
"""
Get caching statistics.
"""
total_requests = self.cache_hits + self.cache_misses
hit_rate = (self.cache_hits / total_requests * 100) if total_requests > 0 else 0
return {
"cache_hits": self.cache_hits,
"cache_misses": self.cache_misses,
"hit_rate": f"{hit_rate:.1f}%",
"cache_size": len(self._cache)
}
def clear_cache(self) -> None:
"""
Clear all cached data.
"""
self._cache.clear()
print("CachingProxy: Cache cleared")
def demonstrate_caching_proxy():
"""
Demonstrate the Caching Proxy Pattern with performance measurements.
"""
print("=== Caching Proxy Pattern Demonstration ===\n")
# Create real service and proxy
print("1. Setting up services")
database_service = DatabaseService()
caching_proxy = CachingProxy(database_service, cache_ttl=10)
print()
# First request - cache miss
print("2. First request (cache miss)")
start_time = time.time()
result1 = caching_proxy.fetch_data("SELECT * FROM users")
end_time = time.time()
print(f" Result: {result1['results']}")
print(f" Time taken: {end_time - start_time:.3f}s")
print(f" Cache stats: {caching_proxy.get_cache_stats()}")
print()
# Second request - cache hit
print("3. Same request again (cache hit)")
start_time = time.time()
result2 = caching_proxy.fetch_data("SELECT * FROM users")
end_time = time.time()
print(f" Result: {result2['results']}")
print(f" Time taken: {end_time - start_time:.3f}s")
print(f" Cache stats: {caching_proxy.get_cache_stats()}")
print()
# Different request - cache miss
print("4. Different request (cache miss)")
start_time = time.time()
result3 = caching_proxy.fetch_data("SELECT * FROM products")
end_time = time.time()
print(f" Result: {result3['results']}")
print(f" Time taken: {end_time - start_time:.3f}s")
print(f" Cache stats: {caching_proxy.get_cache_stats()}")
print()
# User profile requests
print("5. User profile requests")
users = [1, 2, 1, 3, 2, 1] # Note repeated user IDs
for user_id in users:
start_time = time.time()
profile = caching_proxy.get_user_profile(user_id)
end_time = time.time()
print(f" User {user_id}: {profile['name']} ({end_time - start_time:.3f}s)")
print(f" Final cache stats: {caching_proxy.get_cache_stats()}")
print()
# Cache expiration demo
print("6. Cache expiration demo")
print(" Waiting for cache to expire...")
time.sleep(11) # Wait for cache TTL to expire
start_time = time.time()
result4 = caching_proxy.fetch_data("SELECT * FROM users")
end_time = time.time()
print(f" After expiration: {end_time - start_time:.3f}s")
print(f" Cache stats: {caching_proxy.get_cache_stats()}")
if __name__ == "__main__":
demonstrate_caching_proxy()
This example demonstrates a protection proxy that implements comprehensive security controls including authentication, authorization, access logging, and account lockout mechanisms. The proxy ensures that only authenticated users with appropriate permissions can access the document service.
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Any
from enum import Enum
import time
from datetime import datetime
class Permission(Enum):
"""
Enumeration of different permission levels.
"""
READ = "read"
WRITE = "write"
DELETE = "delete"
ADMIN = "admin"
class User:
"""
Represents a user with authentication and authorization details.
"""
def __init__(self, username: str, role: str, permissions: List[Permission]):
self.username = username
self.role = role
self.permissions = permissions
self.last_login = datetime.now()
self.failed_attempts = 0
self.is_locked = False
def has_permission(self, permission: Permission) -> bool:
"""Check if user has a specific permission."""
return permission in self.permissions or Permission.ADMIN in self.permissions
def __str__(self) -> str:
return f"User(username='{self.username}', role='{self.role}')"
class DocumentService(ABC):
"""
Abstract interface for document operations.
Both real service and protection proxy implement this interface.
"""
@abstractmethod
def read_document(self, doc_id: str) -> Dict[str, Any]:
"""Read a document by its ID."""
pass
@abstractmethod
def write_document(self, doc_id: str, content: str) -> bool:
"""Write/update a document."""
pass
@abstractmethod
def delete_document(self, doc_id: str) -> bool:
"""Delete a document."""
pass
@abstractmethod
def list_documents(self) -> List[str]:
"""List all available documents."""
pass
class RealDocumentService(DocumentService):
"""
Real subject that performs actual document operations.
This represents the core business logic without security concerns.
"""
def __init__(self) -> None:
# Simulate document storage
self.documents: Dict[str, Dict[str, Any]] = {
"doc1": {
"content": "Public document content",
"created_at": "2024-01-01",
"owner": "system"
},
"doc2": {
"content": "Confidential business plan",
"created_at": "2024-01-02",
"owner": "admin"
},
"doc3": {
"content": "User manual draft",
"created_at": "2024-01-03",
"owner": "writer"
}
}
print("RealDocumentService: Initialized document storage")
def read_document(self, doc_id: str) -> Dict[str, Any]:
"""
Read document without any security checks.
"""
print(f"RealDocumentService: Reading document {doc_id}")
if doc_id not in self.documents:
raise FileNotFoundError(f"Document {doc_id} not found")
return self.documents[doc_id].copy()
def write_document(self, doc_id: str, content: str) -> bool:
"""
Write/update document without any security checks.
"""
print(f"RealDocumentService: Writing document {doc_id}")
if doc_id in self.documents:
self.documents[doc_id]["content"] = content
self.documents[doc_id]["modified_at"] = datetime.now().isoformat()
else:
self.documents[doc_id] = {
"content": content,
"created_at": datetime.now().isoformat(),
"owner": "unknown"
}
return True
def delete_document(self, doc_id: str) -> bool:
"""
Delete document without any security checks.
"""
print(f"RealDocumentService: Deleting document {doc_id}")
if doc_id in self.documents:
del self.documents[doc_id]
return True
return False
def list_documents(self) -> List[str]:
"""
List all documents without any security checks.
"""
print("RealDocumentService: Listing all documents")
return list(self.documents.keys())
class ProtectionProxy(DocumentService):
"""
Protection proxy that implements security, authentication, and authorization.
Controls access to the real document service based on user permissions.
"""
def __init__(self, real_service: DocumentService):
self.real_service = real_service
self.current_user: Optional[User] = None
self.access_log: List[Dict[str, Any]] = []
self.max_failed_attempts = 3
print("ProtectionProxy: Security layer initialized")
def authenticate(self, username: str, password: str) -> bool:
"""
Authenticate user with username and password.
In real implementation, this would check against a database.
"""
# Simulate user database
users_db = {
"admin": {
"password": "admin123",
"role": "administrator",
"permissions": [Permission.READ, Permission.WRITE, Permission.DELETE, Permission.ADMIN]
},
"writer": {
"password": "writer123",
"role": "content_writer",
"permissions": [Permission.READ, Permission.WRITE]
},
"reader": {
"password": "reader123",
"role": "viewer",
"permissions": [Permission.READ]
}
}
if username in users_db and users_db[username]["password"] == password:
user_data = users_db[username]
self.current_user = User(username, user_data["role"], user_data["permissions"])
self.current_user.failed_attempts = 0
self._log_access("AUTHENTICATION", "SUCCESS", f"User {username} authenticated")
print(f"ProtectionProxy: User {username} authenticated successfully")
return True
else:
if username in users_db:
# Simulate account lockout after failed attempts
if self.current_user and self.current_user.username == username:
self.current_user.failed_attempts += 1
if self.current_user.failed_attempts >= self.max_failed_attempts:
self.current_user.is_locked = True
print(f"ProtectionProxy: Account {username} locked due to failed attempts")
self._log_access("AUTHENTICATION", "FAILED", f"Failed login attempt for {username}")
print(f"ProtectionProxy: Authentication failed for {username}")
return False
def logout(self) -> None:
"""
Log out current user.
"""
if self.current_user:
self._log_access("LOGOUT", "SUCCESS", f"User {self.current_user.username} logged out")
print(f"ProtectionProxy: User {self.current_user.username} logged out")
self.current_user = None
else:
print("ProtectionProxy: No user to log out")
def _check_authentication(self) -> None:
"""
Check if user is authenticated.
"""
if not self.current_user:
raise PermissionError("User not authenticated. Please log in first.")
if self.current_user.is_locked:
raise PermissionError("Account is locked due to too many failed attempts.")
def _check_permission(self, required_permission: Permission) -> None:
"""
Check if current user has required permission.
"""
if not self.current_user.has_permission(required_permission):
raise PermissionError(f"User {self.current_user.username} does not have {required_permission.value} permission")
def _log_access(self, operation: str, status: str, details: str) -> None:
"""
Log access attempts for security auditing.
"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"user": self.current_user.username if self.current_user else "anonymous",
"operation": operation,
"status": status,
"details": details
}
self.access_log.append(log_entry)
def read_document(self, doc_id: str) -> Dict[str, Any]:
"""
Read document with security checks.
"""
self._check_authentication()
self._check_permission(Permission.READ)
try:
result = self.real_service.read_document(doc_id)
self._log_access("READ", "SUCCESS", f"Document {doc_id} read successfully")
return result
except Exception as e:
self._log_access("READ", "FAILED", f"Failed to read document {doc_id}: {str(e)}")
raise
def write_document(self, doc_id: str, content: str) -> bool:
"""
Write document with security checks.
"""
self._check_authentication()
self._check_permission(Permission.WRITE)
try:
result = self.real_service.write_document(doc_id, content)
self._log_access("WRITE", "SUCCESS", f"Document {doc_id} written successfully")
return result
except Exception as e:
self._log_access("WRITE", "FAILED", f"Failed to write document {doc_id}: {str(e)}")
raise
def delete_document(self, doc_id: str) -> bool:
"""
Delete document with security checks.
"""
self._check_authentication()
self._check_permission(Permission.DELETE)
try:
result = self.real_service.delete_document(doc_id)
self._log_access("DELETE", "SUCCESS", f"Document {doc_id} deleted successfully")
return result
except Exception as e:
self._log_access("DELETE", "FAILED", f"Failed to delete document {doc_id}: {str(e)}")
raise
def list_documents(self) -> List[str]:
"""
List documents with security checks.
"""
self._check_authentication()
self._check_permission(Permission.READ)
try:
result = self.real_service.list_documents()
self._log_access("LIST", "SUCCESS", "Documents listed successfully")
return result
except Exception as e:
self._log_access("LIST", "FAILED", f"Failed to list documents: {str(e)}")
raise
def get_access_log(self) -> List[Dict[str, Any]]:
"""
Get access log for security auditing (admin only).
"""
self._check_authentication()
self._check_permission(Permission.ADMIN)
return self.access_log.copy()
def demonstrate_protection_proxy():
"""
Demonstrate the Protection Proxy Pattern with different user roles.
"""
print("=== Protection Proxy Pattern Demonstration ===\n")
# Create real service and protection proxy
print("1. Setting up services")
real_service = RealDocumentService()
proxy = ProtectionProxy(real_service)
print()
# Test cases for different users
test_cases = [
("admin", "admin123", [
("list_documents", []),
("read_document", ["doc1"]),
("write_document", ["doc4", "New admin document"]),
("delete_document", ["doc3"])
]),
("writer", "writer123", [
("read_document", ["doc1"]),
("write_document", ["doc2", "Updated content"]),
("delete_document", ["doc1"]) # Should fail - no delete permission
]),
("reader", "reader123", [
("read_document", ["doc1"]),
("write_document", ["doc1", "Trying to write"]) # Should fail - no write permission
])
]
for username, password, operations in test_cases:
print(f"2. Testing with user: {username}")
# Authentication
if proxy.authenticate(username, password):
print(f" ✓ Authentication successful")
else:
print(f" ✗ Authentication failed")
continue
# Execute operations
for operation, args in operations:
try:
print(f" Attempting: {operation}({', '.join(map(str, args))})")
method = getattr(proxy, operation)
result = method(*args)
print(f" ✓ Success: {result}")
except PermissionError as e:
print(f" ✗ Permission denied: {e}")
except Exception as e:
print(f" ✗ Error: {e}")
proxy.logout()
print()
# Test failed authentication
print("3. Testing authentication failures")
print(" Attempting login with wrong password...")
if not proxy.authenticate("admin", "wrongpass"):
print(" ✓ Correctly rejected invalid credentials")
print(" Attempting operations without authentication...")
try:
proxy.read_document("doc1")
except PermissionError as e:
print(f" ✓ Correctly blocked unauthorized access: {e}")
print()
# Show access log (admin only)
print("4. Checking access log (admin only)")
if proxy.authenticate("admin", "admin123"):
access_log = proxy.get_access_log()
print(f" Total log entries: {len(access_log)}")
print(" Recent entries:")
for entry in access_log[-5:]: # Show last 5 entries
print(f" {entry['timestamp'][:19]} | {entry['user']:<8} | {entry['operation']:<12} | {entry['status']}")
proxy.logout()
if __name__ == "__main__":
demonstrate_protection_proxy()
This example demonstrates a smart reference proxy that provides advanced resource management features including reference counting, copy-on-write semantics, thread safety, and automatic cleanup. The proxy helps manage expensive resources efficiently while providing transparent access to clients.
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Any, Set
import time
import threading
from datetime import datetime
import weakref
class Resource(ABC):
"""
Abstract interface for managed resources.
Both real resource and smart reference proxy implement this interface.
"""
@abstractmethod
def process_data(self, data: str) -> str:
"""Process some data."""
pass
@abstractmethod
def get_info(self) -> Dict[str, Any]:
"""Get resource information."""
pass
class ExpensiveResource(Resource):
"""
Real subject representing an expensive resource that needs careful management.
This could be a database connection, file handle, or network connection.
"""
_instance_count = 0
_active_instances: Set['ExpensiveResource'] = set()
def __init__(self, resource_id: str) -> None:
ExpensiveResource._instance_count += 1
self.resource_id = resource_id
self.instance_number = ExpensiveResource._instance_count
self.created_at = datetime.now()
self.last_used = self.created_at
self.use_count = 0
self.is_active = True
# Simulate expensive initialization
time.sleep(0.1)
ExpensiveResource._active_instances.add(self)
print(f"ExpensiveResource: Created instance #{self.instance_number} ({resource_id})")
def process_data(self, data: str) -> str:
"""
Process data with the expensive resource.
"""
if not self.is_active:
raise RuntimeError(f"Resource {self.resource_id} has been disposed")
self.use_count += 1
self.last_used = datetime.now()
print(f"ExpensiveResource: Processing data with {self.resource_id} (use #{self.use_count})")
# Simulate processing time
time.sleep(0.05)
return f"Processed by {self.resource_id}: {data.upper()}"
def get_info(self) -> Dict[str, Any]:
"""
Get resource information.
"""
if not self.is_active:
raise RuntimeError(f"Resource {self.resource_id} has been disposed")
return {
"resource_id": self.resource_id,
"instance_number": self.instance_number,
"created_at": self.created_at.isoformat(),
"last_used": self.last_used.isoformat(),
"use_count": self.use_count,
"is_active": self.is_active
}
def dispose(self) -> None:
"""
Dispose of the resource and clean up.
"""
if self.is_active:
print(f"ExpensiveResource: Disposing instance #{self.instance_number} ({self.resource_id})")
self.is_active = False
ExpensiveResource._active_instances.discard(self)
def __del__(self) -> None:
"""
Destructor - ensure resource is properly disposed.
"""
self.dispose()
@classmethod
def get_active_count(cls) -> int:
"""Get number of active resource instances."""
return len(cls._active_instances)
class SmartReferenceProxy(Resource):
"""
Smart reference proxy that provides:
- Reference counting for automatic cleanup
- Copy-on-write semantics
- Thread-safe access
- Automatic disposal of unused resources
"""
def __init__(self, resource_id: str) -> None:
self.resource_id = resource_id
self._real_resource: Optional[ExpensiveResource] = None
self._reference_count = 0
self._lock = threading.Lock()
self._copied = False
# Weak reference to track proxy instances
self._weak_refs: List[weakref.ref] = []
print(f"SmartReferenceProxy: Created proxy for {resource_id}")
def _ensure_resource(self) -> ExpensiveResource:
"""
Ensure the real resource exists (lazy initialization).
"""
with self._lock:
if self._real_resource is None:
print(f"SmartReferenceProxy: Creating real resource for {self.resource_id}")
self._real_resource = ExpensiveResource(self.resource_id)
return self._real_resource
def _increment_reference(self) -> None:
"""
Increment reference count.
"""
with self._lock:
self._reference_count += 1
print(f"SmartReferenceProxy: Reference count for {self.resource_id}: {self._reference_count}")
def _decrement_reference(self) -> None:
"""
Decrement reference count and clean up if necessary.
"""
with self._lock:
self._reference_count -= 1
print(f"SmartReferenceProxy: Reference count for {self.resource_id}: {self._reference_count}")
if self._reference_count <= 0 and self._real_resource is not None:
print(f"SmartReferenceProxy: Auto-disposing resource {self.resource_id}")
self._real_resource.dispose()
self._real_resource = None
def acquire(self) -> 'SmartReferenceProxy':
"""
Acquire a reference to this proxy.
"""
self._increment_reference()
return self
def release(self) -> None:
"""
Release a reference to this proxy.
"""
self._decrement_reference()
def copy(self) -> 'SmartReferenceProxy':
"""
Create a copy of this proxy (copy-on-write).
"""
with self._lock:
if not self._copied and self._real_resource is not None:
# Mark as copied - future writes will create new instance
self._copied = True
print(f"SmartReferenceProxy: Marked {self.resource_id} for copy-on-write")
# Create new proxy instance
new_proxy = SmartReferenceProxy(f"{self.resource_id}_copy")
return new_proxy
def process_data(self, data: str) -> str:
"""
Process data with copy-on-write semantics.
"""
with self._lock:
# If this is a copied resource and we're about to write, create new instance
if self._copied and self._real_resource is not None:
print(f"SmartReferenceProxy: Copy-on-write triggered for {self.resource_id}")
old_resource = self._real_resource
self._real_resource = ExpensiveResource(f"{self.resource_id}_new")
self._copied = False
# Release reference to old resource
old_resource.dispose()
resource = self._ensure_resource()
return resource.process_data(data)
def get_info(self) -> Dict[str, Any]:
"""
Get resource information.
"""
resource = self._ensure_resource()
info = resource.get_info()
# Add proxy-specific information
info.update({
"proxy_reference_count": self._reference_count,
"proxy_copied": self._copied
})
return info
def __enter__(self) -> 'SmartReferenceProxy':
"""
Context manager entry - acquire reference.
"""
return self.acquire()
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
"""
Context manager exit - release reference.
"""
self.release()
class ResourceManager:
"""
Manager for handling multiple smart reference proxies.
"""
def __init__(self) -> None:
self._resources: Dict[str, SmartReferenceProxy] = {}
self._lock = threading.Lock()
def get_resource(self, resource_id: str) -> SmartReferenceProxy:
"""
Get or create a smart reference proxy for a resource.
"""
with self._lock:
if resource_id not in self._resources:
self._resources[resource_id] = SmartReferenceProxy(resource_id)
return self._resources[resource_id]
def cleanup_unused_resources(self) -> None:
"""
Clean up resources that are no longer referenced.
"""
with self._lock:
to_remove = []
for resource_id, proxy in self._resources.items():
if proxy._reference_count <= 0:
to_remove.append(resource_id)
for resource_id in to_remove:
print(f"ResourceManager: Cleaning up unused resource {resource_id}")
del self._resources[resource_id]
def get_stats(self) -> Dict[str, Any]:
"""
Get statistics about managed resources.
"""
return {
"total_proxies": len(self._resources),
"active_resources": ExpensiveResource.get_active_count(),
"total_created": ExpensiveResource._instance_count
}
def demonstrate_smart_reference_proxy():
"""
Demonstrate the Smart Reference Proxy Pattern.
"""
print("=== Smart Reference Proxy Pattern Demonstration ===\n")
# Create resource manager
manager = ResourceManager()
print("1. Basic resource management")
print(f" Initial stats: {manager.get_stats()}")
# Get resource and use it
resource1 = manager.get_resource("database_connection")
print(f" Stats after creating proxy: {manager.get_stats()}")
# Acquire reference and use resource
with resource1.acquire():
result = resource1.process_data("hello world")
print(f" Result: {result}")
info = resource1.get_info()
print(f" Resource info: {info['resource_id']}, uses: {info['use_count']}")
print(f" Stats after use: {manager.get_stats()}")
print()
print("2. Reference counting demonstration")
# Multiple references to same resource
ref1 = manager.get_resource("shared_resource").acquire()
ref2 = manager.get_resource("shared_resource").acquire()
ref3 = manager.get_resource("shared_resource").acquire()
print(f" Stats with multiple refs: {manager.get_stats()}")
# Release references one by one
ref1.release()
ref2.release()
print(f" Stats after releasing 2 refs: {manager.get_stats()}")
ref3.release()
print(f" Stats after releasing all refs: {manager.get_stats()}")
print()
print("3. Copy-on-write demonstration")
original = manager.get_resource("document").acquire()
# Process data with original
result1 = original.process_data("original data")
print(f" Original result: {result1}")
# Create copy
copy_proxy = original.copy()
copy_ref = copy_proxy.acquire()
# Process data with copy (should trigger copy-on-write)
result2 = copy_ref.process_data("modified data")
print(f" Copy result: {result2}")
# Process data with original again
result3 = original.process_data("more original data")
print(f" Original after copy: {result3}")
print(f" Stats after copy-on-write: {manager.get_stats()}")
# Clean up
original.release()
copy_ref.release()
print()
print("4. Context manager usage")
with manager.get_resource("temp_resource") as temp:
result = temp.process_data("temporary data")
print(f" Temp result: {result}")
print(f" Stats after context manager: {manager.get_stats()}")
print()
print("5. Resource cleanup")
manager.cleanup_unused_resources()
print(f" Final stats: {manager.get_stats()}")
if __name__ == "__main__":
demonstrate_smart_reference_proxy()
When to Use the Proxy Pattern
The Proxy Pattern is most beneficial when you need to:
- Control Access: Implement security, authentication, or authorization
- Improve Performance: Add caching, lazy loading, or connection pooling
- Add Functionality: Provide logging, monitoring, or debugging capabilities
- Manage Resources: Handle expensive object creation or cleanup automatically
- Provide Abstraction: Hide complexity of remote services or external systems
Real-World Applications
- Web Frameworks: Reverse proxies, load balancers, and API gateways
- Database Systems: Connection pooling and query result caching
- Security Systems: Authentication and authorization layers
- Distributed Systems: Service mesh and circuit breakers
- Development Tools: Debugging proxies and performance profilers