MCPcopy
hub / github.com/celery/celery / FallbackContext

Class FallbackContext

celery/utils/objects.py:46–91  ·  view source on GitHub ↗

Context workaround. The built-in ``@contextmanager`` utility does not work well when wrapping other contexts, as the traceback is wrong when the wrapped context raises. This solves this problem and can be used instead of ``@contextmanager`` in this example:: @contextma

Source from the content-addressed store, hash-verified

44
45
46class FallbackContext:
47 """Context workaround.
48
49 The built-in ``@contextmanager`` utility does not work well
50 when wrapping other contexts, as the traceback is wrong when
51 the wrapped context raises.
52
53 This solves this problem and can be used instead of ``@contextmanager``
54 in this example::
55
56 @contextmanager
57 def connection_or_default_connection(connection=None):
58 if connection:
59 # user already has a connection, shouldn't close
60 # after use
61 yield connection
62 else:
63 # must've new connection, and also close the connection
64 # after the block returns
65 with create_new_connection() as connection:
66 yield connection
67
68 This wrapper can be used instead for the above like this::
69
70 def connection_or_default_connection(connection=None):
71 return FallbackContext(connection, create_new_connection)
72 """
73
74 def __init__(self, provided, fallback, *fb_args, **fb_kwargs):
75 self.provided = provided
76 self.fallback = fallback
77 self.fb_args = fb_args
78 self.fb_kwargs = fb_kwargs
79 self._context = None
80
81 def __enter__(self):
82 if self.provided is not None:
83 return self.provided
84 context = self._context = self.fallback(
85 *self.fb_args, **self.fb_kwargs
86 ).__enter__()
87 return context
88
89 def __exit__(self, *exc_info):
90 if self._context is not None:
91 return self._context.__exit__(*exc_info)
92
93
94class getitem_property:

Callers 2

connection_or_acquireMethod · 0.90
producer_or_acquireMethod · 0.90

Calls

no outgoing calls

Tested by

no test coverage detected