Method: flask.helpers.stream_with_context
Calls: 8, Exceptions: 2, Paths: 2Back
Path 1: 6 calls (0.75)
TestStreaming.test_streaming_with_context.
stream_with_context.
1def stream_with_context(
2 generator_or_function: (
3 t.Iterator[t.AnyStr] | t.Callable[..., t.Iterator[t.AnyStr]]
4 )
5) -> t.Iterator[t.AnyStr]:
6 """Request contexts disappear when the response is started on the server.
7 This is done for efficiency reasons and to make it less likely to encounter
8 memory leaks with badly written WSGI middlewares. The downside is that if
9 you are using streamed responses, the generator cannot access request bound
10 information any more.
11
12 This function however can help you keep the context around for longer::
13
14 from flask import stream_with_context, request, Response
15
16 @app.route('/stream')
17 def streamed_response():
18 @stream_with_context
19 def generate():
20 yield 'Hello '
21 yield request.args['name']
22 yield '!'
23 return Response(generate())
24
25 Alternatively it can also be used around a specific generator::
26
27 from flask import stream_with_context, request, Response
28
29 @app.route('/stream')
30 def streamed_response():
31 def generate():
32 yield 'Hello '
33 yield request.args['name']
34 yield '!'
35 return Response(stream_with_context(generate()))
36
37 .. versionadded:: 0.9
38 """
39 try:
40 gen = iter(generator_or_function) # type: ignore
41 except TypeError:
42
43 def decorator(*args: t.Any, **kwargs: t.Any) -> t.Any:
44 gen = generator_or_function(*args, **kwargs) # type: ignore
45 return stream_with_context(gen)
46
47 return update_wrapper(decorator, generator_or_function) # type: ignore
48
49 def generator() -> t.Generator:
50 ctx = _cv_request.get(None)
51 if ctx is None:
52 raise RuntimeError(
53 "'stream_with_context' can only be used when a request"
54 " context is active, such as in a view function."
55 )
56 with ctx:
57 # Dummy sentinel. Has to be inside the context block or we're
58 # not actually keeping the context around.
59 yield None
60
61 # The try/finally is here so that if someone passes a WSGI level
62 # iterator in we're still running the cleanup logic. Generators
63 # don't need that because they are closed on their destruction
64 # automatically.
65 try:
66 yield from gen
67 finally:
68 if hasattr(gen, "close"):
69 gen.close()
70
71 # The trick is to start the generator. Then the code execution runs until
72 # the first dummy None is yielded at which point the context was already
73 # pushed. This item is discarded. Then when the iteration continues the
74 # real generator is executed.
75 wrapped_g = generator()
76 next(wrapped_g)
77 return wrapped_g
Path 2: 2 calls (0.25)
TestStreaming.test_streaming_with_context_as_decorator.
TestStreaming.test_streaming_with_context_as_decorator.
TypeError (2)
1def stream_with_context(
2 generator_or_function: (
3 t.Iterator[t.AnyStr] | t.Callable[..., t.Iterator[t.AnyStr]]
4 )
5) -> t.Iterator[t.AnyStr]:
6 """Request contexts disappear when the response is started on the server.
7 This is done for efficiency reasons and to make it less likely to encounter
8 memory leaks with badly written WSGI middlewares. The downside is that if
9 you are using streamed responses, the generator cannot access request bound
10 information any more.
11
12 This function however can help you keep the context around for longer::
13
14 from flask import stream_with_context, request, Response
15
16 @app.route('/stream')
17 def streamed_response():
18 @stream_with_context
19 def generate():
20 yield 'Hello '
21 yield request.args['name']
22 yield '!'
23 return Response(generate())
24
25 Alternatively it can also be used around a specific generator::
26
27 from flask import stream_with_context, request, Response
28
29 @app.route('/stream')
30 def streamed_response():
31 def generate():
32 yield 'Hello '
33 yield request.args['name']
34 yield '!'
35 return Response(stream_with_context(generate()))
36
37 .. versionadded:: 0.9
38 """
39 try:
40 gen = iter(generator_or_function) # type: ignore
41 except TypeError:
42
43 def decorator(*args: t.Any, **kwargs: t.Any) -> t.Any:
44 gen = generator_or_function(*args, **kwargs) # type: ignore
45 return stream_with_context(gen)
46
47 return update_wrapper(decorator, generator_or_function) # type: ignore
48
49 def generator() -> t.Generator:
50 ctx = _cv_request.get(None)
51 if ctx is None:
52 raise RuntimeError(
53 "'stream_with_context' can only be used when a request"
54 " context is active, such as in a view function."
55 )
56 with ctx:
57 # Dummy sentinel. Has to be inside the context block or we're
58 # not actually keeping the context around.
59 yield None
60
61 # The try/finally is here so that if someone passes a WSGI level
62 # iterator in we're still running the cleanup logic. Generators
63 # don't need that because they are closed on their destruction
64 # automatically.
65 try:
66 yield from gen
67 finally:
68 if hasattr(gen, "close"):
69 gen.close()
70
71 # The trick is to start the generator. Then the code execution runs until
72 # the first dummy None is yielded at which point the context was already
73 # pushed. This item is discarded. Then when the iteration continues the
74 # real generator is executed.
75 wrapped_g = generator()
76 next(wrapped_g)
77 return wrapped_g