Контроль доступа к объектам. Virtual Proxy для отложенной загрузки. Property-based lazy loading, __getattr__ и дескрипторы.
Proxy — это не просто обёртка, это контроль доступа к объекту.
Proxy — структурный паттерн, где объект-заместитель контролирует доступ к другому объекту.
# ❌ ПЛОХО: Логика доступа размазана по коду
class Image:
def __init__(self, filename: str):
print(f"Loading {filename}")
self.filename = filename
self.data = self._load_image()
def _load_image(self):
# Дорогостоящая загрузка
return "image_data"
def display(self):
print(f"Displaying {self.filename}")
# Клиент должен помнить о проверках
image = Image("photo.jpg") # Загружается сразу!
if user.is_authenticated:
image.display()# ✅ ХОРОШО: Загрузка при первом использовании
class Image:
def __init__(self, filename: str):
self.filename = filename
self._data = None
def _load_image(self):
print(f"Loading {self.filename}")
self._data = "image_data"
def display(self):
if self._data is None:
self._load_image()
print(f"Displaying {self.filename}")
class ImageProxy:
"""Virtual Proxy — отложенная загрузка"""
def __init__(self, filename: str):
self.filename = filename
self._image = None
def display(self):
if self._image is None:
self._image = Image(self.filename)
self._image.display()
# Использование
image = ImageProxy("photo.jpg") # Не загружается
image.display() # Загружается и отображается
image.display() # Используется кэш# ✅ Protection Proxy — проверка прав
from datetime import datetime
class Document:
def __init__(self, content: str):
self.content = content
def display(self, user: str):
print(f"Content: {self.content}")
class DocumentProxy:
def __init__(self, content: str, owner: str):
self._document = Document(content)
self._owner = owner
def display(self, user: str):
if user != self._owner:
print(f"Access denied: {user} is not {self._owner}")
return
self._document.display(user)
# Использование
doc = DocumentProxy("Secret", "alice")
doc.display("alice") # Content: Secret
doc.display("bob") # Access denied# ✅ Smart Proxy — логирование и кэширование
class RealSubject:
def request(self) -> str:
print("RealSubject: Handling request")
return "Response"
class Proxy:
def __init__(self, subject: RealSubject):
self._subject = subject
self._cache = {}
self._request_count = 0
def request(self, query: str) -> str:
self._request_count += 1
print(f"Proxy: Request #{self._request_count}")
# Кэширование
if query in self._cache:
print("Proxy: Returning cached result")
return self._cache[query]
# Логирование
print(f"Proxy: Logging query '{query}'")
# Делегирование
result = self._subject.request()
self._cache[query] = result
return result
# Использование
real = RealSubject()
proxy = Proxy(real)
proxy.request("query1") # Выполняется, кэшируется
proxy.request("query1") # Из кэша
proxy.request("query2") # Выполняется, кэшируется# ✅ Lazy Loading через property
class DatabaseConnection:
def __init__(self, url: str):
self.url = url
self._connection = None
@property
def connection(self):
if self._connection is None:
print(f"Connecting to {self.url}")
self._connection = self._create_connection()
return self._connection
def _create_connection(self):
# Имитация подключения
return f"Connection({self.url})"
def query(self, sql: str):
conn = self.connection # Ленивая инициализация
print(f"Executing: {sql}")
return []
# Использование
db = DatabaseConnection("postgresql://localhost")
# Ничего не загружается
result = db.query("SELECT * FROM users")
# Connecting to postgresql://localhost
# Executing: SELECT * FROM users# ✅ Lazy Loading через __getattr__
class LazyLoader:
def __init__(self, module_name: str):
self._module_name = module_name
self._module = None
def __getattr__(self, name: str):
if self._module is None:
print(f"Loading module: {self._module_name}")
import importlib
self._module = importlib.import_module(self._module_name)
return getattr(self._module, name)
# Использование
requests = LazyLoader("requests")
# Модуль не загружается
response = requests.get("https://example.com")
# Loading module: requests
# Выполняется requests.get()# ✅ Lazy Loading через дескриптор
class LazyProperty:
"""Дескриптор для ленивого вычисления"""
def __init__(self, func):
self.func = func
self.name = func.__name__
def __get__(self, obj, objtype=None):
if obj is None:
return self
print(f"Computing {self.name}")
value = self.func(obj)
setattr(obj, self.name, value) # Кэширование
return value
class Data:
@LazyProperty
def expensive_data(self):
import time
time.sleep(1) # Имитация вычисления
return [1, 2, 3, 4, 5]
# Использование
data = Data()
print(data.expensive_data) # Вычисляется (1 сек)
print(data.expensive_data) # Из кэша (мгновенно)import time
from functools import wraps
class RateLimiter:
"""Rate Limiting Proxy"""
def __init__(self, calls_per_second: int):
self.min_interval = 1.0 / calls_per_second
self.last_call = 0
def wait(self):
now = time.time()
elapsed = now - self.last_call
if elapsed < self.min_interval:
time.sleep(self.min_interval - elapsed)
self.last_call = time.time()
class APIClient:
def __init__(self, base_url: str):
self.base_url = base_url
def get(self, endpoint: str):
print(f"GET {self.base_url}/{endpoint}")
return {"status": "ok"}
class RateLimitedProxy:
def __init__(self, client: APIClient, rate_limit: int):
self._client = client
self._limiter = RateLimiter(rate_limit)
def get(self, endpoint: str):
self._limiter.wait()
return self._client.get(endpoint)
# Использование
api = APIClient("https://api.example.com")
proxy = RateLimitedProxy(api, rate_limit=2) # 2 вызова/сек
proxy.get("users") # Выполняется сразу
proxy.get("posts") # Ждёт 0.5 сек
proxy.get("comments") # Ждёт 0.5 секclass RealPaymentGateway:
def charge(self, amount: float) -> bool:
# Реальное списание денег
print(f"Charging ${amount}")
return True
class MockPaymentGateway:
def charge(self, amount: float) -> bool:
# Mock для тестов
print(f"[MOCK] Would charge ${amount}")
return True
class PaymentGatewayProxy:
"""Proxy для переключения real/mock"""
def __init__(self, use_mock: bool = False):
if use_mock:
self._gateway = MockPaymentGateway()
else:
self._gateway = RealPaymentGateway()
def charge(self, amount: float) -> bool:
return self._gateway.charge(amount)
# В production
gateway = PaymentGatewayProxy(use_mock=False)
# В тестах
gateway = PaymentGatewayProxy(use_mock=True)from typing import List, Optional
class LargeDataset:
def __init__(self, filename: str):
self.filename = filename
self._data: Optional[List] = None
def _load(self):
print(f"Loading {self.filename}")
with open(self.filename) as f:
self._data = f.readlines()
def __len__(self) -> int:
if self._data is None:
self._load()
return len(self._data)
def __getitem__(self, index):
if self._data is None:
self._load()
return self._data[index]
def __iter__(self):
if self._data is None:
self._load()
return iter(self._data)
# Использование
dataset = LargeDataset("huge_file.txt")
# Файл не загружается
print(f"Total lines: {len(dataset)}") # Загружается
print(f"First line: {dataset[0]}") # Из кэша# Django QuerySet — это Lazy Proxy
from myapp.models import User
# Не выполняется запрос
qs = User.objects.filter(active=True)
# Выполняется запрос при итерации
for user in qs:
print(user.name)
# Или при access
first = qs[0]
# Кэширование
list(qs) # Запрос выполняется
list(qs) # Из кэшаИзучите тему Chain of Responsibility и Middleware для цепочек обработчиков.
Далее: Chain of Responsibility и Middleware