FilesExpand file tree

master

/

proxy_pattern_demo.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

186 lines (147 loc) · 6.16 KB

master

/

proxy_pattern_demo.py

Copy path

Top

File metadata and controls

• Code
• Blame

186 lines (147 loc) · 6.16 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

from abc import ABC, abstractmethod

import time

import os

# Subject Interface

class DataSource(ABC):

"""Abstract interface for data sources."""

@abstractmethod

def read_data(self) -> str:

"""Read data from the source."""

pass

@abstractmethod

def write_data(self, data: str) -> None:

"""Write data to the source."""

pass

# Real Subject

class RealFileDataSource(DataSource):

"""Real implementation that actually reads/writes files."""

def __init__(self, filename: str):

self.filename = filename

print(f"RealFileDataSource: Opening file {filename}")

def read_data(self) -> str:

print(f"RealFileDataSource: Reading from {self.filename}")

try:

with open(self.filename, 'r') as file:

return file.read()

except FileNotFoundError:

return ""

def write_data(self, data: str) -> None:

print(f"RealFileDataSource: Writing to {self.filename}")

with open(self.filename, 'w') as file:

file.write(data)

# Virtual Proxy - Lazy Loading

class LazyFileDataSource(DataSource):

"""Virtual proxy that creates the real object only when needed."""

def __init__(self, filename: str):

self.filename = filename

self._real_datasource = None

print(f"LazyFileDataSource: Proxy created for {filename} (not opened yet)")

def _get_real_datasource(self) -> RealFileDataSource:

"""Lazy initialization of the real data source."""

if self._real_datasource is None:

print(f"LazyFileDataSource: Initializing real data source for {self.filename}")

self._real_datasource = RealFileDataSource(self.filename)

return self._real_datasource

def read_data(self) -> str:

return self._get_real_datasource().read_data()

def write_data(self, data: str) -> None:

self._get_real_datasource().write_data(data)

# Protection Proxy - Access Control

class ProtectedFileDataSource(DataSource):

"""Protection proxy that controls access based on user permissions."""

def __init__(self, datasource: DataSource, user_role: str):

self._datasource = datasource

self.user_role = user_role

self.allowed_roles = ["admin", "editor"]

def _check_permission(self, operation: str) -> bool:

"""Check if the user has permission for the operation."""

if self.user_role not in self.allowed_roles:

print(f"ProtectedFileDataSource: Access denied for {self.user_role} - {operation} not allowed")

return False

print(f"ProtectedFileDataSource: Access granted for {self.user_role} - {operation}")

return True

def read_data(self) -> str:

if self._check_permission("read"):

return self._datasource.read_data()

return "Access denied"

def write_data(self, data: str) -> None:

if self._check_permission("write"):

self._datasource.write_data(data)

else:

print("Write operation blocked")

# Smart Proxy - Caching and Logging

class SmartFileDataSource(DataSource):

"""Smart proxy that adds caching and logging functionality."""

def __init__(self, datasource: DataSource):

self._datasource = datasource

self._cache = {}

self._access_log = []

def read_data(self) -> str:

cache_key = "read_data"

if cache_key in self._cache:

print("SmartFileDataSource: Returning cached data")

self._log_access("read", "cache")

return self._cache[cache_key]

print("SmartFileDataSource: Reading fresh data")

data = self._datasource.read_data()

self._cache[cache_key] = data

self._log_access("read", "fresh")

return data

def write_data(self, data: str) -> None:

print("SmartFileDataSource: Writing data and clearing cache")

self._datasource.write_data(data)

self._cache.clear() # Invalidate cache on write

self._log_access("write", "direct")

def _log_access(self, operation: str, source: str) -> None:

"""Log access operations."""

timestamp = time.strftime("%H:%M:%S")

log_entry = f"[{timestamp}] {operation} from {source}"

self._access_log.append(log_entry)

def get_access_log(self) -> list:

"""Get the access log."""

return self._access_log.copy()

# Client code

def demonstrate_proxy(proxy: DataSource, name: str) -> None:

"""Demonstrate proxy functionality."""

print(f"\n--- Demonstrating {name} ---")

# Read data

data = proxy.read_data()

print(f"Read data: '{data}'")

# Write data

proxy.write_data("Hello, Proxy Pattern!")

# Read again

data = proxy.read_data()

print(f"Read data again: '{data}'")

if __name__ == "__main__":

print("Demonstrating Proxy Design Pattern in Python:")

print("=" * 60)

filename = "test_file.txt"

# Clean up any existing file

if os.path.exists(filename):

os.remove(filename)

# Virtual Proxy demonstration

print("\n1. Virtual Proxy (Lazy Loading):")

lazy_proxy = LazyFileDataSource(filename)

demonstrate_proxy(lazy_proxy, "Lazy Proxy")

# Protection Proxy demonstration

print("\n2. Protection Proxy (Access Control):")

real_ds = RealFileDataSource(filename)

protected_admin = ProtectedFileDataSource(real_ds, "admin")

protected_guest = ProtectedFileDataSource(real_ds, "guest")

print("Admin access:")

demonstrate_proxy(protected_admin, "Protected Proxy (Admin)")

print("\nGuest access:")

demonstrate_proxy(protected_guest, "Protected Proxy (Guest)")

# Smart Proxy demonstration

print("\n3. Smart Proxy (Caching and Logging):")

smart_proxy = SmartFileDataSource(RealFileDataSource(filename))

demonstrate_proxy(smart_proxy, "Smart Proxy")

print("\nAccess log:")

for entry in smart_proxy.get_access_log():

print(f" {entry}")

# Clean up

if os.path.exists(filename):

os.remove(filename)

print("\nThe Proxy Pattern provides a surrogate that controls access")

print("to another object, adding functionality like lazy loading,")

print("access control, caching, and logging without changing the original object.")