github/sprockets.mixins.mediatype
1
0
Fork 0
mirror of https://github.com/sprockets/sprockets.mixins.mediatype.git synced 2024-11-21 19:28:38 +00:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #41 from dave-shawley/add-typing

Browse source 
Add type annotations
This commit is contained in:
Kevin Vance 2021-10-05 10:22:49 -04:00 committed by GitHub
commit 4dbc74076d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
17 changed files with 610 additions and 175 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

4
README.rst
View file

@ -101,7 +101,7 @@ appropriate HTTP exceptions will be raised.
.. |Documentation| image:: https://readthedocs.org/projects/sprocketsmixinsmedia-type/badge/?version=latest
:target: https://sprocketsmixinsmedia-type.readthedocs.org/
.. |Build Badge| image:: https://travis-ci.org/sprockets/sprockets.mixins.mediatype.svg
:target: https://travis-ci.org/sprockets/sprockets.mixins.mediatype
.. |Build Badge| image:: https://github.com/sprockets/sprockets.mixins.mediatype/actions/workflows/testing.yml/badge.svg
:target: https://github.com/sprockets/sprockets.mixins.mediatype/actions/workflows/testing.yml
.. |Package Info| image:: https://img.shields.io/pypi/v/sprockets.mixins.mediatype.svg
:target: https://pypi.python.org/pypi/sprockets.mixins.mediatype

64
docs/api.rst
View file

@ -1,7 +1,27 @@
API Documentation
=================
.. currentmodule:: sprockets.mixins.mediatype.content
The easiest way to use this library is to:
#. call :func:`.install` when you create your application instance and specify a
default content type
#. call :func:`.add_transcoder` to install transcoders for the content types that
you support -- use :func:`.add_binary_content_type` and/or
:func:`.add_text_content_type` if you don't want to define a
:class:`~sprockets.mixins.mediatype.type_info.Transcoder` class.
#. include :class:`.ContentMixin` in your handler's inheritance chain
#. call :meth:`~.ContentMixin.get_request_body` to retrieve a request body
sent in any of the supported content types
#. call :meth:`~.ContentMixin.send_response` to send a response in any of the
supported content types
The :class:`.ContentMixin` will take care of inspecting the :http:header:`Content-Type`
header and deserialize the request as well as implementing the
:rfc:`proactive content negotiation algorithm <7231#section-3.4.1>` described in
:rfc:`7231` to serialize a response object appropriately.
Content Type Handling
---------------------
.. autoclass:: ContentMixin
@ -11,7 +31,7 @@ Content Type Registration
-------------------------
.. autofunction:: install
.. autofunction:: get_settings
.. autofunction:: add_transcoder
.. autofunction:: set_default_content_type
@ -19,7 +39,7 @@ Content Type Registration
.. autofunction:: add_text_content_type
.. autofunction:: add_transcoder
.. autofunction:: get_settings
.. autoclass:: ContentSettings
:members:
@ -33,3 +53,43 @@ Bundled Transcoders
.. autoclass:: MsgPackTranscoder
:members:
.. _type-info:
Python Type Information
-----------------------
The ``sprockets.mixins.mediatype.type_info`` module contains a number of
convenience type definitions for those you you who take advantage of type
annotations.
.. currentmodule:: sprockets.mixins.mediatype.type_info
Interface Types
~~~~~~~~~~~~~~~
.. autoclass:: Transcoder
:members:
.. autodata:: Serializable
.. autodata:: Deserialized
Convenience Types
~~~~~~~~~~~~~~~~~
.. autodata:: PackBFunction
.. autodata:: UnpackBFunction
.. autodata:: DumpSFunction
.. autodata:: LoadSFunction
Contract Types
~~~~~~~~~~~~~~
.. autoclass:: HasSettings
:members:
.. autoclass:: DefinesIsoFormat
:members:

31
docs/conf.py
View file

