Skip to content

Commit

Permalink
add type annotations to StreamingResponse (#1708)
Browse files Browse the repository at this point in the history
  • Loading branch information
adriangb committed Jun 26, 2022
1 parent 1225ae8 commit d69dc54
Showing 1 changed file with 10 additions and 2 deletions.
12 changes: 10 additions & 2 deletions starlette/responses.py
Expand Up @@ -215,10 +215,18 @@ def __init__(
self.headers["location"] = quote(str(url), safe=":/%#?=@[]!$&'()*+,;")


Content = typing.Union[str, bytes]
SyncContentStream = typing.Iterator[Content]
AsyncContentStream = typing.AsyncIterable[Content]
ContentStream = typing.Union[AsyncContentStream, SyncContentStream]


class StreamingResponse(Response):
body_iterator: AsyncContentStream

def __init__(
self,
content: typing.Any,
content: ContentStream,
status_code: int = 200,
headers: typing.Optional[typing.Mapping[str, str]] = None,
media_type: typing.Optional[str] = None,
Expand Down Expand Up @@ -257,7 +265,7 @@ async def stream_response(self, send: Send) -> None:
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
async with anyio.create_task_group() as task_group:

async def wrap(func: typing.Callable[[], typing.Coroutine]) -> None:
async def wrap(func: "typing.Callable[[], typing.Awaitable[None]]") -> None:
await func()
task_group.cancel_scope.cancel()

Expand Down

0 comments on commit d69dc54

Please sign in to comment.