代理模式Proxy Pattern, since 2024-06

(2024.06.02 Sun)
代理模式是结构模式(structural pattern)的一种，为真实对象提供了代替用的接口，扮演中介(intermediary)、代理(surrogate)的角色。代理接收到对真实对象的访问/请求，执行与真实对象相同的功能和动作，可加入额外的功能且不需要修改真实对象。

使用场景

若干可能使用代理模式的场景：

• 真实对象加载开销过大(expensive object loading)：处理复杂或资源密集型(resource-intensive)对象时，使用虚拟代理(virtual proxy)，可扮演placeholder的角色，仅在使用时加载对象的全部(load the full object on demand)，优化资源使用
• 远程对象访问：对象在其他地址空间，用户需要以访问本地资源的方式访问，适合建立远程代理(remote proxy)，用于管理连接技术细节(如建立网络连接、序列化serialisation、去序列化)，使得远程对象用起来像本地对象
• 增强安全性：保护代理(protection proxy)可增强接入控制，比如加入客户端的授权和验证(authorisation & authentication)，保护敏感资源等
• 智能引用(smart reference)：释放不再使用的重型对象(heavyweight object)场景，该代理跟踪活跃用户并周期检查，如果没有客户端使用对象，代理则释放对象和系统资源。也监测对对象的修改，允许在不修改的情况下重用

案例

概念案例 conceptual example

该案例用于解释

• 代理模式的由哪些类组成
• 每个类扮演什么角色
• 模式间的元素如何关联

首先定义一个Subject接口，声明RealSubject和Proxy两个类的基本操作。只要使用了该接口的RealSubject类在客户端可用，用户就可以使用Proxy类代替RealSubject类。

from abc import ABC, abstractmethod

class Subject(ABC):
    @abstractmethod
    def request(self) -> None:
        pass

接下来定义RealSubject类，该类中包含核心业务逻辑。而Proxy类与RealSubject有相同接口，可以解决RealSubject中或反应慢或敏感的问题，比如矫正输入数据等。Proxy中最常见的应用包括惰性加载(lazy loading)，缓存，控制连接(controlling the access)，记录日志等。Proxy可执行应用中的一个，并根据结果进行RealSubject相同的方法

class RealSubject(Subject):
    def request(self) -> None:
        print("RealSubject: Handling request.")

class Proxy(Subject):
    def __init__(self, real_subject: RealSubject) -> None:
        self._real_subject = real_subject

    def request(self) -> None:
        if self.check_access():
            self._real_subject.request()
            self.log_access()

    def check_access(self) -> bool:
        print("Proxy: Checking access prior to firing a real request.")
        return True

    def log_access(self) -> None:
        print("Proxy: Logging the time of request.", end="")

下面是客户单代码，客户端通过Subject接口和所有对象交互。

def client_code(subject: Subject) -> None:
    subject.request()

下面运行这些代码

if __name__ == "__main__":
    print("Client: Executing the client code with a real subject:")
    real_subject = RealSubject()
    client_code(real_subject)
    print("Client: Executing the same client code with a proxy:")
    proxy = Proxy(real_subject)
    client_code(proxy)

运行结果如下

Client: Executing the client code with a real subject:
RealSubject: Handling request.
Client: Executing the same client code with a proxy:
Proxy: Checking access prior to firing a real request.
RealSubject: Handling request.
Proxy: Logging the time of request.

数据库缓存 Caching Proxy for Database Queries

(2024.06.03 Mon at KLN)
该案例中，CacheProxy设计成为客户端与数据库对象RealDatabaseQuery之间的媒介，检测查询结果是否已经进入缓存，并在规定的缓存周期内(specified cache duration)返回缓存结果。

如果缓存已经过期或结果不在缓存中，CacheProxy代理从RealDatabaseQuery中执行查询，将结果推入缓存，并返回给客户端。CacheProxy降低了数据库服务器的负载，通过缓存的方式加速了数据查询的速度和效率。

首先定义数据库接口DatabaseQuery，定义了抽象类的基本方法。

from abc import ABC, abstractmethod
import time

class DatabaseQuery(ABC):
    @abstractmethod
    def execute_query(self, query: str):
        pass

定义数据库类RealDatabaseQuery

class RealDatabaseQuery(DatabaseQuery):
    def execute_query(self, query: str):
        print(f"Execute query: {query}")
        # do something 
        return f"results for query: {query}"