@ -3,10 +3,7 @@ import os
import pkg_resources
needs_sphinx = '4.0'
extensions = [
'sphinx.ext.autodoc', 'sphinx.ext.viewcode', 'sphinx.ext.intersphinx',
'sphinx.ext.extlinks', 'sphinxcontrib.httpdomain'
]
extensions = ['sphinx.ext.viewcode', 'sphinxcontrib.httpdomain']
master_doc = 'index'
project = 'sprockets.mixins.mediatype'
copyright = '2015-2021, AWeber Communications'
@ -24,14 +21,40 @@ html_sidebars = {
'**': ['about.html', 'navigation.html'],
}
# https://www.sphinx-doc.org/en/master/usage/extensions/intersphinx.html
extensions.append('sphinx.ext.intersphinx')
intersphinx_mapping = {
'ietfparse': ('https://ietfparse.readthedocs.io/en/latest', None),
'python': ('https://docs.python.org/3', None),
'requests': ('https://requests.readthedocs.org/en/latest/', None),
'sprockets': ('https://sprockets.readthedocs.org/en/latest/', None),
'tornado': ('https://www.tornadoweb.org/en/stable/', None),
}
# https://www.sphinx-doc.org/en/master/usage/extensions/autodoc.html
# We need to define type aliases for both the simple name (e.g., Deserialized)
# and the prefixed name (e.g., type_info.Deserialized) since both forms
# appear in the typing annotations.
extensions.append('sphinx.ext.autodoc')
autodoc_type_aliases = {
alias: f'sprockets.mixins.mediatype.type_info.{alias}'
for alias in {
'DefinesIsoFormat', 'Deserialized', 'DumpSFunction', 'HasSettings',
'LoadSFunction', 'MsgPackable', 'PackBFunction', 'Serializable',
'Transcoder', 'UnpackBFunction'
}
}
autodoc_type_aliases.update({
f'type_info.{alias}': f'sprockets.mixins.mediatype.type_info.{alias}'
for alias in {
'DefinesIsoFormat', 'Deserialized', 'DumpSFunction', 'HasSettings',
'LoadSFunction', 'MsgPackable', 'PackBFunction', 'Serializable',
'Transcoder', 'UnpackBFunction'
}
})
# https://www.sphinx-doc.org/en/master/usage/extensions/extlinks.html
extensions.append('sphinx.ext.extlinks')
extlinks = {
'compare': ('https://github.com/sprockets/sprockets.mixins.mediatype'
'/compare/%s', '%s')

8
docs/history.rst
View file

@ -1,6 +1,14 @@
Version History
===============
:compare:`Next <3.0.4...master>`
--------------------------------
- Add type annotations (see :ref:`type-info`)
- Return a "406 Not Acceptable" if the :http:header:`Accept` header values cannot be matched
and there is no default content type configured
- Deprecate not having a default content type configured
- Fail gracefully when a transcoder does not exist for the default content type
:compare:`3.0.4 <3.0.3...3.0.4>` (2 Nov 2020)
---------------------------------------------
- Return a "400 Bad Request" when an invalid Content-Type header is received

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)

15
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
@ -65,12 +72,18 @@ warning-is-error = 1
strict = 1
[coverage:report]
fail_under = 95
fail_under = 100
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

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

