-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathobject-pool-context.py
More file actions
60 lines (43 loc) · 1.26 KB
/
Copy pathobject-pool-context.py
File metadata and controls
60 lines (43 loc) · 1.26 KB
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
"""
Now we are going to create a context manager that automatically
acquires and releases objects
"""
from typing import List
class PoolManager():
def __init__(self, pool):
self.pool = pool
def __enter__(self):
self.obj = self.pool.acquire()
return self.obj
def __exit__(self, type, value, traceback):
self.pool.release(self.obj)
class Reusable:
def test(self):
print(f"Using object {id(self)}")
class ReusablePool:
def __init__(self, size):
self.size = size
self.free = []
self.in_use = []
for _ in range(size):
self.free.append(Reusable())
def acquire(self) -> Reusable:
if(len(self.free) <= 0):
raise Exception("No more objects are available")
r = self.free[0]
self.free.remove(r)
self.in_use.append(r)
return r
def release(self, r: Reusable):
self.in_use.remove(r)
self.free.append(r)
pool = ReusablePool(2)
with PoolManager(pool) as r:
r.test()
with PoolManager(pool) as r2:
r2.test()
with PoolManager(pool) as r3:
r3.test()
# you need reset your object when you release it
# object pool is responsible to reset.
# if you dont reset properly, are gonna to leaking information