pigwig

pigwig

PigWig(routes: RouteDefinition | Callable[[], RouteDefinition], template_dir: str | None = None, template_engine: type = JinjaTemplateEngine, cookie_secret: bytes | None = None, http_exception_handler: HTTPExceptionHandler = default_http_exception_handler, exception_handler: ExceptionHandler = default_exception_handler, response_done_handler: Callable[[Request, Response], Any] | None = None)

main WSGI entrypoint. this is a class but defines a __call__ so instances of it can be passed directly to WSGI servers.

Parameters:

• routes (list | function) –

  a list of 3-tuples: (method, path, handler) or a function that returns such a list * method is the HTTP method/verb (GET, POST, etc.) * path can either be a static path (/foo/bar) or have params (/post/<id>). params can be prefixed with path: to eat up the rest of the path (/tree/<path:subdir> matches /tree/a/b/c). params are passed to the handler as keyword arguments. params cannot be optional, but you can map two routes to a handler that takes an optional argument. params must make up the entire path segment - you cannot have /post_<id>. * handler is a function taking a Request positional argument and any number of param keyword arguments having two identical static routes or two overlapping param segments (/foo/<bar> and /foo/<baz>) with the same method raises an exceptions.RouteConflict

• template_dir (str, default: None ) –

  if specified, a template_engine is created with this as the argument. for pigwig.templates_jinja.JinjaTemplateEngine, this should be an absolute path or it will be relative to the current working directory.

• template_engine (type, default: JinjaTemplateEngine ) –

  a class that takes a template_dir in the constructor and has a .stream method that takes template_name, context as arguments (passed from user code - for jinja2, context is a dictionary)

• cookie_secret (str, default: None ) –

  app-wide secret used for signing secure cookies. see Request.get_secure_cookie

• http_exception_handler (HTTPExceptionHandler, default: default_http_exception_handler ) –

  a function that will be called when an exceptions.HTTPException is raised. it will be passed the original exception, wsgi.errors, the Request, and a reference to this PigWig instance. it must return a Response and should almost certainly have the code of the original exception. exceptions raised here can be handled by exception_handler.

• exception_handler (ExceptionHandler, default: default_exception_handler ) –

  a function that will be called when any other exception is raised. it will be passed the same arguments as http_exception_handler and must also return a Response. be careful: raising an exception here is bad.

• response_done_handler (Callable[[Request, Response], Any] | None, default: None ) –

  a function that will be called when control has been returned back to the WSGI server. it will be passed a request and response. be careful: raising an exception here is very bad.

  has the following instance attrs:

  • routes - an internal representation of the route tree - not the list passed to the constructor
  • template_engine
  • cookie_secret
  • http_exception_handler
  • exception_handler

Source code in pigwig/pigwig.py

def __init__(self, routes: RouteDefinition | Callable[[], RouteDefinition], template_dir: str | None=None,
		template_engine: type=JinjaTemplateEngine, cookie_secret: bytes | None=None,
		http_exception_handler: HTTPExceptionHandler=default_http_exception_handler,
		exception_handler: ExceptionHandler=default_exception_handler,
		response_done_handler: Callable[[Request, Response], Any] | None=None) -> None:
	if callable(routes):
		routes = routes()
	self.routes = build_route_tree(routes)

	if template_dir:
		self.template_engine = template_engine(template_dir)
	else:
		self.template_engine = None

	self.cookie_secret = cookie_secret
	self.http_exception_handler = http_exception_handler
	self.exception_handler = exception_handler
	self.response_done_handler = response_done_handler

content_handlers: dict[str, Callable[[io.BufferedIOBase, int | None, dict[str, str]], Any]] instance-attribute

map of content types to body parsers

__call__(environ: dict, start_response: Callable) -> Iterable[bytes]

main WSGI entrypoint

Source code in pigwig/pigwig.py