@ -27,12 +27,23 @@ 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
import warnings
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'
@ -52,13 +63,9 @@ class ContentSettings:
The settings instance contains the list of available content
types and handlers associated with them. Each handler implements
a simple interface:
- ``to_bytes(dict, encoding:str) -> bytes``
- ``from_bytes(bytes, encoding:str) -> dict``
Use the :func:`add_binary_content_type` and :func:`add_text_content_type`
helper functions to modify the settings for the application.
the :class:`~sprockets.mixins.mediatype.type_info.Transcoder`
interface. Use :func:`add_transcoder` to add support for a
specific content type to the application.
This class acts as a mapping from content-type string to the
appropriate handler instance. Add new content types and find
@ -76,27 +83,32 @@ class ContentSettings:
def make_application():
app = web.Application([('/', SomeHandler)])
add_binary_content_type(app, 'application/msgpack',
msgpack.packb, msgpack.unpackb)
add_text_content_type(app, 'application/json', 'utf-8',
json.dumps, json.loads)
add_transcoder(app, transcoders.JSONTranscoder())
add_transcoder(app, transcoders.MsgPackTranscoder())
return app
Of course, that is quite tedious, so use the :class:`.ContentMixin`
instead.
instead of using the settings directly.
"""
def __init__(self):
default_encoding: typing.Union[str, None]
_available_types: typing.List[datastructures.ContentType]
_default_content_type: typing.Union[str, None]
_handlers: typing.Dict[str, type_info.Transcoder]
def __init__(self) -> None:
self._handlers = {}
self._available_types = []
self.default_content_type = None
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 +119,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.
@ -121,22 +139,28 @@ class ContentSettings:
"""
return self._available_types
@property
def default_content_type(self) -> typing.Union[str, None]:
return self._default_content_type
def install(application, default_content_type, encoding=None):
"""
Install the media type management settings.
@default_content_type.setter
def default_content_type(self, new_value: typing.Union[str, None]) -> None:
if new_value is None:
warnings.warn(
DeprecationWarning(
'Using sprockets.mixins.mediatype without a default'
' content type is deprecated and will become an error'
' in a future version'))
self._default_content_type = new_value
: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,54 +168,75 @@ 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.
:param tornado.web.Application application:
:keyword bool force_instance: if :data:`True` then create the
:param application:
:param force_instance: if :data:`True` then create the
instance if it does not exist
:return: the content settings instance
:rtype: sprockets.mixins.mediatype.content.ContentSettings
:return: the content settings instance or :data:`None` if
`force_instance` is not :data:`True` and :func:`.install`
has not been called
"""
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.
:param tornado.web.Application application: the application to modify
:param str content_type: the content type to add
:param application: the application to modify
:param content_type: the content type to add
:param pack: function that packs a dictionary to a byte string.
``pack(dict) -> bytes``
See :any:`type_info.PackBFunction`
:param unpack: function that takes a byte string and returns a
dictionary. ``unpack(bytes) -> dict``
dictionary. See :any:`type_info.UnpackBFunction`
"""
add_transcoder(application,
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.
:param tornado.web.Application application: the application to modify
:param str content_type: the content type to add
:param str default_encoding: encoding to use when one is unspecified
:param application: the application to modify
:param content_type: the content type to add
:param default_encoding: encoding to use when one is unspecified
:param dumps: function that dumps a dictionary to a string.
``dumps(dict, encoding:str) -> str``
See :any:`type_info.DumpSFunction`
:param loads: function that loads a dictionary from a string.
``loads(str, encoding:str) -> dict``
See :any:`type_info.LoadSFunction`
Note that the ``charset`` parameter is stripped from `content_type`
if it is present.
@ -206,49 +251,37 @@ 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.
:param tornado.web.Application application: the application to modify
:param transcoder: object that translates between :class:`bytes` and
:class:`object` instances
:param str content_type: the content type to add. If this is
unspecified or :data:`None`, then the transcoder's ``content_type``
attribute is used.
:param application: the application to modify
:param transcoder: object that translates between :class:`bytes`
and object instances
:param content_type: the content type to add. If this is
unspecified or :data:`None`, then the transcoder's
``content_type`` attribute is used.
The `transcoder` instance is required to implement the following
simple protocol:
.. attribute:: transcoder.content_type
:class:`str` that identifies the MIME type that the transcoder
implements.
.. method:: transcoder.to_bytes(inst_data, encoding=None) -> bytes
:param object inst_data: the object to encode
:param str encoding: character encoding to apply or :data:`None`
:returns: the encoded :class:`bytes` instance
.. method:: transcoder.from_bytes(data_bytes, encoding=None) -> object
:param bytes data_bytes: the :class:`bytes` instance to decode
:param str encoding: character encoding to use or :data:`None`
:returns: the decoded :class:`object` instance
The `transcoder` instance is required to implement the
:class:`~sprockets.mixins.mediatype.type_info.Transcoder`
protocol.
"""
settings = get_settings(application, force_instance=True)
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.
:param tornado.web.Application application: the application to modify
:param str content_type: the content type to default to
:param str|None encoding: encoding to use when one is unspecified
:param application: the application to modify
:param content_type: the content type to default to
:param encoding: encoding to use when one is unspecified
"""
settings = get_settings(application, force_instance=True)
@ -256,7 +289,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 +309,26 @@ 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.
This method implements proactive content negotiation as
described in :rfc:`7231#section-3.4.1` using the
:http:header:`Accept` request header or the configured
default content type if the header is not present. The
selected response type is cached and returned. It will
be used when :meth:`.send_response` is called.
Note that this method is called by :meth:`.send_response`
so you will seldom need to call it directly.
"""
if self._best_response_match is None:
settings = get_settings(self.application, force_instance=True)
acceptable = headers.parse_accept(
@ -303,11 +348,11 @@ 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.
:raise web.HTTPError:
:raise tornado.web.HTTPError:
- if the content type cannot be matched, then the status code
is set to 415 Unsupported Media Type.
- if decoding the content body fails, then the status code is
@ -345,19 +390,41 @@ 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.
:param dict body: the body to serialize
:param bool set_content_type: should the :http:header:`Content-Type`
:param body: the body to serialize
:param set_content_type: should the :http:header:`Content-Type`
header be set? Defaults to :data:`True`
The transcoder for the response is selected by calling
:meth:`.get_response_content_type` which chooses an
appropriate transcoder based on the :http:header:`Accept`
header from the request.
"""
settings = get_settings(self.application, force_instance=True)
handler = settings[self.get_response_content_type()]
content_type, data_bytes = handler.to_bytes(body)
if set_content_type:
self.set_header('Content-Type', content_type)
self.add_header('Vary', 'Accept')
self.write(data_bytes)
response_type = self.get_response_content_type()
if response_type is None:
self._logger.error('failed to find a suitable response '
'content type for request')
self._logger.error('please set a default content type')
raise web.HTTPError(406)
try:
handler = settings[response_type]
except KeyError:
self._logger.error(
'no transcoder for the selected response content type %s, '
'is the default content type %r correct?', response_type,
settings.default_content_type)
raise web.HTTPError(500)
else:
content_type, data_bytes = handler.to_bytes(body)
if set_content_type:
self.set_header('Content-Type', content_type)
self.add_header('Vary', 'Accept')
self.write(data_bytes)

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

@ -7,14 +7,20 @@ 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:
"""
Pack and unpack binary types.
:param str content_type: registered content type
:param content_type: registered content type
:param pack: function that transforms an object instance
into :class:`bytes`
:param unpack: function that transforms :class:`bytes`
@ -24,17 +30,21 @@ 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`.
:param object inst_data: object to encode
:param str encoding: ignored
:param inst_data: object to encode
:param encoding: ignored
:returns: :class:`tuple` of the selected content
type and the :class:`bytes` representation of
`inst_data`
@ -42,15 +52,15 @@ 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`
:param bytes data_bytes: stream of bytes to decode
:param str encoding: ignored
:param dict content_parameters: optional :class:`dict` of
content type parameters from the :mailheader:`Content-Type`
header
:param data_bytes: stream of bytes to decode
:param encoding: ignored
:returns: decoded :class:`object` instance
"""
@ -61,12 +71,12 @@ class TextContentHandler:
"""
Transcodes between textual and object representations.
:param str content_type: registered content type
:param content_type: registered content type
:param dumps: function that transforms an object instance
into a :class:`str`
:param loads: function that transforms a :class:`str`
into an object instance
:param str default_encoding: encoding to apply when
:param default_encoding: encoding to apply when
transcoding from the underlying body :class:`byte`
instance
@ -76,18 +86,23 @@ 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`.
:param object inst_data: object to encode
:param str encoding: character set used to encode the bytes
:param inst_data: object to encode
:param encoding: character set used to encode the bytes
returned from the ``dumps`` function. This defaults to
:attr:`default_encoding`
:returns: :class:`tuple` of the selected content
@ -100,17 +115,17 @@ 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`
:param bytes data: stream of bytes to decode
:param str encoding: character set used to decode the incoming
:param data: stream of bytes to decode
:param encoding: character set used to decode the incoming
bytes before calling the ``loads`` function. This defaults
to :attr:`default_encoding`
:param dict content_parameters: optional :class:`dict` of
content type parameters from the :mailheader:`Content-Type`
header
:returns: decoded :class:`object` instance
"""

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

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

@ -5,28 +5,31 @@ 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
except ImportError: # pragma: no cover
umsgpack = None # type: ignore
from sprockets.mixins.mediatype import handlers
from sprockets.mixins.mediatype import handlers, type_info
class JSONTranscoder(handlers.TextContentHandler):
"""
JSON transcoder instance.
:param str content_type: the content type that this encoder instance
:param content_type: the content type that this encoder instance
implements. If omitted, ``application/json`` is used. This is
passed directly to the ``TextContentHandler`` initializer.
:param str default_encoding: the encoding to use if none is specified.
:param default_encoding: the encoding to use if none is specified.
If omitted, this defaults to ``utf-8``. This is passed directly to
the ``TextContentHandler`` initializer.
@ -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,31 +64,20 @@ 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.
:param object obj: the object to encode
:param obj: the object to encode
:return: the encoded object
:raises TypeError: when `obj` cannot be encoded
@ -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))
@ -119,7 +114,7 @@ class MsgPackTranscoder(handlers.BinaryContentHandler):
"""
Msgpack Transcoder instance.
:param str content_type: the content type that this encoder instance
:param content_type: the content type that this encoder instance
implements. If omitted, ``application/msgpack`` is used. This
is passed directly to the ``BinaryContentHandler`` initializer.
@ -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
"""
...

