github/sprockets.mixins.mediatype
1
0
Fork 0
mirror of https://github.com/sprockets/sprockets.mixins.mediatype.git synced 2024-12-27 03:00:26 +00:00
Code Issues Projects Releases Packages Wiki Activity

Add type annotations.

Browse source 
The next commit is a pile of documentation updates that I didn't want
cluttering up this commit.
This commit is contained in:
Dave Shawley 2021-09-18 08:38:51 -04:00
parent a528661c98
commit acc0a1db14
No known key found for this signature in database
GPG key ID: F41A8A99298F8EED
12 changed files with 300 additions and 81 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
.github/workflows/testing.yml vendored
View file

@ -26,6 +26,9 @@ jobs:
- name: Formatting
run: |
yapf -pqr docs setup.py sprockets tests.py
- name: Typing
run: |
mypy sprockets/mixins/mediatype/ examples.py
test:
runs-on: ubuntu-latest
@ -43,8 +46,6 @@ jobs:
python -m pip install --upgrade pip setuptools
python -m pip install '.[ci]'
python -m pip install -e '.[msgpack]'
- name: Dump packages
run: python -m pip freeze
- name: Run tests
run: |
coverage run -m unittest tests.py

2
MANIFEST.in
View file

@ -1,4 +1,6 @@
include LICENSE
include tests.py
include sprockets/mixins/mediatype/py.typed
graft docs
graft requires
graft typestubs

11
examples.py
View file