def __call__(self, environ: dict, start_response: Callable) -> Iterable[bytes]:
	""" main WSGI entrypoint """
	errors = cast(TextIO, environ.get('wsgi.errors', sys.stderr))
	try:
		if environ['REQUEST_METHOD'] == 'OPTIONS':
			start_response('200 OK', copy.copy(Response.DEFAULT_HEADERS))
			return []

		request, err = self.build_request(environ)
		try:
			try:
				if err:
					raise err

				handler, kwargs = self.routes.route(request.method, request.path)
				response = handler(request, **kwargs)
			except exceptions.HTTPException as e:
				response = self.http_exception_handler(e, errors, request, self)
		except Exception as e: # something went wrong in handler or http_exception_handler
			response = self.exception_handler(e, errors, request, self)

		if isinstance(response.body, str):
			response.body = [response.body.encode('utf-8')]
		elif isinstance(response.body, bytes):
			response.body = [response.body]
		elif response.body is None:
			response.body = []
		elif not isgenerator(response.body):
			raise Exception('unhandled view response type: %s' % type(response.body))

		status_line = '%d %s' % (response.code, http.client.responses[response.code])
		start_response(status_line, response.headers)
		if self.response_done_handler:
			self.response_done_handler(request, response)
		return response.body
	except Exception: # something went very wrong handling OPTIONS, in error handling, or in sending the response
		errors.write(traceback.format_exc())
		start_response('500 Internal Server Error', [])
		return [b'internal server error']

build_request(environ: dict) -> tuple[Request, Exception | None]

builds Response objects. for internal use.

Source code in pigwig/pigwig.py

def build_request(self, environ: dict) -> tuple[Request, Exception | None]:
	""" builds [Response][] objects. for internal use. """
	method = environ['REQUEST_METHOD']
	path = environ['PATH_INFO'].encode('latin-1').decode('utf-8') # https://github.com/python/cpython/issues/60883
	query: Mapping[str, list[str] | str] = {}
	headers = HTTPHeaders()
	cookies = http.cookies.SimpleCookie()
	body = {}
	err = None

	try:
		qs = environ.get('QUERY_STRING')
		if qs:
			query = parse_qs(qs)

		content_length_str: str | None = environ.get('CONTENT_LENGTH')
		if content_length_str:
			headers['Content-Length'] = content_length_str
			content_length = int(content_length_str)
		else:
			content_length = None
		content_type = environ.get('CONTENT_TYPE')
		if content_type:
			headers['Content-Type'] = content_type
			media_type, params = multipart.parse_header(content_type)
			handler = self.content_handlers.get(media_type)
			if handler:
				body = handler(environ['wsgi.input'], content_length, params)

		http_cookie = environ.get('HTTP_COOKIE')
		if http_cookie:
			cookies.load(http_cookie)

		for key, val in environ.items():
			if key.startswith('HTTP_'):
				headers[key[5:].replace('_', '-')] = val
	except Exception as e:
		err = e

	return Request(self, method, path, query, headers, body, cookies, environ), err

main(host: str = '0.0.0.0', port: int | None = None) -> None

sets up the autoreloader and runs a wsgiref.simple_server. useful for development.

Source code in pigwig/pigwig.py

def main(self, host: str='0.0.0.0', port: int | None=None) -> None:
	"""
	sets up the autoreloader and runs a
	[wsgiref.simple_server](https://docs.python.org/3/library/wsgiref.html#module-wsgiref.simple_server).
	useful for development.
	"""

	have_reloader = True
	if sys.platform == 'linux':
		from . import reloader_linux as reloader
	elif sys.platform == 'darwin':
		try:
			from . import reloader_osx as reloader
		except ImportError:
			have_reloader = False
			print('install MacFSEvents for auto-reloading')
	else:
		have_reloader = False
		print('no reloader available for', sys.platform)
	if have_reloader:
		reloader.init()

	if hasattr(self.template_engine, 'jinja_env'):
		self.template_engine.jinja_env.auto_reload = True

	if port is None:
		port = 8000
		if len(sys.argv) == 2:
			port = int(sys.argv[1])
	server = wsgiref.simple_server.make_server(host, port, self)
	print('listening on', port)
	server.serve_forever()

default_http_exception_handler(e: exceptions.HTTPException, errors: TextIO, request: Request, app: 'PigWig') -> Response

Source code in pigwig/pigwig.py

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)

default_exception_handler(e: Exception, errors: TextIO, request: Request, app: 'PigWig') -> Response

Source code in pigwig/pigwig.py

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)