mypy

pigwig/exceptions.py

@@ -6,7 +6,7 @@ class HTTPException(Exception):
 	:param body: unlike in :class:`.Response`, must be a ``str``
 	'''
 
-	def __init__(self, code, body):
+	def __init__(self, code: int, body: str) -> None:
 		super().__init__(code, body)
 		self.code = code
 		self.body = body

pigwig/inotify.py

@@ -1,6 +1,6 @@
 from collections import namedtuple
-import ctypes
-import ctypes.util
+import ctypes # type:ignore
+import ctypes.util # type:ignore
 from enum import IntEnum
 import errno
 import os

pigwig/pigwig.py

@@ -7,23 +7,28 @@
 import sys
 import textwrap
 import traceback
+from typing import Any, BinaryIO, Callable, Dict, Iterable, List, TextIO, Tuple, Union
 import urllib.parse
-import wsgiref.simple_server
+import wsgiref.simple_server # type: ignore
 
 from . import exceptions, multipart
 from .request_response import HTTPHeaders, Request, Response
 from .routes import build_route_tree
 from .templates_jinja import JinjaTemplateEngine
 
-def default_http_exception_handler(e, errors, request, app):
+def default_http_exception_handler(e: exceptions.HTTPException, errors: TextIO, request: Request,
+		app: 'PigWig') -> Response:
 	errors.write(textwrap.indent(e.body, '\t') + '\n')
 	return Response(e.body.encode('utf-8', 'replace'), e.code)
 
-def default_exception_handler(e, errors, request, app):
+def default_exception_handler(e: Exception, errors: TextIO, request: Request, app: 'PigWig') -> Response:
 	tb = traceback.format_exc()
 	errors.write(tb)
 	return Response(tb.encode('utf-8', 'replace'), 500)
 
+HTTPExceptionHandler = Callable[[exceptions.HTTPException, TextIO, Request, 'PigWig'], Response]
+ExceptionHandler = Callable[[Exception, TextIO, Request, 'PigWig'], Response]
+
 class PigWig:
 	'''
 		main WSGI entrypoint. this is a class but defines a :func:`.__call__` so instances of it can
@@ -82,9 +87,10 @@ class PigWig:
 		* ``exception_handler``
 	'''
 
-	def __init__(self, routes, template_dir=None, template_engine=JinjaTemplateEngine,
-			cookie_secret=None, http_exception_handler=default_http_exception_handler,
-			exception_handler=default_exception_handler, response_done_handler=None):
+	def __init__(self, routes, template_dir: str=None,
+			template_engine: type=JinjaTemplateEngine, cookie_secret: bytes=None,
+			http_exception_handler: HTTPExceptionHandler=default_http_exception_handler,
+			exception_handler: ExceptionHandler=default_exception_handler, response_done_handler=None) -> None:
 		if callable(routes):
 			routes = routes()
 		self.routes = build_route_tree(routes)
@@ -99,7 +105,7 @@ def __init__(self, routes, template_dir=None, template_engine=JinjaTemplateEngin
 		self.exception_handler = exception_handler
 		self.response_done_handler = response_done_handler
 
-	def __call__(self, environ, start_response):
+	def __call__(self, environ: Dict, start_response: Callable) -> Iterable[bytes]:
 		''' main WSGI entrypoint '''
 		errors = environ.get('wsgi.errors', sys.stderr)
 		try:
@@ -139,14 +145,15 @@ def __call__(self, environ, start_response):
 			start_response('500 Internal Server Error', [])
 			return [b'internal server error']
 
-	def build_request(self, environ):
+	def build_request(self, environ: Dict) -> Tuple[Request, Exception]:
 		''' builds :class:`.Response` objects. for internal use. '''
 		method = environ['REQUEST_METHOD']
 		path = environ['PATH_INFO']
-		query = {}
-		headers = HTTPHeaders()
-		cookies = http.cookies.SimpleCookie()
-		body = err = None
+		query = {} # type: Dict[str, Union[List[str], str]]
+		headers = HTTPHeaders() # type: ignore
+		cookies = http.cookies.SimpleCookie() # type: ignore
+		body = None # type: Union[Tuple[Any, int], Dict]
+		err = None
 
 		try:
 			qs = environ.get('QUERY_STRING')
@@ -161,7 +168,7 @@ def build_request(self, environ):
 			content_type = environ.get('CONTENT_TYPE')
 			if content_type:
 				headers['Content-Type'] = content_type
-				media_type, params = cgi.parse_header(content_type)
+				media_type, params = cgi.parse_header(content_type) # type: ignore
 				handler = self.content_handlers.get(media_type)
 				if handler:
 					body = handler(environ['wsgi.input'], content_length, params)
@@ -212,12 +219,12 @@ def main(self, host='0.0.0.0', port=None):
 		server.serve_forever()
 
 	@staticmethod
-	def handle_urlencoded(body, length, params):
+	def handle_urlencoded(body: BinaryIO, length: int, params: Dict[str, str]) -> Dict[str, Union[str, List[str]]]:
 		charset = params.get('charset', 'utf-8')
 		return parse_qs(body.read(length).decode(charset))
 
 	@staticmethod
-	def handle_json(body, length, params):
+	def handle_json(body: BinaryIO, length: int, params: Dict[str, str]) -> Any:
 		charset = params.get('charset', 'utf-8')
 		return json.loads(body.read(length).decode(charset))
 
@@ -230,17 +237,20 @@ def handle_multipart(body, length, params):
 				form[k] = v[0]
 		return form
 
+	content_handlers = None # type: Dict[str, Callable[[BinaryIO, int, Dict[str, str]], Any]]
+
 PigWig.content_handlers = {
 	'application/json': PigWig.handle_json,
 	'application/x-www-form-urlencoded': PigWig.handle_urlencoded,
 	'multipart/form-data': PigWig.handle_multipart,
 }
 
-def parse_qs(qs):
+def parse_qs(qs: str) -> Dict[str, Union[str, List[str]]]:
 	if not qs:
 		return {}
 	try:
-		parsed = urllib.parse.parse_qs(qs, keep_blank_values=True, strict_parsing=True, errors='strict')
+		parsed = urllib.parse.parse_qs(qs, keep_blank_values=True, strict_parsing=True,
+				errors='strict') # type: Dict[str, Any]
 	except UnicodeDecodeError as e:
 		qs_trunc = qs
 		if len(qs_trunc) > 24:

pigwig/reloader_osx.py

@@ -1,7 +1,7 @@
 import os
 import sys
 
-from fsevents import Observer, Stream
+from fsevents import Observer, Stream # type: ignore
 
 observer = Observer()
 observer.daemon = True

pigwig/request_response.py

@@ -1,10 +1,12 @@
 from collections import UserDict
 import copy
+import datetime
 import hashlib
 import hmac
 import http.cookies
 import json as jsonlib
 import time
+from typing import Any, Dict, Generator, List, Tuple, Union
 
 from . import exceptions
 
@@ -25,7 +27,8 @@ class Request:
 	  handed down from the server
 	'''
 
