0
# Scoping and Lifecycle Management
1
2
Object lifecycle control through built-in and custom scopes, managing when and how instances are created and shared across the application.
3
4
## Capabilities
5
6
### Built-in Scope Constants
7
8
Pre-defined scope identifiers for common object lifecycle patterns.
9
10
```python { .api }
11
SINGLETON: _SingletonScopeId
12
"""
13
Scope where a single instance is created and shared across all requests.
14
Default scope for implicit bindings.
15
"""
16
17
PROTOTYPE: _PrototypeScopeId
18
"""
19
Scope where a new instance is created for each request.
20
"""
21
```
22
23
### Scope Base Class
24
25
Abstract base class for implementing custom object scopes with specific lifecycle behaviors.
26
27
```python { .api }
28
class Scope(object):
29
def provide(self, binding_key, default_provider_fn):
30
"""
31
Provides an instance according to the scope's lifecycle policy.
32
33
Parameters:
34
- binding_key: unique identifier for the binding
35
- default_provider_fn: function that creates new instances
36
37
Returns:
38
Instance according to scope policy (cached, new, etc.)
39
"""
40
```
41
42
### Built-in Scope Implementations
43
44
Concrete implementations of common scope patterns.
45
46
```python { .api }
47
class SingletonScope(object):
48
def provide(self, binding_key, default_provider_fn):
49
"""
50
Returns cached instance if exists, creates and caches new instance otherwise.
51
Thread-safe implementation using re-entrant locks.
52
"""
53
54
class PrototypeScope(object):
55
def provide(self, binding_key, default_provider_fn):
56
"""
57
Always returns a new instance by calling default_provider_fn().
58
"""
59
```
60
61
## Usage Examples
62
63
### Default Singleton Scope
64
65
```python
66
import pinject
67
68
class DatabaseConnection(object):
69
def __init__(self):
70
self.connection_id = id(self)
71
print(f"Creating connection {self.connection_id}")
72
73
class ServiceA(object):
74
def __init__(self, database_connection):
75
self.db = database_connection
76
77
class ServiceB(object):
78
def __init__(self, database_connection):
79
self.db = database_connection
80
81
obj_graph = pinject.new_object_graph()
82
83
# Both services get the same connection instance (singleton is default)
84
service_a = obj_graph.provide(ServiceA) # "Creating connection ..."
85
service_b = obj_graph.provide(ServiceB) # No additional creation message
86
87
print(service_a.db.connection_id == service_b.db.connection_id) # True
88
```
89
90
### Explicit Prototype Scope
91
92
```python
93
import pinject
94
95
class RequestHandler(object):
96
def __init__(self):
97
self.request_id = id(self)
98
99
class MyBindingSpec(pinject.BindingSpec):
100
def configure(self, bind):
101
# Each request gets a new handler instance
102
bind('request_handler').to_class(RequestHandler, in_scope=pinject.PROTOTYPE)
103
104
class WebService(object):
105
def __init__(self, request_handler):
106
self.handler = request_handler
107
108
obj_graph = pinject.new_object_graph(binding_specs=[MyBindingSpec()])
109
110
# Each service gets a different handler instance
111
service1 = obj_graph.provide(WebService)
112
service2 = obj_graph.provide(WebService)
113
114
print(service1.handler.request_id == service2.handler.request_id) # False
115
```
116
117
### Provider Methods with Scopes
118
119
```python
120
import pinject
121
import threading
122
import time
123
124
class ThreadLocalScope(pinject.Scope):
125
def __init__(self):
126
self._local = threading.local()
127
128
def provide(self, binding_key, default_provider_fn):
129
if not hasattr(self._local, 'instances'):
130
self._local.instances = {}
131
132
if binding_key not in self._local.instances:
133
self._local.instances[binding_key] = default_provider_fn()
134
135
return self._local.instances[binding_key]
136
137
class SessionData(object):
138
def __init__(self):
139
self.thread_id = threading.current_thread().ident
140
self.created_at = time.time()
141
142
THREAD_LOCAL = "thread_local"
143
144
class ScopingBindingSpec(pinject.BindingSpec):
145
@pinject.provides('session_data', in_scope=THREAD_LOCAL)
146
def provide_session_data(self):
147
return SessionData()
148
149
obj_graph = pinject.new_object_graph(
150
binding_specs=[ScopingBindingSpec()],
151
id_to_scope={THREAD_LOCAL: ThreadLocalScope()}
152
)
153
154
class UserService(object):
155
def __init__(self, session_data):
156
self.session = session_data
157
158
# Same thread gets same session, different threads get different sessions
159
user_service = obj_graph.provide(UserService)
160
print(f"Session for thread {user_service.session.thread_id}")
161
```
162
163
### Custom Scope Implementation
164
165
```python
166
import pinject
167
import weakref
168
169
class WeakReferenceScope(pinject.Scope):
170
"""Custom scope that holds weak references to instances."""
171
172
def __init__(self):
173
self._instances = {}
174
175
def provide(self, binding_key, default_provider_fn):
176
# Check if we have a live weak reference
177
if binding_key in self._instances:
178
instance = self._instances[binding_key]()
179
if instance is not None:
180
return instance
181
182
# Create new instance and store weak reference
183
instance = default_provider_fn()
184
self._instances[binding_key] = weakref.ref(instance)
185
return instance
186
187
class CacheService(object):
188
def __init__(self):
189
self.data = {}
190
print("CacheService created")
191
192
WEAK_REF_SCOPE = "weak_reference"
193
194
class CacheBindingSpec(pinject.BindingSpec):
195
def configure(self, bind):
196
bind('cache_service').to_class(
197
CacheService,
198
in_scope=WEAK_REF_SCOPE
199
)
200
201
obj_graph = pinject.new_object_graph(
202
binding_specs=[CacheBindingSpec()],
203
id_to_scope={WEAK_REF_SCOPE: WeakReferenceScope()}
204
)
205
206
# Instance is created and weakly cached
207
cache1 = obj_graph.provide(CacheService) # "CacheService created"
208
cache2 = obj_graph.provide(CacheService) # Same instance, no creation message
209
210
print(cache1 is cache2) # True
211
212
# If we delete references and force garbage collection,
213
# next request creates a new instance
214
del cache1, cache2
215
import gc
216
gc.collect()
217
218
cache3 = obj_graph.provide(CacheService) # "CacheService created" (new instance)
219
```
220
221
### Scope Dependencies and Validation
222
223
```python
224
import pinject
225
226
class DatabaseConnection(object):
227
def __init__(self):
228
self.connection_id = id(self)
229
230
class RequestContext(object):
231
def __init__(self, database_connection):
232
self.db = database_connection
233
234
REQUEST_SCOPE = "request"
235
APPLICATION_SCOPE = "application"
236
237
class RequestScope(pinject.Scope):
238
def provide(self, binding_key, default_provider_fn):
239
# Always create new instance for request scope
240
return default_provider_fn()
241
242
class MyBindingSpec(pinject.BindingSpec):
243
def configure(self, bind):
244
bind('database_connection').to_class(
245
DatabaseConnection,
246
in_scope=APPLICATION_SCOPE
247
)
248
bind('request_context').to_class(
249
RequestContext,
250
in_scope=REQUEST_SCOPE
251
)
252
253
def validate_scope_usage(from_scope, to_scope):
254
"""Validate that request scope can use application scope."""
255
if from_scope == REQUEST_SCOPE and to_scope == APPLICATION_SCOPE:
256
return True
257
return from_scope == to_scope
258
259
obj_graph = pinject.new_object_graph(
260
binding_specs=[MyBindingSpec()],
261
id_to_scope={
262
APPLICATION_SCOPE: pinject.Scope(), # Singleton-like
263
REQUEST_SCOPE: RequestScope()
264
},
265
is_scope_usable_from_scope=validate_scope_usage
266
)
267
268
# Request context gets new instance, but shares database connection
269
context1 = obj_graph.provide(RequestContext)
270
context2 = obj_graph.provide(RequestContext)
271
272
print(context1 is context2) # False (different request contexts)
273
print(context1.db is context2.db) # True (shared database connection)
274
```