summaryrefslogtreecommitdiff
path: root/wsgitools/applications.py
diff options
context:
space:
mode:
Diffstat (limited to 'wsgitools/applications.py')
-rw-r--r--wsgitools/applications.py44
1 files changed, 29 insertions, 15 deletions
diff --git a/wsgitools/applications.py b/wsgitools/applications.py
index 9894cf8..f51fccf 100644
--- a/wsgitools/applications.py
+++ b/wsgitools/applications.py
@@ -1,4 +1,7 @@
import os.path
+import typing
+
+from wsgitools.internal import Environ, HeaderList, StartResponse
__all__ = []
@@ -9,17 +12,21 @@ class StaticContent:
receives with method GET or HEAD (content stripped). If not present, a
content-length header is computed.
"""
- def __init__(self, status, headers, content, anymethod=False):
+ content: typing.Iterable[bytes]
+
+ def __init__(
+ self,
+ status: str,
+ headers: HeaderList,
+ content: typing.Union[bytes, typing.Iterable[bytes]],
+ anymethod: bool = False,
+ ):
"""
- @type status: str
@param status: is the HTTP status returned to the browser (ex: "200 OK")
- @type headers: list
@param headers: is a list of C{(header, value)} pairs being delivered as
HTTP headers
- @type content: bytes
@param content: contains the data to be delivered to the client. It is
either a string or some kind of iterable yielding strings.
- @type anymethod: boolean
@param anymethod: determines whether any request method should be
answered with this response instead of a 501
"""
@@ -40,7 +47,9 @@ class StaticContent:
if length >= 0:
if not [v for h, v in headers if h.lower() == "content-length"]:
headers.append(("Content-length", str(length)))
- def __call__(self, environ, start_response):
+ def __call__(
+ self, environ: Environ, start_response: StartResponse
+ ) -> typing.Iterable[bytes]:
"""wsgi interface"""
assert isinstance(environ, dict)
if environ["REQUEST_METHOD"].upper() not in ["GET", "HEAD"] and \
@@ -61,20 +70,21 @@ class StaticFile:
request it receives with method GET or HEAD (content stripped). If not
present, a content-length header is computed.
"""
- def __init__(self, filelike, status="200 OK", headers=list(),
- blocksize=4096):
+ def __init__(
+ self,
+ filelike: typing.Union[str, typing.BinaryIO],
+ status: str = "200 OK",
+ headers: HeaderList = list(),
+ blocksize: int = 4096,
+ ):
"""
- @type status: str
@param status: is the HTTP status returned to the browser
- @type headers: [(str, str)]
@param headers: is a list of C{(header, value)} pairs being delivered as
HTTP headers
- @type filelike: str or file-like
@param filelike: may either be an path in the local file system or a
file-like that must support C{read(size)} and C{seek(0)}. If
C{tell()} is present, C{seek(0, 2)} and C{tell()} will be used
to compute the content-length.
- @type blocksize: int
@param blocksize: the content is provided in chunks of this size
"""
self.filelike = filelike
@@ -82,7 +92,9 @@ class StaticFile:
self.headers = headers
self.blocksize = blocksize
- def _serve_in_chunks(self, stream):
+ def _serve_in_chunks(
+ self, stream: typing.BinaryIO
+ ) -> typing.Iterator[bytes]:
"""internal method yielding data from the given stream"""
while True:
data = stream.read(self.blocksize)
@@ -92,7 +104,9 @@ class StaticFile:
if isinstance(self.filelike, str):
stream.close()
- def __call__(self, environ, start_response):
+ def __call__(
+ self, environ: Environ, start_response: StartResponse
+ ) -> typing.Iterable[bytes]:
"""wsgi interface"""
assert isinstance(environ, dict)
@@ -102,7 +116,7 @@ class StaticFile:
[("Content-length", str(len(resp)))])
return [resp]
- stream = None
+ stream: typing.Optional[typing.BinaryIO] = None
size = -1
try:
if isinstance(self.filelike, str):