StructuralPythonverifiedVerified

Proxy Pattern in Python

Provides a surrogate or placeholder for another object to control access, add lazy initialization, caching, logging, or access control.

How to Implement the Proxy Pattern in Python

1Step 1: Define the subject interface and real implementation

from typing import Protocol


class Subject(Protocol):
    def request(self) -> None: ...


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

2Step 2: Create the proxy with access control and logging

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

    def _check_access(self) -> bool:
        print("Proxy: checking access before forwarding request.")
        return True

    def _log_access(self) -> None:
        print("Proxy: logging time of request.")

    def request(self) -> None:
        if self._check_access():
            self._real_subject.request()
            self._log_access()

3Step 3: Access the real subject through the proxy

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


real = RealSubject()
client_code(real)

print("---")

proxy = ProxySubject(real)
client_code(proxy)

"""Caching Proxy with TTL for any async data fetcher."""

import logging
import time
from typing import Any, Generic, Protocol, TypeVar

logger = logging.getLogger(__name__)

T = TypeVar("T")


# [step] Define the data source protocol
class DataSource(Protocol[T]):
    async def fetch(self, key: str) -> T: ...


# [step] Implement the caching proxy with per-key TTL
class CachingProxy(Generic[T]):
    def __init__(self, origin: DataSource[T], ttl_ms: int = 60_000) -> None:
        self._origin = origin
        self._ttl_ms = ttl_ms
        self._cache: dict[str, tuple[T, float]] = {}

    async def fetch(self, key: str) -> T:
        cached = self._cache.get(key)
        if cached and time.monotonic() < cached[1]:
            logger.info('[Cache HIT]  key="%s"', key)
            return cached[0]

        logger.info('[Cache MISS] key="%s" -- fetching from origin', key)
        value = await self._origin.fetch(key)
        self._cache[key] = (value, time.monotonic() + self._ttl_ms / 1000)
        return value

    def invalidate(self, key: str) -> None:
        self._cache.pop(key, None)
        logger.info('[Cache] invalidated key="%s"', key)

    def invalidate_all(self) -> None:
        self._cache.clear()
        logger.info("[Cache] cleared all entries")


# [step] Example origin and usage
from dataclasses import dataclass


@dataclass(frozen=True)
class WeatherData:
    temp: float
    condition: str


class WeatherApiClient:
    async def fetch(self, city: str) -> WeatherData:
        logger.info("[API] GET /weather?city=%s", city)
        import asyncio
        await asyncio.sleep(0.05)
        return WeatherData(temp=22.0, condition="sunny")


async def main() -> None:
    import asyncio
    proxy: CachingProxy[WeatherData] = CachingProxy(WeatherApiClient(), ttl_ms=5_000)
    await proxy.fetch("London")   # MISS
    await proxy.fetch("London")   # HIT
    proxy.invalidate("London")
    await proxy.fetch("London")   # MISS


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

Proxy Pattern Architecture

hourglass_empty

Rendering diagram...

lightbulb

Proxy Pattern in the Real World

“A corporate receptionist acts as a proxy for the CEO. When someone wants to meet the CEO, the receptionist checks credentials, schedules the meeting, and logs the visit before granting access. The visitor interacts with the receptionist using the same protocol they would use with the CEO—the receptionist simply adds control and record-keeping around that access.”