Coverage for src/lektor_ng/admin/utils.py: 100%

1from collections.abc import Callable, Iterable, Iterator 

2from functools import update_wrapper 

3from itertools import chain 

4from typing import Any 

5 

6from flask import Response, json 

7 

8 

9def eventstream(f: Callable[..., Iterable[Any]]) -> Callable[..., Response]: 

10 def new_func(*args: Any, **kwargs: Any) -> Response: 

11 def generate() -> Iterator[bytes]: 

12 for event in chain(f(*args, **kwargs), (None,)): 

13 yield f"data: {json.dumps(event)}\n\n".encode() 

14 

15 return Response(generate(), mimetype="text/event-stream", direct_passthrough=True) 

16 

17 return update_wrapper(new_func, f)