0
# Abstract Classes
1
2
Core abstract classes and proxy objects for creating thenable objects. These provide the foundation for vine's promise system and enable custom implementations of promise-like behavior.
3
4
## Capabilities
5
6
### Thenable Abstract Base Class
7
8
Abstract base class defining the interface for all promise-like objects in vine.
9
10
```python { .api }
11
class Thenable:
12
def then(self, on_success, on_error=None):
13
"""
14
Chain callback to execute when thenable is fulfilled.
15
16
Parameters:
17
- on_success: callable, function to execute on successful completion
18
- on_error: callable, optional error handler for exceptions
19
20
Returns:
21
Implementation-specific return value (typically a thenable)
22
"""
23
24
def throw(self, exc=None, tb=None, propagate=True):
25
"""
26
Throw exception through the thenable chain.
27
28
Parameters:
29
- exc: Exception, exception to throw (uses current if None)
30
- tb: traceback, traceback object for exception
31
- propagate: bool, whether to re-raise if no error handler
32
"""
33
34
def cancel(self):
35
"""
36
Cancel the thenable operation.
37
38
Stops execution and cleans up resources.
39
"""
40
41
@classmethod
42
def register(cls, other):
43
"""
44
Register a class as implementing Thenable interface.
45
46
Parameters:
47
- other: class to register as Thenable
48
49
Returns:
50
The registered class (enables use as decorator)
51
"""
52
53
@classmethod
54
def __subclasshook__(cls, C):
55
"""
56
Check if class C supports thenable interface.
57
58
Returns True if class has 'then' method in its MRO.
59
"""
60
```
61
62
**Usage Examples:**
63
64
```python
65
from vine import Thenable, promise
66
67
# Check if object is thenable
68
p = promise()
69
print(isinstance(p, Thenable)) # True
70
71
# Register custom class as thenable
72
@Thenable.register
73
class CustomThenable:
74
def then(self, on_success, on_error=None):
75
# Custom implementation
76
pass
77
78
def throw(self, exc=None, tb=None, propagate=True):
79
# Custom implementation
80
pass
81
82
def cancel(self):
83
# Custom implementation
84
pass
85
86
# Now CustomThenable instances are considered Thenable
87
custom = CustomThenable()
88
print(isinstance(custom, Thenable)) # True
89
```
90
91
### Thenable Proxy
92
93
Proxy class that delegates thenable operations to a target promise.
94
95
```python { .api }
96
class ThenableProxy:
97
def _set_promise_target(self, p):
98
"""
99
Set the target promise to proxy operations to.
100
101
Parameters:
102
- p: promise object to use as proxy target
103
"""
104
105
def then(self, on_success, on_error=None):
106
"""
107
Delegate then() call to target promise.
108
109
Parameters:
110
- on_success: callable, success callback
111
- on_error: callable, error callback
112
113
Returns:
114
Result of target promise's then() method
115
"""
116
117
def cancel(self):
118
"""
119
Delegate cancel() call to target promise.
120
121
Returns:
122
Result of target promise's cancel() method
123
"""
124
125
def throw(self, exc=None, tb=None, propagate=True):
126
"""
127
Delegate throw() call to target promise.
128
129
Parameters:
130
- exc: Exception to throw
131
- tb: traceback object
132
- propagate: whether to re-raise
133
134
Returns:
135
Result of target promise's throw() method
136
"""
137
138
def throw1(self, exc=None):
139
"""
140
Delegate throw1() call to target promise.
141
142
Parameters:
143
- exc: Exception to throw
144
145
Returns:
146
Result of target promise's throw1() method
147
"""
148
149
@property
150
def cancelled(self) -> bool:
151
"""Proxy to target promise's cancelled property."""
152
153
@property
154
def ready(self) -> bool:
155
"""Proxy to target promise's ready property."""
156
157
@property
158
def failed(self) -> bool:
159
"""Proxy to target promise's failed property."""
160
```
161
162
**Usage Examples:**
163
164
```python
165
from vine import promise
166
from vine.abstract import ThenableProxy
167
168
# Create proxy for delayed promise creation
169
class DelayedPromiseProxy(ThenableProxy):
170
def __init__(self, promise_factory):
171
self.promise_factory = promise_factory
172
self._p = None
173
174
def _ensure_target(self):
175
if self._p is None:
176
self._p = self.promise_factory()
177
return self._p
178
179
def then(self, on_success, on_error=None):
180
target = self._ensure_target()
181
return target.then(on_success, on_error)
182
183
# Usage
184
def create_expensive_promise():
185
print("Creating expensive promise...")
186
return promise(lambda: expensive_computation())
187
188
# Proxy delays creation until actually needed
189
proxy = DelayedPromiseProxy(create_expensive_promise)
190
191
# Promise only created when then() is called
192
proxy.then(lambda result: print(f"Got: {result}"))
193
```
194
195
## Advanced Usage
196
197
### Custom Thenable Implementation
198
199
```python
200
from vine import Thenable
201
import asyncio
202
203
@Thenable.register
204
class AsyncThenable:
205
"""Thenable that wraps asyncio coroutines."""
206
207
def __init__(self, coro):
208
self.coro = coro
209
self.callbacks = []
210
self.error_callbacks = []
211
self._result = None
212
self._error = None
213
self._done = False
214
215
def then(self, on_success, on_error=None):
216
if self._done:
217
if self._error:
218
if on_error:
219
on_error(self._error)
220
else:
221
on_success(self._result)
222
else:
223
self.callbacks.append(on_success)
224
if on_error:
225
self.error_callbacks.append(on_error)
226
return self
227
228
def throw(self, exc=None, tb=None, propagate=True):
229
self._error = exc or Exception("Thenable error")
230
self._done = True
231
for callback in self.error_callbacks:
232
callback(self._error)
233
if propagate and not self.error_callbacks:
234
raise self._error
235
236
def cancel(self):
237
if hasattr(self.coro, 'cancel'):
238
self.coro.cancel()
239
self._done = True
240
241
async def run(self):
242
"""Execute the wrapped coroutine."""
243
try:
244
self._result = await self.coro
245
self._done = True
246
for callback in self.callbacks:
247
callback(self._result)
248
except Exception as e:
249
self.throw(e)
250
251
# Usage
252
async def async_operation():
253
await asyncio.sleep(1)
254
return "Async result"
255
256
# Wrap async operation in custom thenable
257
async_thenable = AsyncThenable(async_operation())
258
async_thenable.then(lambda result: print(f"Got: {result}"))
259
260
# Run the async operation
261
asyncio.run(async_thenable.run())
262
```
263
264
### Thenable Duck Typing
265
266
```python
267
# Vine automatically recognizes objects with then() method
268
class CustomPromiseLike:
269
def __init__(self, value):
270
self.value = value
271
self.ready = False
272
273
def then(self, callback, on_error=None):
274
if self.ready:
275
callback(self.value)
276
return self
277
278
def complete(self):
279
self.ready = True
280
281
# Duck typing - automatically recognized as Thenable
282
custom = CustomPromiseLike("test")
283
print(isinstance(custom, Thenable)) # True (due to __subclasshook__)
284
285
# Can be used with vine utilities
286
from vine import maybe_promise
287
p = maybe_promise(custom) # Returns custom object as-is
288
```
289
290
### Proxy Patterns
291
292
```python
293
class LoggingThenableProxy(ThenableProxy):
294
"""Proxy that logs all thenable operations."""
295
296
def __init__(self, target):
297
self._set_promise_target(target)
298
self.operation_log = []
299
300
def then(self, on_success, on_error=None):
301
self.operation_log.append(f"then() called with {on_success}")
302
return super().then(on_success, on_error)
303
304
def cancel(self):
305
self.operation_log.append("cancel() called")
306
return super().cancel()
307
308
def throw(self, exc=None, tb=None, propagate=True):
309
self.operation_log.append(f"throw() called with {exc}")
310
return super().throw(exc, tb, propagate)
311
312
# Usage
313
original_promise = promise(lambda: "result")
314
logged_promise = LoggingThenableProxy(original_promise)
315
316
logged_promise.then(lambda x: print(x))
317
logged_promise()
318
319
print(logged_promise.operation_log) # Shows all operations
320
```