Permalink
Browse files

chore: pass type-checking

  • Loading branch information...
Bogdanp committed May 12, 2018
1 parent 341decb commit 6fa54c489ae34b95466d40e63f7fb7adc635ea0e
Showing with 75 additions and 23 deletions.
  1. +2 −1 .gitignore
  2. +9 −0 README.md
  3. +14 −11 headers.py
  4. +28 −0 headers.pyi
  5. +4 −0 requirements-dev.txt
  6. +8 −4 response.py
  7. +6 −6 server.py
  8. +4 −1 setup.cfg
@@ -1 +1,2 @@
__pycache__
__pycache__
.mypy_cache
@@ -6,6 +6,7 @@ application from scratch in Python.
You'll need Python 3.6+ to run any of this code. Start by reading
`server.py`.
# Tags
There is a tag for each part of the series:
@@ -21,10 +22,18 @@ There is a tag for each part of the series:
[part-3]: https://defn.io/2018/03/20/web-app-from-scratch-03/
## Type-checking
This repo uses Python 3 type annotations which can be type-checked
using [mypy]. Run `pip install mypy` and then `mypy server.py` to
type check the code.
## License
web-app-from-scratch is licensed under Apache 2.0. Please see
[LICENSE] for licensing details.
[LICENSE]: https://github.com/Bogdanp/falcon_sugar/blob/master/LICENSE
[mypy]: https://mypy.readthedocs.io
@@ -1,29 +1,32 @@
import typing
from collections import defaultdict
HeadersDict = typing.Dict[str, typing.List[str]]
HeadersGenerator = typing.Generator[typing.Tuple[str, str], None, None]
class Headers:
def __init__(self) -> None:
"""A mapping from lower-cased header names to lists of string values.
"""
def __init__(self):
self._headers = defaultdict(list)
def add(self, name: str, value: str) -> None:
def add(self, name, value):
self._headers[name.lower()].append(value)
def get_all(self, name: str) -> typing.List[str]:
def get_all(self, name):
return self._headers[name.lower()]
def get(self, name: str, default: typing.Optional[str] = None) -> typing.Optional[str]:
def get(self, name, default=None):
try:
return self.get_all(name)[-1]
except IndexError:
return default
def __iter__(self) -> HeadersGenerator:
def get_int(self, name):
try:
return int(self.get(name))
except ValueError:
return None
def __iter__(self):
for name, values in self._headers.items():
for value in values:
yield name, value
@@ -0,0 +1,28 @@
from typing import Dict, Generator, List, Optional, Tuple, overload
HeadersDict = Dict[str, List[str]]
HeadersGenerator = Generator[Tuple[str, str], None, None]
class Headers:
_headers: HeadersDict
def add(self, name: str, value: str) -> None:
...
def get_all(self, name: str) -> List[str]:
...
@overload
def get(self, name: str) -> Optional[str]:
...
@overload
def get(self, name: str, default: str) -> str:
...
def get_int(self, name: str) -> Optional[int]:
...
def __iter__(self) -> HeadersGenerator:
...
@@ -0,0 +1,4 @@
isort
flake8
flake8-quotes
mypy
@@ -18,11 +18,15 @@ class Response:
encoding: An encoding for the content, if provided.
"""
status: bytes
headers: Headers
body: typing.IO[bytes]
def __init__(
self,
status: str = "200 OK",
headers: typing.Optional[Headers] = None,
body: typing.Optional[typing.IO] = None,
body: typing.Optional[typing.IO[bytes]] = None,
content: typing.Optional[str] = None,
encoding: str = "utf-8"
) -> None:
@@ -40,7 +44,7 @@ def __init__(
def send(self, sock: socket.socket) -> None:
"""Write this response to a socket.
"""
content_length = self.headers.get("content-length")
content_length = self.headers.get_int("content-length")
if content_length is None:
try:
body_stat = os.fstat(self.body.fileno())
@@ -51,12 +55,12 @@ def send(self, sock: socket.socket) -> None:
self.body.seek(0, os.SEEK_SET)
if content_length > 0:
self.headers.add("content-length", content_length)
self.headers.add("content-length", str(content_length))
headers = b"HTTP/1.1 " + self.status + b"\r\n"
for header_name, header_value in self.headers:
headers += f"{header_name}: {header_value}\r\n".encode()
sock.sendall(headers + b"\r\n")
if content_length > 0:
sock.sendfile(self.body)
sock.sendfile(self.body) # type: ignore
@@ -3,12 +3,12 @@
import os
import socket
import typing
from queue import Empty, Queue
from threading import Thread
from typing import Callable, List, Tuple
from request import Request
from response import Response
from threading import Thread
from typing import Callable, List, Tuple
from queue import Queue, Empty
LOGGER = logging.getLogger(__name__)
@@ -77,12 +77,12 @@ def handle_client(self, client_sock: socket.socket, client_addr: typing.Tuple[st
class HTTPServer:
def __init__(self, host="127.0.0.1", port=9000, worker_count=16) -> None:
self.handlers = []
self.handlers: List[Tuple[str, HandlerT]] = []
self.host = host
self.port = port
self.worker_count = worker_count
self.worker_backlog = worker_count * 8
self.connection_queue = Queue(self.worker_backlog)
self.connection_queue: Queue = Queue(self.worker_backlog)
def mount(self, path_prefix: str, handler: HandlerT) -> None:
"""Mount a request handler at a particular path. Handler
@@ -163,6 +163,6 @@ def app(request: Request) -> Response:
server = HTTPServer()
server.mount("/static", serve_static("www")),
server.mount("/static", serve_static("www"))
server.mount("", wrap_auth(app))
server.serve_forever()
@@ -4,4 +4,7 @@ max-line-length = 120
[flake8]
ignore = E402,F403,F811
max-complexity = 20
max-line-length = 120
max-line-length = 120
inline-quotes = "
multiline-quotes = """

0 comments on commit 6fa54c4

Please sign in to comment.