@ -1,28 +1,29 @@
import logging
import signal
import typing
from sprockets.mixins.mediatype import content, transcoders
from tornado import ioloop, web
class SimpleHandler(content.ContentMixin, web.RequestHandler):
def post(self):
def post(self) -> None:
body = self.get_request_body()
self.set_status(200)
self.send_response(body)
def make_application(**settings):
def make_application(**settings: typing.Any) -> web.Application:
application = web.Application([('/', SimpleHandler)], **settings)
content.set_default_content_type(application, 'application/json',
content.set_default_content_type(application,
'application/json',
encoding='utf-8')
content.add_transcoder(application, transcoders.MsgPackTranscoder())
content.add_transcoder(application, transcoders.JSONTranscoder())
return application
def _signal_handler(signo, _):
def _signal_handler(signo: int, _: typing.Any) -> None:
logging.info('received signal %d, stopping application', signo)
iol = ioloop.IOLoop.instance()
iol.add_callback_from_signal(iol.stop)

13
setup.cfg
View file

@ -33,8 +33,10 @@ namespace_packages =
sprockets
sprockets.mixins
packages = find:
include_package_data = True
install_requires =
ietfparse>=1.5.1,<2
typing-extensions>=3.10 ; python_version<"3.8"
tornado>=5,<7
[options.extras_require]
@ -43,10 +45,12 @@ msgpack =
ci =
coverage==5.5
flake8==3.9.2
mypy==0.910
yapf==0.31.0
dev =
coverage==5.5
flake8==3.9.2
mypy==0.910
sphinx==4.2.0
sphinx-rtd-theme==1.0.0
sphinxcontrib-httpdomain==1.7.0
@ -57,6 +61,9 @@ docs =
sphinx-rtd-theme==1.0.0
sphinxcontrib-httpdomain==1.7.0
[options.package_data]
sprockets.mixins.mediatype = py.typed
[build_sphinx]
fresh-env = 1
warning-is-error = 1
@ -70,7 +77,13 @@ show_missing = 1
[coverage:run]
branch = 1
omit =
sprockets/mixins/mediatype/type_info.py
source = sprockets
[flake8]
exclude = build,env,.eggs
[mypy]
mypy_path = typestubs
strict = True

118
sprockets/mixins/mediatype/content.py
View file

@ -27,12 +27,21 @@ adds content handling methods to :class:`~tornado.web.RequestHandler`
instances.
"""
import logging
from __future__ import annotations
from ietfparse import algorithms, errors, headers
import logging
import typing
try:
from typing import Literal
except ImportError: # pragma: no cover
# "ignore" is required to avoid an incompatible import
# error due to different bindings of _SpecialForm
from typing_extensions import Literal # type: ignore
from ietfparse import algorithms, datastructures, errors, headers
from tornado import web
from . import handlers
from sprockets.mixins.mediatype import handlers, type_info
logger = logging.getLogger(__name__)
SETTINGS_KEY = 'sprockets.mixins.mediatype.ContentSettings'
@ -86,17 +95,24 @@ class ContentSettings:
instead.
"""
def __init__(self):
default_content_type: typing.Union[str, None]
default_encoding: typing.Union[str, None]
_handlers: typing.Dict[str, type_info.Transcoder]
_available_types: typing.List[datastructures.ContentType]
def __init__(self) -> None:
self._handlers = {}
self._available_types = []
self.default_content_type = None
self.default_encoding = None
def __getitem__(self, content_type):
def __getitem__(self, content_type: str) -> type_info.Transcoder:
parsed = headers.parse_content_type(content_type)
return self._handlers[str(parsed)]
def __setitem__(self, content_type, handler):
def __setitem__(self, content_type: str,
handler: type_info.Transcoder) -> None:
parsed = headers.parse_content_type(content_type)
content_type = str(parsed)
if content_type in self._handlers:
@ -107,11 +123,17 @@ class ContentSettings:
self._available_types.append(parsed)
self._handlers[content_type] = handler
def get(self, content_type, default=None):
def get(
self,
content_type: str,
default: typing.Union[type_info.Transcoder, None] = None
) -> typing.Union[type_info.Transcoder, None]:
"""Retrieve the handler for a specific content type."""
return self._handlers.get(content_type, default)
@property
def available_content_types(self):
def available_content_types(
self) -> typing.Sequence[datastructures.ContentType]:
"""
List of the content types that are registered.
@ -122,21 +144,13 @@ class ContentSettings:
return self._available_types
def install(application, default_content_type, encoding=None):
"""
Install the media type management settings.
:param tornado.web.Application application: the application to
install a :class:`.ContentSettings` object into.
:param str|NoneType default_content_type:
:param str|NoneType encoding:
:returns: the content settings instance
:rtype: sprockets.mixins.mediatype.content.ContentSettings
"""
def install(application: type_info.HasSettings,
default_content_type: typing.Optional[str],
encoding: typing.Optional[str] = None) -> ContentSettings:
"""Install the media type management settings and return it"""
try:
settings = application.settings[SETTINGS_KEY]
settings = typing.cast(ContentSettings,
application.settings[SETTINGS_KEY])
except KeyError:
settings = application.settings[SETTINGS_KEY] = ContentSettings()
settings.default_content_type = default_content_type
@ -144,7 +158,23 @@ def install(application, default_content_type, encoding=None):
return settings
def get_settings(application, force_instance=False):
@typing.overload
def get_settings(
application: type_info.HasSettings,
force_instance: Literal[False] = False
) -> typing.Union[ContentSettings, None]:
... # pragma: no cover
@typing.overload
def get_settings(application: type_info.HasSettings,
force_instance: Literal[True]) -> ContentSettings:
... # pragma: no cover
def get_settings(
application: type_info.HasSettings,
force_instance: bool = False) -> typing.Union[ContentSettings, None]:
"""
Retrieve the media type settings for a application.
@ -157,14 +187,16 @@ def get_settings(application, force_instance=False):
"""
try:
return application.settings[SETTINGS_KEY]
return typing.cast(ContentSettings, application.settings[SETTINGS_KEY])
except KeyError:
if not force_instance:
return None
return install(application, None)
def add_binary_content_type(application, content_type, pack, unpack):
def add_binary_content_type(application: type_info.HasSettings,
content_type: str, pack: type_info.PackBFunction,
unpack: type_info.UnpackBFunction) -> None:
"""
Add handler for a binary content type.
@ -180,8 +212,10 @@ def add_binary_content_type(application, content_type, pack, unpack):
handlers.BinaryContentHandler(content_type, pack, unpack))
def add_text_content_type(application, content_type, default_encoding, dumps,
loads):
def add_text_content_type(application: type_info.HasSettings,
content_type: str, default_encoding: str,
dumps: type_info.DumpSFunction,
loads: type_info.LoadSFunction) -> None:
"""
Add handler for a text content type.
@ -206,7 +240,9 @@ def add_text_content_type(application, content_type, default_encoding, dumps,
default_encoding))
def add_transcoder(application, transcoder, content_type=None):
def add_transcoder(application: type_info.HasSettings,
transcoder: type_info.Transcoder,
content_type: typing.Optional[str] = None) -> None:
"""
Register a transcoder for a specific content type.
@ -242,7 +278,9 @@ def add_transcoder(application, transcoder, content_type=None):
settings[content_type or transcoder.content_type] = transcoder
def set_default_content_type(application, content_type, encoding=None):
def set_default_content_type(application: type_info.HasSettings,
content_type: str,
encoding: typing.Optional[str] = None) -> None:
"""
Store the default content type for an application.
@ -256,7 +294,7 @@ def set_default_content_type(application, content_type, encoding=None):
settings.default_encoding = encoding
class ContentMixin:
class ContentMixin(web.RequestHandler):
"""
Mix this in to add some content handling methods.
@ -276,14 +314,15 @@ class ContentMixin:
using ``self.write()``.
"""
def initialize(self):
def initialize(self) -> None:
super().initialize()
self._request_body = None
self._best_response_match = None
self._request_body: typing.Optional[type_info.Deserialized] = None
self._best_response_match: typing.Optional[str] = None
self._logger = getattr(self, 'logger', logger)
def get_response_content_type(self):
"""Figure out what content type will be used in the response."""
def get_response_content_type(self) -> typing.Union[str, None]:
"""Select the content type will be used in the response.
"""
if self._best_response_match is None:
settings = get_settings(self.application, force_instance=True)
acceptable = headers.parse_accept(
@ -303,7 +342,7 @@ class ContentMixin:
return self._best_response_match
def get_request_body(self):
def get_request_body(self) -> type_info.Deserialized:
"""
Fetch (and cache) the request body as a dictionary.
@ -345,7 +384,9 @@ class ContentMixin:
return self._request_body
def send_response(self, body, set_content_type=True):
def send_response(self,
body: type_info.Serializable,
set_content_type: typing.Optional[bool] = True) -> None:
"""
Serialize and send ``body`` in the response.
@ -355,7 +396,8 @@ class ContentMixin:
"""
settings = get_settings(self.application, force_instance=True)
handler = settings[self.get_response_content_type()]
# TODO -- account for get_response_type returning None
handler = settings[self.get_response_content_type()] # type: ignore
content_type, data_bytes = handler.to_bytes(body)
if set_content_type:
self.set_header('Content-Type', content_type)