定义数据查询的代理CacheProxy

class CacheProxy(DatabaseQuery):
    def __init__(self, real_database_query: DatabaseQuery, cache_duration: int):
        self._real_database_query = real_database_query
        self._cache_duration = cache_duration
        self._cache = {}

    def execute_query(self, query: str) -> Union[str, int]:
        if query in self._cache and time.time() - self._cache[query]["timestamp"] <= self._cache_duration:
            # results in cache
            print(f"Returning cached results: {query}")
            return self._cache[query]["result"]
        else:
            # perform query in RealDatabaseQuery
            result = self._real_database_query.execute_query(query)
            self._cache[query] = {"result": result, "timestamp": time.time()}
            return result

客户端代码

if __name__ == "__main__":
    real_database_query = RealDatabaseQuery()
    cache_proxy = CacheProxy(real_database_query, 3) 
    # query for the 1st time, will query real database directly
    print(cache_proxy.execute_query("SELECT * FROM table_1")
    print(cache_proxy.execute_query("SELECT * FROM table_2")
    time.sleep(2)
    # query for the 2nd time, will read cache which with shorter response time
    print(cache_proxy.execute_query("SELECT * FROM table_1")
    print(cache_proxy.execute_query("SELECT * FROM table_2")

性能监控代理 Monitoring Proxy for performance metrics

(2024.06.05 Wed)
该代理用于对真实对象的执行性能做统计，记录日志且不会影响真实对象的功能。该代理可惰性实例化(lazily instantiate)真实对象，确保系统资源被保留到对象运行之时。

定义接口

from abc import ABC, abstractmethod
from datetime import datetime

class Subject(ABC):
    @abstractmethod
    def perform_operation(self):
        pass

定义真实对象

class RealSubject(Subject):
    def perform_operation(self):
        """
        The Real Subject's method representing a costly operation.
        """
        print("RealSubject: Performing a costly operation...")

定义代理对象

class MonitoringProxy(Subject):
    def __init__(self):
        self._real_subject = None

    def perform_operation(self):
        """
        The Proxy's method, which monitors and adds performance metrics.
        Lazily instantiates the Real Subject on the first call.
        """
        if self._real_subject is None:
            print("MonitoringProxy: Lazy loading the RealSubject...")
            self._real_subject = RealSubject()
        
        start_time = datetime.now()
        print(
          f"MonitoringProxy: Recording operation start time - {start_time}"
        )
        
        # Delegate the operation to the Real Subject
        self._real_subject.perform_operation()
        
        end_time = datetime.now()
        execution_time = end_time - start_time
        print(f"MonitoringProxy: Recording operation end time - {end_time}")
        print(
          f"MonitoringProxy: Operation executed in {execution_time} seconds"
        )

定义客户端

if __name__ == "__main__":
    monitoring_proxy = MonitoringProxy()
    monitoring_proxy.perform_operation()

当代理对象调用perform_operation方法时，MonitoringProxy记录运行的开始和结束时间，以计算总运行时间，且无需改变RealSubject对象的行为。

其他应用案例

• Virtual Proxy for Image Loading: In photo editing apps, a Virtual Proxy represents high-resolution images, loading them on demand, optimizing memory, and enhancing startup times.

• Remote Proxy for Web Services: When dealing with remote web services or APIs, a Remote Proxy simplifies communication, handling authentication, encryption, and network connectivity details.

• Lazy Loading for Expensive Objects: In large datasets or complex objects, use a Proxy to delay object creation until necessary, common in CAD software.

• Access Control with Security Proxy: Ensure secure data access by employing a Security Proxy, enforcing permissions and authentication checks.

• Logging Proxy for Debugging: Debugging is streamlined with a Logging Proxy recording method calls and parameters, aiding issue diagnosis without code modifications.

• Counting Proxy for Resource Management: In resource-critical systems like concurrent programming, a Counting Proxy tracks object references, releasing them when no longer required.

• Load Balancing with Proxy Servers: Proxy servers in web apps distribute network requests across backend servers, ensuring availability and load balancing.

• Transactional Proxy for Database Operations: Database apps benefit from a Transactional Proxy handling transactions, ensuring data integrity through rollback and commit mechanisms.

Reference

1 medium,Design Patterns in Python: Proxy, The Powerful Surrogate, by Amir Lavasani
2 Refactoring Guru, Proxy in Python