112
tests.py
View file

@ -4,10 +4,12 @@ import json
import os
import pickle
import struct
import unittest
import typing
import unittest.mock
import uuid
from tornado import testing
from ietfparse import algorithms
from tornado import httputil, testing, web
import umsgpack
from sprockets.mixins.mediatype import content, handlers, transcoders
@ -61,8 +63,15 @@ def pack_bytes(payload):
class SendResponseTests(testing.AsyncHTTPTestCase):
application: typing.Union[None, web.Application]
def setUp(self):
self.application = None
super().setUp()
def get_app(self):
return examples.make_application()
self.application = examples.make_application()
return self.application
def test_that_content_type_default_works(self):
response = self.fetch('/',
@ -129,6 +138,43 @@ class SendResponseTests(testing.AsyncHTTPTestCase):
self.assertEqual(response.code, 200)
self.assertEqual(response.headers['Content-Type'], 'expected/content')
def test_that_no_default_content_type_will_406(self):
# NB if the Accept header is omitted, then a default of `*/*` will
# be used which results in a match against any registered handler.
# Using an accept header forces the "no match" case.
settings = content.get_settings(self.application, force_instance=True)
settings.default_content_type = None
settings.default_encoding = None
response = self.fetch('/',
method='POST',
body='{}',
headers={
'Accept': 'application/xml',
'Content-Type': 'application/json',
})
self.assertEqual(response.code, 406)
def test_misconfigured_default_content_type(self):
settings = content.get_settings(self.application, force_instance=True)
settings.default_content_type = 'application/xml'
response = self.fetch('/',
method='POST',
body='{}',
headers={'Content-Type': 'application/json'})
self.assertEqual(response.code, 500)
def test_that_response_content_type_can_be_set(self):
class FooGenerator(content.ContentMixin, web.RequestHandler):
def get(self):
self.set_header('Content-Type', 'application/foo+json')
self.send_response({'foo': 'bar'}, set_content_type=False)
self.application.add_handlers(r'.*', [web.url(r'/foo', FooGenerator)])
response = self.fetch('/foo')
self.assertEqual(200, response.code)
self.assertEqual('application/foo+json',
response.headers.get('Content-Type'))
class GetRequestBodyTests(testing.AsyncHTTPTestCase):
def setUp(self):
@ -188,6 +234,49 @@ class GetRequestBodyTests(testing.AsyncHTTPTestCase):
self.assertEqual(response.code, 400)
class MixinCacheTests(unittest.TestCase):
def setUp(self):
super().setUp()
self.transcoder = transcoders.JSONTranscoder()
application = unittest.mock.Mock()
application.settings = {}
application.ui_methods = {}
content.install(application, 'application/json', 'utf-8')
content.add_transcoder(application, self.transcoder)
request = httputil.HTTPServerRequest(
'POST',
'/',
body=b'{}',
connection=unittest.mock.Mock(),
headers=httputil.HTTPHeaders({'Content-Type': 'application/json'}),
)
self.handler = content.ContentMixin(application, request)
def test_that_best_response_type_is_cached(self):
with unittest.mock.patch(
'sprockets.mixins.mediatype.content.algorithms.'
'select_content_type',
side_effect=algorithms.select_content_type
) as select_content_type:
first = self.handler.get_response_content_type()
second = self.handler.get_response_content_type()
self.assertIs(first, second)
self.assertEqual(1, select_content_type.call_count)
def test_that_request_body_is_cached(self):
self.transcoder.from_bytes = unittest.mock.Mock(
wraps=self.transcoder.from_bytes)
first = self.handler.get_request_body()
second = self.handler.get_request_body()
self.assertIs(first, second)
self.assertEqual(1, self.transcoder.from_bytes.call_count)
class JSONTranscoderTests(unittest.TestCase):
def setUp(self):
super().setUp()
@ -264,6 +353,11 @@ class ContentSettingsTests(unittest.TestCase):
'json')
self.assertEqual(settings['application/json; charset=utf-8'], handler)
def test_that_setting_no_default_content_type_warns(self):
settings = content.ContentSettings()
with self.assertWarns(DeprecationWarning):
settings.default_content_type = None
class ContentFunctionTests(unittest.TestCase):
def setUp(self):
@ -414,3 +508,15 @@ class MsgPackTranscoderTests(unittest.TestCase):
dumped = self.transcoder.packb(data)
self.assertEqual(self.transcoder.unpackb(dumped), data)
self.assertEqual(dumped, pack_bytes(data))
def test_that_dicts_are_sent_as_maps(self):
data = {'compact': True, 'schema': 0}
dumped = self.transcoder.packb(data)
self.assertEqual(b'\x82\xA7compact\xC3\xA6schema\x00', dumped)
def test_that_transcoder_creation_fails_if_umsgpack_is_missing(self):
with unittest.mock.patch(
'sprockets.mixins.mediatype.transcoders.umsgpack',
new_callable=lambda: None):
with self.assertRaises(RuntimeError):
transcoders.MsgPackTranscoder()

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,
]:
...