33
sprockets/mixins/mediatype/handlers.py
View file

@ -7,8 +7,14 @@ Basic content handlers.
to text before calling functions that encode & decode text
"""
from __future__ import annotations
import typing
from tornado import escape
from sprockets.mixins.mediatype import type_info
class BinaryContentHandler:
"""
@ -24,12 +30,16 @@ class BinaryContentHandler:
and unpacking functions.
"""
def __init__(self, content_type, pack, unpack):
def __init__(self, content_type: str, pack: type_info.PackBFunction,
unpack: type_info.UnpackBFunction) -> None:
self._pack = pack
self._unpack = unpack
self.content_type = content_type
def to_bytes(self, inst_data, encoding=None):
def to_bytes(
self,
inst_data: type_info.Serializable,
encoding: typing.Optional[str] = None) -> typing.Tuple[str, bytes]:
"""
Transform an object into :class:`bytes`.
@ -42,7 +52,10 @@ class BinaryContentHandler:
"""
return self.content_type, self._pack(inst_data)
def from_bytes(self, data_bytes, encoding=None):
def from_bytes(
self,
data_bytes: bytes,
encoding: typing.Optional[str] = None) -> type_info.Deserialized:
"""
Get an object from :class:`bytes`
@ -76,13 +89,18 @@ class TextContentHandler:
that tornado expects.
"""
def __init__(self, content_type, dumps, loads, default_encoding):
def __init__(self, content_type: str, dumps: type_info.DumpSFunction,
loads: type_info.LoadSFunction,
default_encoding: str) -> None:
self._dumps = dumps
self._loads = loads
self.content_type = content_type
self.default_encoding = default_encoding
def to_bytes(self, inst_data, encoding=None):
def to_bytes(
self,
inst_data: type_info.Serializable,
encoding: typing.Optional[str] = None) -> typing.Tuple[str, bytes]:
"""
Transform an object into :class:`bytes`.
@ -100,7 +118,10 @@ class TextContentHandler:
dumped = self._dumps(escape.recursive_unicode(inst_data))
return content_type, dumped.encode(selected)
def from_bytes(self, data, encoding=None):
def from_bytes(
self,
data: bytes,
encoding: typing.Optional[str] = None) -> type_info.Deserialized:
"""
Get an object from :class:`bytes`

0
sprockets/mixins/mediatype/py.typed Normal file
View file

54
sprockets/mixins/mediatype/transcoders.py
View file

@ -5,18 +5,21 @@ Bundled media type transcoders.
- :class:`.MsgPackTranscoder` implements msgpack encoding/decoding
"""
from __future__ import annotations
import base64
import json
import typing
import uuid
import collections
import collections.abc
try:
import umsgpack
except ImportError:
umsgpack = None
umsgpack = None # type: ignore
from sprockets.mixins.mediatype import handlers
from sprockets.mixins.mediatype import handlers, type_info
class JSONTranscoder(handlers.TextContentHandler):
@ -47,9 +50,12 @@ class JSONTranscoder(handlers.TextContentHandler):
:meth:`.loads` is called.
"""
dump_options: typing.Dict[str, typing.Any]
load_options: typing.Dict[str, typing.Any]
def __init__(self,
content_type='application/json',
default_encoding='utf-8'):
content_type: str = 'application/json',
default_encoding: str = 'utf-8') -> None:
super().__init__(content_type, self.dumps, self.loads,
default_encoding)
self.dump_options = {
@ -58,27 +64,16 @@ class JSONTranscoder(handlers.TextContentHandler):
}
self.load_options = {}
def dumps(self, obj):
"""
Dump a :class:`object` instance into a JSON :class:`str`
:param object obj: the object to dump
:return: the JSON representation of :class:`object`
"""
def dumps(self, obj: type_info.Serializable) -> str:
"""Dump a :class:`object` instance into a JSON :class:`str`"""
return json.dumps(obj, **self.dump_options)
def loads(self, str_repr):
"""
Transform :class:`str` into an :class:`object` instance.
def loads(self, str_repr: str) -> type_info.Deserialized:
"""Transform :class:`str` into an :class:`object` instance."""
return typing.cast(type_info.Deserialized,
json.loads(str_repr, **self.load_options))
:param str str_repr: the UNICODE representation of an object
:return: the decoded :class:`object` representation
"""
return json.loads(str_repr, **self.load_options)
def dump_object(self, obj):
def dump_object(self, obj: type_info.Serializable) -> str:
"""
Called to encode unrecognized object.
@ -109,7 +104,7 @@ class JSONTranscoder(handlers.TextContentHandler):
if isinstance(obj, uuid.UUID):
return str(obj)
if hasattr(obj, 'isoformat'):
return obj.isoformat()
return typing.cast(type_info.DefinesIsoFormat, obj).isoformat()
if isinstance(obj, (bytes, bytearray, memoryview)):
return base64.b64encode(obj).decode('ASCII')
raise TypeError('{!r} is not JSON serializable'.format(obj))
@ -132,22 +127,23 @@ class MsgPackTranscoder(handlers.BinaryContentHandler):
"""
PACKABLE_TYPES = (bool, int, float)
def __init__(self, content_type='application/msgpack'):
def __init__(self, content_type: str = 'application/msgpack') -> None:
if umsgpack is None:
raise RuntimeError('Cannot import MsgPackTranscoder, '
'umsgpack is not available')
super().__init__(content_type, self.packb, self.unpackb)
def packb(self, data):
def packb(self, data: type_info.Serializable) -> bytes:
"""Pack `data` into a :class:`bytes` instance."""
return umsgpack.packb(self.normalize_datum(data))
def unpackb(self, data):
def unpackb(self, data: bytes) -> type_info.Deserialized:
"""Unpack a :class:`object` from a :class:`bytes` instance."""
return umsgpack.unpackb(data)
def normalize_datum(self, datum):
def normalize_datum(
self, datum: type_info.Serializable) -> type_info.MsgPackable:
"""
Convert `datum` into something that umsgpack likes.
@ -226,7 +222,7 @@ class MsgPackTranscoder(handlers.BinaryContentHandler):
datum = datum.tobytes()
if hasattr(datum, 'isoformat'):
datum = datum.isoformat()
datum = typing.cast(type_info.DefinesIsoFormat, datum).isoformat()
if isinstance(datum, (bytes, str)):
return datum

