Proxy Pattern

1
2
3
4
5

+---------+       +-----------+        +-------------+
| Client  | --->  |   Proxy   | -----> | RealSubject |
+---------+       +-----------+        +-------------+
                         |
           [권한 검사, 로깅, 지연 로딩 등 수행]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

class RealImage:
    def __init__(self, filename):
        self.filename = filename
        self.load_from_disk()

    def load_from_disk(self):
        print(f"[LOAD] Loading {self.filename} from disk…")

    def display(self):
        print(f"[DISPLAY] Displaying {self.filename}")

class ImageProxy:
    def __init__(self, filename):
        self.filename = filename
        self._real_image = None

    def display(self):
        if not self._real_image:
            self._real_image = RealImage(self.filename)
        self._real_image.display()

# 사용
img = ImageProxy("photo.jpg")
img.display()  # 디스크에서 로딩 후 표시

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

class CommandExecutor:
    def run_command(self, cmd):
        print(f"[EXEC] Running command: {cmd}")

class CommandExecutorProxy:
    def __init__(self, user):
        self.user = user
        self._executor = CommandExecutor()

    def run_command(self, cmd):
        if self.user != "admin":
            raise PermissionError("You are not allowed to run this command.")
        self._executor.run_command(cmd)

# 사용
proxy = CommandExecutorProxy("guest")
proxy.run_command("rm -rf /")  # 접근 거부

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

import time

class RemoteService:
    def fetch_data(self):
        time.sleep(1)
        return "[DATA] Remote data from server"

class RemoteProxy:
    def __init__(self):
        self.remote = RemoteService()

    def fetch_data(self):
        print("[INFO] Connecting to remote…")
        return self.remote.fetch_data()

# 사용
proxy = RemoteProxy()
print(proxy.fetch_data())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

class Database:
    def query(self, sql):
        print(f"[DB] Executing SQL: {sql}")
        return f"Result for: {sql}"

class CachingProxy:
    def __init__(self):
        self._db = Database()
        self._cache = {}

    def query(self, sql):
        if sql in self._cache:
            print("[CACHE] Returning cached result")
            return self._cache[sql]
        result = self._db.query(sql)
        self._cache[sql] = result
        return result

# 사용
proxy = CachingProxy()
print(proxy.query("SELECT * FROM users"))
print(proxy.query("SELECT * FROM users"))  # 캐시 사용

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

class SmartResource:
    def __init__(self):
        self.usage_count = 0

    def use(self):
        self.usage_count += 1
        print(f"[USE] Resource used {self.usage_count} times.")

class SmartProxy:
    def __init__(self):
        self._resource = SmartResource()

    def use(self):
        print("[LOG] Before using resource")
        self._resource.use()
        print("[LOG] After using resource")

# 사용
proxy = SmartProxy()
proxy.use()
proxy.use()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

class APIService:
    def get(self):
        print("[API] Returning data from real API")

class APIProxy:
    def __init__(self):
        self._service = APIService()

    def get(self):
        print("[STATIC] Logging before call")
        self._service.get()

# 사용
proxy = APIProxy()
proxy.get()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

class RealService:
    def fetch(self):
        print("[SERVICE] Fetching data")

    def save(self):
        print("[SERVICE] Saving data")

class DynamicProxy:
    def __init__(self, target):
        self._target = target

    def __getattr__(self, name):
        attr = getattr(self._target, name)
        if callable(attr):
            def wrapper(*args, **kwargs):
                print(f"[DYNAMIC] Before calling '{name}'")
                result = attr(*args, **kwargs)
                print(f"[DYNAMIC] After calling '{name}'")
                return result
            return wrapper
        return attr

# 사용
proxy = DynamicProxy(RealService())
proxy.fetch()
proxy.save()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

class KafkaClient:
    def send(self, topic, message):
        print(f"[Kafka] Send to {topic}: {message}")

class KafkaProxy:
    def __init__(self, client):
        self.client = client

    def send(self, topic, message):
        print("[Proxy] Logging event")
        return self.client.send(topic, message)

# 인증 프록시 포함한 체인 구성
kafka = KafkaProxy(KafkaClient())
auth_kafka = AuthProxy(kafka)
auth_kafka.get_order = lambda x: kafka.send("order", {"id": x})  # 위임
auth_kafka.get_order("E100")

1
2
3
4
5
6

[User]
   |
   v
[DocumentProxy] ----> [Document]
   |
[권한 체크]

1
2

[Client] → [UserServiceProxy] → [UserServiceImpl(DB)]
      ↘ (캐시 확인/저장)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   Web Client    │───▶│  Image Proxy    │───▶│  Image Service  │
└─────────────────┘    │                 │    │                 │
                       │ • 지연 로딩      │    │ • 실제 이미지    │
                       │ • 캐싱          │    │ • 파일 시스템    │
                       │ • 접근 제어     │    │ • CDN 연동      │
                       └─────────────────┘    └─────────────────┘
                                │
                       ┌─────────────────┐
                       │  Cache Server   │
                       │                 │
                       │ • Redis         │
                       │ • 이미지 캐싱    │
                       └─────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