-	def __init__(self, app, method, path, query, headers, body, cookies, wsgi_environ):
+	def __init__(self, app, method: str, path: str, query: Dict[str, Union[str, List[str]]],
+			headers: 'HTTPHeaders', body, cookies, wsgi_environ: Dict[str, Any]) -> None:
 		self.app = app
 		self.method = method
 		self.path = path
@@ -35,7 +38,7 @@ def __init__(self, app, method, path, query, headers, body, cookies, wsgi_enviro
 		self.cookies = cookies
 		self.wsgi_environ = wsgi_environ
 
-	def get_secure_cookie(self, key, max_time):
+	def get_secure_cookie(self, key: str, max_time: datetime.timedelta) -> str:
 		'''
 		decode and verify a cookie set with :func:`Response.set_secure_cookie`
 
@@ -93,10 +96,11 @@ class Response:
 		('Access-Control-Allow-Headers', 'Authorization, X-Requested-With, X-Request'),
 	]
 
-	json_encoder = jsonlib.JSONEncoder(indent='\t')
-	simple_cookie = http.cookies.SimpleCookie()
+	json_encoder = jsonlib.JSONEncoder(indent='\t') # type: ignore
+	simple_cookie = http.cookies.SimpleCookie() # type: ignore
 
-	def __init__(self, body=None, code=200, content_type='text/plain', location=None, extra_headers=None):
+	def __init__(self, body=None, code: int=200, content_type: str='text/plain',
+			location: str=None, extra_headers: List[Tuple[str, str]]=None) -> None:
 		self.body = body
 		self.code = code
 
@@ -108,7 +112,9 @@ def __init__(self, body=None, code=200, content_type='text/plain', location=None
 			headers.extend(extra_headers)
 		self.headers = headers
 
-	def set_cookie(self, key, value, domain=None, path='/', expires=None, max_age=None, secure=False, http_only=False):
+	def set_cookie(self, key: str, value: Any, domain: str=None, path: str='/',
+			expires: datetime.datetime=None, max_age: datetime.timedelta=None, secure: bool=False,
+			http_only: bool=False) -> None:
 		'''
 		adds a Set-Cookie header
 
@@ -127,8 +133,7 @@ def set_cookie(self, key, value, domain=None, path='/', expires=None, max_age=No
 		if path:
 			cookie += '; Path=%s' % path
 		if expires:
-			expires = expires.strftime('%a, %d %b %Y %H:%M:%S GMT')
-			cookie += '; Expires=%s' % expires
+			cookie += '; Expires=%s' % expires.strftime('%a, %d %b %Y %H:%M:%S GMT')
 		if max_age is not None:
 			cookie += '; Max-Age=%d' % max_age.total_seconds()
 		if secure:
@@ -137,7 +142,7 @@ def set_cookie(self, key, value, domain=None, path='/', expires=None, max_age=No
 			cookie += '; HttpOnly'
 		self.headers.append(('Set-Cookie', cookie))
 
-	def set_secure_cookie(self, request, key, value, **kwargs):
+	def set_secure_cookie(self, request: Request, key: str, value: Any, **kwargs) -> None:
 		'''
 		this function accepts the same keyword arguments as :func:`.set_cookie` but stores a
 		timestamp and a signature based on ``request.app.cookie_secret``. decode with
@@ -156,7 +161,7 @@ def set_secure_cookie(self, request, key, value, **kwargs):
 		self.set_cookie(key, value_signed, **kwargs)
 
 	@classmethod
-	def json(cls, obj):
+	def json(cls, obj: Any) -> 'Response':
 		'''
 		generate a streaming :class:`.Response` object from an object with an ``application/json``
 		content type. the default :attr:`.json_encoder` indents with tabs - override if you want
@@ -166,7 +171,7 @@ def json(cls, obj):
 		return Response(body, content_type='application/json; charset=utf-8')
 
 	@classmethod
-	def _gen_json(cls, obj):
+	def _gen_json(cls, obj: Any) -> Generator[bytes, None, None]:
 		'''
 		internal use generator for converting
 		`json.JSONEncoder.iterencode <https://docs.python.org/3/library/json.html#json.JSONEncoder.iterencode>`_
@@ -176,7 +181,7 @@ def _gen_json(cls, obj):
 			yield chunk.encode('utf-8')
 
 	@classmethod
-	def render(cls, request, template, context):
+	def render(cls, request: Request, template: str, context: Dict[str, Any]) -> 'Response':
 		'''
 		generate a streaming :class:`.Response` object from a template and a context with a
 		``text/html`` content type.
@@ -192,7 +197,7 @@ def render(cls, request, template, context):
 		response = cls(body, content_type='text/html; charset=utf-8')
 		return response
 
-def _hash(value_ts, cookie_secret):
+def _hash(value_ts: str, cookie_secret: bytes) -> str:
 	h = hmac.new(cookie_secret, value_ts.encode(), hashlib.sha256)
 	signature = h.hexdigest()
 	return signature

pigwig/routes.py

@@ -1,16 +1,19 @@
 import textwrap
+from typing import Callable, Dict, List, Tuple
 import re
 
 from . import exceptions
 
 class RouteNode:
-	def __init__(self):
-		self.method_handlers = {}
-		self.static_children = {}
-		self.param_name = self.param_children = self.param_is_path = None
+	def __init__(self) -> None:
+		self.method_handlers = {} # type: Dict[str, Callable]
+		self.static_children = {} # type: Dict[str, RouteNode]
+		self.param_name = None # type: str
+		self.param_children = None # type: RouteNode
+		self.param_is_path = None # type: bool
 
 	param_re = re.compile(r'<([\w:]+)>')
-	def assign_route(self, path_elements, method, handler):
+	def assign_route(self, path_elements: List[str], method: str, handler: Callable) -> None:
 		if not path_elements or path_elements[0] == '':
 			if len(path_elements) > 1:
 				raise Exception('cannot have consecutive / in routes')
@@ -42,7 +45,7 @@ def assign_route(self, path_elements, method, handler):
 
 		child.assign_route(remaining, method, handler)
 
-	def get_route(self, method, path_elements, params):
+	def get_route(self, method: str, path_elements: List[str], params: Dict[str, str]) -> Tuple[Callable, Dict]:
 		if not path_elements or path_elements[0] == '':
 			handler = self.method_handlers.get(method)
 			if handler is not None:
@@ -66,11 +69,11 @@ def get_route(self, method, path_elements, params):
 			child = self.param_children
 		return child.get_route(method, remaining, params)
 
-	def route(self, method, path):
+	def route(self, method: str, path: str) -> Tuple[Callable, Dict]:
 		path_elements = path[1:].split('/')
 		return self.get_route(method, path_elements, {})
 
-	def __str__(self):
+	def __str__(self) -> str:
 		rval = []
 		for method, handler in self.method_handlers.items():
 			rval.append('%s: %s,' % (method, handler))
@@ -83,7 +86,7 @@ def __str__(self):
 			rval.append('%s: %s' % (name, self.param_children))
 		return '{\n%s\n}' % textwrap.indent('\n'.join(rval), '\t')
 
-def build_route_tree(routes):
+def build_route_tree(routes: List[Tuple[str, str, Callable]]) -> RouteNode:
 	root_node = RouteNode()
 	for method, path, handler in routes:
 		path_elements = path[1:].split('/')

pigwig/templates_jinja.py

@@ -1,5 +1,5 @@
 try:
-	import jinja2
+	import jinja2 # type: ignore
 except ImportError:
 	jinja2 = None
 

pigwig/tests/test_pigwig.py

@@ -2,7 +2,7 @@
 import math
 import textwrap
 import unittest
-from unittest import mock
+from unittest import mock # type: ignore
 
 from pigwig import PigWig
 from pigwig.exceptions import HTTPException