I've reversed my ideas on typing.

Simon de Vlieger 2 years ago
parent 93219a5c08
commit f431ea0e83
  1. 8
  2. 18
  3. 11
  4. 4
  5. 6
  6. 20
  7. 4
  8. 2
  9. 38
  10. 31

@ -81,11 +81,3 @@ Contributing
The source code for ``gb`` lives on GitHub where you can also submit issues and
pull requests. It mostly needs help by people with the ability to test in
various clients and libraries that might still support the gopher protocol.
An often asked question is why ``gb`` does not use any of Python 3.6+'s type
annotations. The answer is quite simply that ``gb`` wants to support ``pypy`` as
well as CPython. When ``pypy`` catches up to 3.6 type annotations will be added.
``gb`` loves to run on ``pypy`` so give it a whirl!

@ -16,7 +16,7 @@ modes = {
def bail(message):
def bail(message: str) -> None:
"""Write a message to stderr and exit unsuccesfully."""
print(message, file=sys.stderr)
raise SystemExit(1)
@ -63,19 +63,27 @@ def bail(message):
"path", required=True, type=click.Path(exists=True), envvar="GB_PATH"
def main(mode, host, port, path, verbose, magic, utf8):
def main(
mode: str,
host: str,
port: int,
path: click.Path,
verbose: int,
magic: bool,
utf8: bool,
) -> int:
"""`gb` or gopherball is a modern server for the Gopher protocol."""
if port < 0 or port > 65535:
return bail("Invalid port supplied.")
bail("Invalid port supplied.")
host = ipaddress.ip_address(host)
except ValueError:
return bail("Unparseable IP supplied.")
bail("Unparseable IP supplied.")
if mode not in modes:
return bail("Invalid mode supplied.")
bail("Invalid mode supplied.")
if utf8:
encoding = "utf8"

@ -1,4 +1,7 @@
from typing import List
import gb.protocol
import gb.entry
class Document:
@ -6,13 +9,15 @@ class Document:
how those entries are joined together and if any extra information is
def __init__(self):
entries: List[gb.entry.Entry]
def __init__(self) -> None:
self.entries = []
def add_entry(self, entry):
def add_entry(self, entry: gb.entry.Entry) -> None:
def __str__(self):
def __str__(self) -> str:
return (
gb.protocol.template.format(e=e) for e in self.entries

@ -12,7 +12,7 @@ class Entry:
host = "localhost"
port = "7070"
def __init__(self, text, selector):
def __init__(self, text: str, selector: str) -> None:
self.text = text
self.selector = selector
@ -24,7 +24,7 @@ class Text(Entry):
class Directory(Entry):
code = 1
def __init__(self, text, selector):
def __init__(self, text: str, selector: str) -> None:
self.text = text + "/"
self.selector = selector

@ -1,3 +1,5 @@
from typing import Union, Type, Text
import magic
import gb.entry
@ -6,9 +8,9 @@ import gb.entry
mime_entry = {"text": gb.entry.Text}
def guess_type(path):
def guess_type(path: str) -> Union[Type[Text], str, None]:
"""Guess the type of a path and return the corresponding Entry for that
type with its data loaded."""
guess = magic.Magic(mime=True).from_file(path).split("/")[0]
return mime_entry.get(guess, gb.entry.Binary)
return mime_entry.get(guess, gb.entry.Binary) # type: ignore

@ -8,16 +8,19 @@ import gb.magic
class Mode:
def __init__(self, base_path, magic):
def __init__(self, base_path: str, magic: bool) -> None:
self.base_path = os.path.abspath(base_path)
self.magic = magic
def lookup(self, path: str) -> str:
raise NotImplementedError()
class ImplicitMode(Mode):
"""ImplicitMode looks up files recursively within a given path auto-
generating any required indexes."""
def lookup(self, path):
def lookup(self, path: str) -> str:
"""Look up a path within our base path and return the contents in
pre-rendered Gopher format!"""
@ -33,7 +36,7 @@ class ImplicitMode(Mode):
raise ValueError()
def _directory(self, path):
def _directory(self, path: str) -> str:
"""Render the files in a directory as gopher data."""
response = gb.document.Document()
@ -44,9 +47,9 @@ class ImplicitMode(Mode):
item = gb.entry.Directory
elif os.path.isfile(entry):
if self.magic:
item = gb.magic.guess_type(entry)
item = gb.magic.guess_type(entry) # type: ignore
item = gb.entry.Binary
item = gb.entry.Binary # type: ignore
item(os.path.basename(entry), entry[len(self.base_path) :])
@ -54,7 +57,7 @@ class ImplicitMode(Mode):
return str(response)
def _file(self, path):
def _file(self, path: str) -> str:
# XXX this fails when in utf8 (or in general), read bytes and then
# XXX either use surrogateescape (or just return bytes) or only do
# XXX so for non-readable files
@ -66,9 +69,6 @@ class ExplicitMode(Mode):
"""ExplicitMode has not yet been implemented but will use a json file for
explicit mapping of selectors to files and their types."""
def __init__(self, base_path, magic):
def __init__(self, base_path: str, magic: bool) -> None:
self.base_path = os.path.abspath(base_path)
self.magic = magic
def lookup(self):
raise NotImplementedError

@ -8,14 +8,14 @@ eof = "."
template = "{e.code}{e.text}\t{e.selector}\t{e.host}\t{e.port}"
def clean_selector(selector):
def clean_selector(selector: str) -> str:
"""Strip the end off of a selector and normalize an empty one."""
selector = selector.rstrip(crlf)
return selector if selector != "" else "/"
def is_valid_selector(selector):
def is_valid_selector(selector: str) -> bool:
"""Validate the selector according to gopher rules."""
# TODO implement <TAB> denoted search

@ -1,7 +1,7 @@
import os
def relativize(base, path):
def relativize(base: str, path: str) -> str:
"""Join base and path making sure that the resulting path lies
within the base. We do this by making path relative if it's an
absolute path then verifying if it's still within the base after

@ -1,5 +1,7 @@
import logging
from typing import Tuple, Any
import tornado.tcpserver
import tornado.iostream
@ -16,7 +18,13 @@ class GopherServer(tornado.tcpserver.TCPServer):
async def handle_stream(self, stream, address):
# XXX TODO has to come from somewhere!
encoding: str
mode: gb.mode.Mode
async def handle_stream(
self, stream: tornado.iostream.IOStream, address: Tuple[Any, ...]
) -> None:
"""A new incoming connection. We wait silently until the client sends
a selector through after which we clean and parse that selector."""
@ -25,48 +33,50 @@ class GopherServer(tornado.tcpserver.TCPServer):
while True:
# clean this up, split on \n but clean both
selector = await stream.read_until(b"\n") # gb.protocol.crlf)
selector = selector.decode(self.encoding)
selector_raw = await stream.read_until(
) # gb.protocol.crlf)
selector_dec = selector_raw.decode(self.encoding)
# clean this up, split on \n but clean both with \r
selector = gb.protocol.clean_selector(selector)
selector_dec = gb.protocol.clean_selector(selector_dec)
# If this is not a valid selector we immediately terminate
# the connection
if not gb.protocol.is_valid_selector(selector):
if not gb.protocol.is_valid_selector(selector_dec):
"%s requested invalid selector %r, terminating.",
await self.close_stream(stream)
log.info("%s requested %r", address, selector)
log.info("%s requested %r", address, selector_dec)
# Use our mode specific lookup to get our response
resp = await self.lookup(stream, selector)
resp = resp.encode(self.encoding)
resp_raw = await self.lookup(stream, selector_dec)
resp_enc = resp_raw.encode(self.encoding)
# Write and exit the connection
await stream.write(resp)
await stream.write(resp_enc)
await self.close_stream(stream)
except ValueError:
log.warning("Looking up file %r failed", selector)
log.warning("Looking up file %r failed", selector_dec)
await self.close_stream(stream)
except tornado.iostream.StreamClosedError:
log.debug("Lost connection from %s", address)
async def close_stream(self, stream):
async def close_stream(self, stream: tornado.iostream.IOStream) -> None:
"""Gopher connections are closed by writing a . on a single line then
closing the underlying transport."""
await stream.write(gb.protocol.eof.encode(self.encoding))
await stream.write(gb.protocol.crlf.encode(self.encoding))
async def lookup(self, stream, data):
async def lookup(self, stream: tornado.iostream.IOStream, data: str) -> str:
"""Lookup a selector on our mode."""
return self.mode.lookup(data)
@ -76,7 +86,7 @@ class ImplicitGopherServer(GopherServer):
while auto generating indexes for directories. If magic is enabled then
the mode will auto-guess filetypes."""
def __init__(self, path, magic, encoding):
def __init__(self, path: str, magic: bool, encoding: str) -> None:
log.info("Starting ImplicitGopherServer with path %s", path)

@ -0,0 +1,31 @@
# Specify the target platform details in config, so your developers are
# free to run mypy on Windows, Linux, or macOS and get consistent
# results.
# flake8-mypy expects the two following for sensible formatting
# show error messages from unrelated files
# suppress errors about unsatisfied imports
# be strict
# The following are off by default. Flip them on if you feel
# adventurous.
# No incremental mode