110
sprockets/mixins/mediatype/type_info.py Normal file
View file

@ -0,0 +1,110 @@
from __future__ import annotations
import typing
import uuid
try:
from typing import Protocol
except ImportError:
# "ignore" is required to avoid an incompatible import
# error due to different bindings of _SpecialForm
from typing_extensions import Protocol # type: ignore
class DefinesIsoFormat(Protocol):
"""An object that has an isoformat method."""
def isoformat(self) -> str:
"""Return the date/time in ISO-8601 format."""
...
class HasSettings(Protocol):
"""Something that quacks like a tornado.web.Application."""
settings: typing.Dict[str, typing.Any]
"""Application settings."""
Serializable = typing.Union[DefinesIsoFormat, None, bool, bytearray, bytes,
float, int, memoryview, str, typing.Mapping,
typing.Sequence, typing.Set, uuid.UUID]
"""Types that can be serialized by this library.
This is the set of types that
:meth:`sprockets.mixins.mediatype.content.ContentMixin.send_response`
is capable for serializing.
"""
Deserialized = typing.Union[None, bytes, typing.Mapping, float, int, list, str]
"""Possible result of deserializing a body.
This is the set of types that
:meth:`sprockets.mixins.mediatype.content.ContentMixin.get_request_body`
might return.
"""
PackBFunction = typing.Callable[[Serializable], bytes]
"""Signature of a binary content handler's serialization hook."""
UnpackBFunction = typing.Callable[[bytes], Deserialized]
"""Signature of a binary content handler's deserialization hook."""
DumpSFunction = typing.Callable[[Serializable], str]
"""Signature of a text content handler's serialization hook."""
LoadSFunction = typing.Callable[[str], Deserialized]
"""Signature of a text content handler's deserialization hook."""
MsgPackable = typing.Union[None, bool, bytes, typing.Dict[typing.Any,
typing.Any], float,
int, typing.List[typing.Any], str]
"""Set of types that the underlying msgpack library can serialize."""
class Transcoder(Protocol):
"""Object that transforms objects to bytes and back again.
Transcoder instances are identified by their `content_type`
instance attribute and registered by calling
:func:`~sprockets.mixins.mediatype.content.add_transcoder`.
They are used to implement request deserialization
(:meth:`~sprockets.mixins.mediatype.content.ContentMixin.get_request_body`)
and response body serialization
(:meth:`~sprockets.mixins.mediatype.content.ContentMixin.send_response`)
"""
content_type: str
"""Canonical content type that this transcoder implements."""
def to_bytes(
self,
inst_data: Serializable,
encoding: typing.Optional[str] = None) -> typing.Tuple[str, bytes]:
"""Serialize `inst_data` into a byte stream and content type spec.
:param inst_data: the data to serialize
:param encoding: optional encoding to use when serializing
The content type is returned since it may contain the encoding
or character set as a parameter. The `encoding` parameter may
not be used by all transcoders.
:returns: tuple of the content type and the resulting bytes
"""
...
def from_bytes(self,
data_bytes: bytes,
encoding: typing.Optional[str] = None) -> Deserialized:
"""Deserialize `bytes` into a Python object instance.
:param data_bytes: byte string to deserialize
:param encoding: optional encoding to use when deserializing
The `encoding` parameter may not be used by all transcoders.
:returns: the decoded Python object
"""
...

6
tox.ini
View file

@ -1,5 +1,5 @@
[tox]
envlist = py37,py38,py39,coverage,docs,lint
envlist = py37,py38,py39,coverage,docs,lint,typecheck
indexserver =
default = https://pypi.python.org/simple
toxworkdir = build/tox
@ -26,3 +26,7 @@ commands =
commands =
flake8 sprockets tests.py
yapf -dr docs setup.py sprockets tests.py
[testenv:typecheck]
commands =
mypy sprockets/mixins/mediatype/ examples.py

2
typestubs/README.rst Normal file
View file

@ -0,0 +1,2 @@
Typestubs for external libraries
================================

27
typestubs/umsgpack.pyi Normal file
View file

@ -0,0 +1,27 @@
import typing
def packb(
obj: typing.Union[
None,
bytes,
typing.Dict[typing.Any, typing.Any],
float,
int,
typing.Sequence[typing.Any],
str,
],
ext_handlers: typing.Optional[typing.Dict[str, typing.Any]] = None,
force_flat_precision: typing.Optional[str] = None) -> bytes:
...
def unpackb(val: bytes) -> typing.Union[
None,
bytes,
typing.Dict[typing.Any, typing.Any],
float,
int,
typing.List[typing.Any],
str,
]:
...