// 전자상거래 이미지 프록시 구현
public class ECommerceImageProxy implements Image {
    private RealImage realImage;
    private String productId;
    private String userId;
    private CacheService cacheService;
    private AuthService authService;
    
    public ECommerceImageProxy(String productId, String userId) {
        this.productId = productId;
        this.userId = userId;
        this.cacheService = new CacheService();
        this.authService = new AuthService();
    }
    
    @Override
    public void display() {
        // 1. 접근 권한 검증
        if (!authService.hasImageAccess(userId, productId)) {
            throw new SecurityException("Access denied for user: " + userId);
        }
        
        // 2. 캐시 확인
        byte[] cachedImage = cacheService.getImage(productId);
        if (cachedImage != null) {
            displayFromCache(cachedImage);
            return;
        }
        
        // 3. 지연 로딩 및 캐시 저장
        if (realImage == null) {
            realImage = new RealImage(productId);
        }
        
        byte[] imageData = realImage.getImageData();
        cacheService.cacheImage(productId, imageData);
        realImage.display();
    }
    
    private void displayFromCache(byte[] imageData) {
        System.out.println("Displaying cached image for product: " + productId);
        // 캐시된 이미지 표시 로직
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

from abc import ABC, abstractmethod

# Subject 인터페이스
class OrderService(ABC):
    @abstractmethod
    def get_order(self, order_id: str):
        pass

# 실제 서비스
class RealOrderService(OrderService):
    def get_order(self, order_id):
        print(f"[Real] Fetching order {order_id}")
        return {"id": order_id, "status": "shipped"}

# 인증 프록시
class AuthProxy(OrderService):
    def __init__(self, real_service: OrderService):
        self.real_service = real_service

    def get_order(self, order_id):
        if not self._is_authorized():
            raise PermissionError("Unauthorized access")
        print("[AuthProxy] Authorized request")
        return self.real_service.get_order(order_id)

    def _is_authorized(self):
        return True  # 실제로는 토큰 기반 인증

# 사용 예
service = AuthProxy(RealOrderService())
print(service.get_order("A1234"))

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211

from abc import ABC, abstractmethod
import time
import hashlib
from typing import Optional, Dict, Any
import logging

# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ImageService(ABC):
    """이미지 서비스 인터페이스 - Subject 역할"""
    
    @abstractmethod
    def load_image(self, image_id: str, size: tuple = None) -> Dict[str, Any]:
        """이미지 로드 추상 메서드"""
        pass

class RealImageService(ImageService):
    """실제 이미지 서비스 - RealSubject 역할"""
    
    def __init__(self):
        self.storage_path = "/images/storage/"
        
    def load_image(self, image_id: str, size: tuple = None) -> Dict[str, Any]:
        """실제 이미지 로딩 로직 - 시간이 오래 걸리는 작업 시뮬레이션"""
        logger.info(f"원본 이미지 로딩 시작: {image_id}")
        
        # 시간이 오래 걸리는 작업 시뮬레이션 (네트워크 I/O, 파일 시스템 접근)
        time.sleep(2)  # 2초 지연 시뮬레이션
        
        # 실제 이미지 데이터 로딩 및 처리
        image_data = {
            "id": image_id,
            "data": f"binary_data_for_{image_id}",
            "size": size if size else (1920, 1080),
            "format": "JPEG",
            "file_size": 2048000,  # 2MB
            "metadata": {
                "created_at": time.time(),
                "source": "original_storage"
            }
        }
        
        logger.info(f"원본 이미지 로딩 완료: {image_id}")
        return image_data

class ImageProxyService(ImageService):
    """이미지 프록시 서비스 - Proxy 역할"""
    
    def __init__(self):
        self._real_service: Optional[RealImageService] = None
        self._cache: Dict[str, Dict[str, Any]] = {}
        self._access_log: List[Dict[str, Any]] = []
        self._cache_ttl: int = 300  # 캐시 TTL: 5분
        
    def _get_real_service(self) -> RealImageService:
        """지연 초기화 - 실제 서비스 객체 생성"""
        if self._real_service is None:
            logger.info("실제 이미지 서비스 초기화")
            self._real_service = RealImageService()
        return self._real_service
    
    def _generate_cache_key(self, image_id: str, size: tuple = None) -> str:
        """캐시 키 생성"""
        cache_data = f"{image_id}_{size}" if size else image_id
        return hashlib.md5(cache_data.encode()).hexdigest()
    
    def _is_cache_valid(self, cache_entry: Dict[str, Any]) -> bool:
        """캐시 유효성 검사"""
        current_time = time.time()
        cache_time = cache_entry.get("cached_at", 0)
        return (current_time - cache_time) < self._cache_ttl
    
    def _log_access(self, image_id: str, cache_hit: bool, response_time: float):
        """접근 로그 기록"""
        log_entry = {
            "image_id": image_id,
            "timestamp": time.time(),
            "cache_hit": cache_hit,
            "response_time": response_time,
            "ip_address": "192.168.1.100"  # 시뮬레이션
        }
        self._access_log.append(log_entry)
        logger.info(f"접근 로그: {log_entry}")
    
    def _check_access_permission(self, image_id: str) -> bool:
        """접근 권한 검사 - 보호 프록시 기능"""
        # 실제 환경에서는 사용자 인증, 권한 검사 등을 수행
        if image_id.startswith("private_"):
            logger.warning(f"비공개 이미지 접근 차단: {image_id}")
            return False
        return True
    
    def _optimize_image(self, image_data: Dict[str, Any], size: tuple = None) -> Dict[str, Any]:
        """이미지 최적화 - 스마트 프록시 기능"""
        if size and size != image_data["size"]:
            # 이미지 리사이징 시뮬레이션
            optimized_data = image_data.copy()
            optimized_data["size"] = size
            optimized_data["file_size"] = int(image_data["file_size"] * 0.6)  # 40% 크기 감소
            optimized_data["metadata"]["optimized"] = True
            optimized_data["metadata"]["optimization_time"] = time.time()
            
            logger.info(f"이미지 최적화 완료: {image_data['id']} -> {size}")
            return optimized_data
        
        return image_data
    
    def load_image(self, image_id: str, size: tuple = None) -> Dict[str, Any]:
        """프록시를 통한 이미지 로딩"""
        start_time = time.time()
        
        # 1. 접근 권한 검사
        if not self._check_access_permission(image_id):
            raise PermissionError(f"이미지 접근 권한이 없습니다: {image_id}")
        
        # 2. 캐시 확인
        cache_key = self._generate_cache_key(image_id, size)
        
        if cache_key in self._cache:
            cache_entry = self._cache[cache_key]
            if self._is_cache_valid(cache_entry):
                # 캐시 히트
                response_time = time.time() - start_time
                self._log_access(image_id, cache_hit=True, response_time=response_time)
                logger.info(f"캐시에서 이미지 반환: {image_id}")
                return cache_entry["data"]
            else:
                # 캐시 만료
                del self._cache[cache_key]
                logger.info(f"만료된 캐시 삭제: {image_id}")
        
        # 3. 캐시 미스 - 실제 서비스에서 이미지 로드
        real_service = self._get_real_service()
        image_data = real_service.load_image(image_id, None)  # 원본 크기로 로드
        
        # 4. 이미지 최적화
        optimized_data = self._optimize_image(image_data, size)
        
        # 5. 캐시에 저장
        self._cache[cache_key] = {
            "data": optimized_data,
            "cached_at": time.time()
        }
        
        # 6. 접근 로그 기록
        response_time = time.time() - start_time
        self._log_access(image_id, cache_hit=False, response_time=response_time)
        
        return optimized_data
    
    def get_cache_stats(self) -> Dict[str, Any]:
        """캐시 통계 정보 반환"""
        total_requests = len(self._access_log)
        cache_hits = sum(1 for log in self._access_log if log["cache_hit"])
        
        return {
            "total_requests": total_requests,
            "cache_hits": cache_hits,
            "cache_hit_rate": cache_hits / total_requests if total_requests > 0 else 0,
            "cached_images": len(self._cache),
            "average_response_time": sum(log["response_time"] for log in self._access_log) / total_requests if total_requests > 0 else 0
        }

# 사용 예시
def main():
    """프록시 패턴 사용 예시"""
    
    # 이미지 프록시 서비스 생성 (실제 서비스는 지연 초기화)
    image_service = ImageProxyService()
    
    print("=== 이미지 프록시 서비스 테스트 ===\n")
    
    # 첫 번째 요청 - 캐시 미스, 실제 서비스 호출
    print("1. 첫 번째 이미지 요청 (캐시 미스)")
    image1 = image_service.load_image("product_001", size=(800, 600))
    print(f"로드된 이미지: {image1['id']}, 크기: {image1['size']}")
    print()
    
    # 두 번째 요청 - 동일 이미지, 캐시 히트
    print("2. 동일 이미지 재요청 (캐시 히트)")
    image2 = image_service.load_image("product_001", size=(800, 600))
    print(f"로드된 이미지: {image2['id']}, 크기: {image2['size']}")
    print()
    
    # 세 번째 요청 - 다른 크기, 캐시 미스
    print("3. 다른 크기로 이미지 요청 (캐시 미스)")
    image3 = image_service.load_image("product_001", size=(400, 300))
    print(f"로드된 이미지: {image3['id']}, 크기: {image3['size']}")
    print()
    
    # 네 번째 요청 - 접근 권한 없는 이미지
    print("4. 비공개 이미지 접근 시도 (권한 오류)")
    try:
        image4 = image_service.load_image("private_secret_001")
    except PermissionError as e:
        print(f"오류: {e}")
    print()
    
    # 캐시 통계 출력
    print("5. 캐시 통계")
    stats = image_service.get_cache_stats()
    print(f"총 요청 수: {stats['total_requests']}")
    print(f"캐시 히트 수: {stats['cache_hits']}")
    print(f"캐시 히트율: {stats['cache_hit_rate']:%}")
    print(f"캐시된 이미지 수: {stats['cached_images']}")
    print(f"평균 응답 시간: {stats['average_response_time']:f}초")

if __name__ == "__main__":
    main()