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

Merge pull request #2 from sprockets/initial-implementation

Browse source 
Initial implementation
This commit is contained in:
Dan Tracy 2015-09-09 16:07:55 -04:00
commit f8352066de
20 changed files with 639 additions and 210 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
.travis.yml
View file

@ -1,13 +1,12 @@
language: python
python:
- 2.7
- 3.2
- 3.3
- 3.4
- pypy
install:
- pip install codecov
- pip install -r test-requirements.txt
- pip install -r requires/installation.txt
- pip install -r requires/testing.txt
script: nosetests
after_success:
- codecov

5
MANIFEST.in
View file

@ -1,3 +1,4 @@
include LICENSE
include README.rst
include test-requirements.txt
include tests.py
graft docs
graft requires

119
README.rst
View file

@ -1,93 +1,66 @@
sprockets.mixins.media_type
===========================
A mixin that performs Content-Type negotiation and request/response (de)serialization.
A mixin that performs Content-Type negotiation and request/response
(de)serialization.
|Version| |Downloads| |Status| |Coverage| |License|
|Documentation| |Build Badge| |Package Info|
Installation
------------
``sprockets.mixins.media_type`` is available on the
`Python Package Index <https://pypi.python.org/pypi/sprockets.mixins.media_type>`_
and can be installed via ``pip`` or ``easy_install``:
This mix-in adds two methods to a ``tornado.web.RequestHandler`` instance:
.. code:: bash
- ``get_request_body() -> dict``: deserializes the request body according
to the HTTP ``Content-Type`` header and returns the deserialized body.
- ``send_response(object)``: serializes the response into the content type
requested by the ``Accept`` header.
pip install sprockets.mixins.media_type
Support for a content types is enabled by calling either the
``add_binary_content_type`` or ``add_text_content_type`` function with the
``tornado.web.Application`` instance, the content type, encoding and decoding
functions as parameters:
Documentation
-------------
http://sprocketsmixinsmedia-type.readthedocs.org/en/latest/
.. code-block:: python
Example
-------
The following example demonstrates how to use the Mix-in to handle media
type validation and serialization.
import json
.. code:: python
from sprockets.mixins import media_type
from tornado import web
from tornado import web, gen
from sprockets.mixins import media_type
def make_application():
application = web.Application([
# insert your handlers here
])
media_type.add_text_content_type(application,
'application/json', 'utf-8',
json.dumps, json.loads)
class MyRequestHandler(media_type.MediaTypeMixin, web.RequestHandler):
return application
@gen.coroutine
def post(self, **kwargs):
# Validate the Content-Type header using the Mix-in
if not self.is_valid_content_type():
self.set_status(415, 'Unsupported content type')
self.finish()
return
The *add content type* functions will add a attribute to the ``Application``
instance that the mix-in uses to manipulate the request and response bodies.
# Deserialize your request payload
data = self.decode_request()
.. code-block:: python
# Ensure that you get some data out of it!
if not data:
self.set_status(400)
self.finish()
return
from sprockets.mixins import media_type
from tornado import web
# Manipulate your data and do business stuff with it
data.pop('the_key')
class SomeHandler(media_type.ContentMixin, web.RequestHandler):
def get(self):
self.send_response({'data': 'value'})
self.finish()
self.set_status(200)
def post(self):
body = self.get_request_body()
# do whatever
self.send_response({'action': 'performed'})
self.finish()
# Automatically serialize your data using the HTTP Accept headers
self.write(data)
Based on the settings stored in the ``Application`` instance and the HTTP
headers, the request and response data will be handled correctly or the
appropriate HTTP exceptions will be raised.
@gen.coroutine
def get(self, some_id):
# Validate the Accept headers using the Mix-in
if not self.is_valid_accept_header():
self.set_status(406, 'Invalid Accept header')
self.finish()
return
# Maybe do some lookups from the database or get some data from somewhere
data = {'some_id': some_id}
self.set_status(200)
# Automatically serialize your data using the HTTP Accept headers
self.write(data)
Version History
---------------
Available at http://sprocketsmixinsmedia-type.readthedocs.org/en/latest//en/latest/history.html
.. |Version| image:: https://img.shields.io/pypi/v/sprockets.mixins.media_type.svg?
:target: http://badge.fury.io/py/sprockets.mixins.media_type
.. |Status| image:: https://img.shields.io/travis/sprockets/sprockets.mixins.media_type.svg?
.. |Documentation| image:: https://readthedocs.org/projects/sprocketsmixinsmedia-type/?badge=latest
:target: https://sprocketsmixinsmedia_type.readthedocs.org/
.. |Build Badge| image:: https://travis-ci.org/sprockets/sprockets.mixins.media_type.svg
:target: https://travis-ci.org/sprockets/sprockets.mixins.media_type
.. |Coverage| image:: https://img.shields.io/codecov/c/github/sprockets/sprockets.mixins.media_type.svg?
:target: https://codecov.io/github/sprockets/sprockets.mixins.media_type?branch=master
.. |Downloads| image:: https://img.shields.io/pypi/dm/sprockets.mixins.media_type.svg?
:target: https://pypi.python.org/pypi/sprockets.mixins.media_type
.. |License| image:: http://img.shields.io/:license-mit-blue.svg
:target: http://doge.mit-license.org
.. |Package Info| image:: https://img.shields.io/pypi/v/sprockets.mixins.media_type.svg
:target: https://pypi.python.org/sprockets.mixins.media_type

19
docs/api.rst
View file

@ -1,2 +1,19 @@
.. automodule:: sprockets.mixins.media_type
API Documentation
=================
.. currentmodule:: sprockets.mixins.media_type
Content Type Handling
---------------------
.. autoclass:: ContentMixin
:members:
Content Type Registration
-------------------------
.. autofunction:: set_default_content_type
.. autofunction:: add_binary_content_type
.. autofunction:: add_text_content_type
.. autoclass:: ContentSettings
:members:

22
docs/conf.py
View file

@ -1,9 +1,11 @@
import alabaster
from sprockets.mixins.media_type import __version__
needs_sphinx = '1.0'
extensions = ['sphinx.ext.autodoc',
'sphinx.ext.viewcode',
'sphinx.ext.intersphinx']
'sphinx.ext.intersphinx',
'sphinxcontrib.autohttp.tornado']
source_suffix = '.rst'
master_doc = 'index'
project = 'sprockets.mixins.media_type'
@ -11,8 +13,26 @@ copyright = '2015, AWeber Communications'
release = __version__
version = '.'.join(release.split('.')[0:1])
pygments_style = 'sphinx'
html_theme = 'alabaster'
html_style = 'custom.css'
html_static_path = ['static']
html_theme_path = [alabaster.get_path()]
html_sidebars = {
'**': ['about.html', 'navigation.html'],
}
html_theme_options = {
'github_user': 'sprockets',
'github_repo': 'sprockets.mixins.media_type',
'description': 'Content-Type negotation mix-in',
'github_banner': True,
'travis_button': True,
'sidebar_width': '230px',
}
intersphinx_mapping = {
'python': ('https://docs.python.org/', None),
'requests': ('https://requests.readthedocs.org/en/latest/', None),
'sprockets': ('https://sprockets.readthedocs.org/en/latest/', None),
'tornado': ('http://tornadoweb.org/en/latest/', None),
}

112
docs/contributing.rst Normal file
View file

@ -0,0 +1,112 @@
How to Contribute
=================
Do you want to contribute fixes or improvements?
**AWesome!** *Thank you very much, and let's get started.*
Set up a development environment
--------------------------------
The first thing that you need is a development environment so that you can
run the test suite, update the documentation, and everything else that is
involved in contributing. The easiest way to do that is to create a virtual
environment for your endevours::
$ virtualenv -p python2.7 env
Don't worry about writing code against previous versions of Python unless
you you don't have a choice. That is why we run our tests through `tox`_.
If you don't have a choice, then install `virtualenv`_ to create the
environment instead. The next step is to install the development tools
that this project uses. These are listed in *requires/development.txt*::
$ env/bin/pip install -qr requires/development.txt
At this point, you will have everything that you need to develop at your
disposal. *setup.py* is the swiss-army knife in your development tool
chest. It provides the following commands:
**./setup.py nosetests**
Run the test suite using `nose`_ and generate a nice coverage report.
**./setup.py build_sphinx**
Generate the documentation using `sphinx`_.
**./setup.py flake8**
Run `flake8`_ over the code and report style violations.
If any of the preceding commands give you problems, then you will have to
fix them **before** your pull request will be accepted.
Running Tests
-------------
The easiest (and quickest) way to run the test suite is to use the
*nosetests* command. It will run the test suite against the currently
installed python version and report not only the test result but the
test coverage as well::
$ ./setup.py nosetests
running nosetests
running egg_info
writing dependency_links to sprockets.mixins.media_type.egg-info/dependency_links.txt
writing top-level names to sprockets.mixins.media_type.egg-info/top_level.txt
writing sprockets.mixins.media_type.egg-info/PKG-INFO
reading manifest file 'sprockets.mixins.media_type.egg-info/SOURCES.txt'
reading manifest template 'MANIFEST.in'
warning: no previously-included files matching '__pycache__'...
warning: no previously-included files matching '*.swp' found ...
writing manifest file 'sprockets.mixins.media_type.egg-info/SOURCES.txt'
...
Name Stmts Miss Branch BrMiss Cover Missing
----------------------------------------------------------------------
...
----------------------------------------------------------------------
TOTAL 95 2 59 2 97%
----------------------------------------------------------------------
Ran 44 tests in 0.054s
OK
That's the quick way to run tests. The slightly longer way is to run
the `detox`_ utility. It will run the test suite against all of the
supported python versions in parallel. This is essentially what Travis-CI
will do when you issue a pull request anyway::
$ env/bin/detox
py27 recreate: /.../sprockets.mixins.media_type/build/tox/py27
GLOB sdist-make: /.../sprockets.mixins.media_type/setup.py
py34 recreate: /.../sprockets.mixins.media_type/build/tox/py34
py27 installdeps: -rtest-requirements.txt, mock
py34 installdeps: -rtest-requirements.txt
py27 inst: /.../sprockets.mixins.media_type/build/tox/dist/sprockets.mixins.media_type-0.0.0.zip
py27 runtests: PYTHONHASHSEED='2156646470'
py27 runtests: commands[0] | /../sprockets.mixins.media_type/build/tox/py27/bin/nosetests
py34 inst: /../sprockets.mixins.media_type/.build/tox/dist/sprockets.mixins.media_type-0.0.0.zip
py34 runtests: PYTHONHASHSEED='2156646470'
py34 runtests: commands[0] | /.../sprockets.mixins.media_type/build/tox/py34/bin/nosetests
_________________________________ summary _________________________________
py27: commands succeeded
py34: commands succeeded
congratulations :)
This is what you want to see. Now you can make your modifications and keep
the tests passing.
Submitting a Pull Request
-------------------------
Once you have made your modifications, gotten all of the tests to pass,
and added any necessary documentation, it is time to contribute back for
posterity. You've probably already cloned this repository and created a
new branch. If you haven't, then checkout what you have as a branch and
roll back *master* to where you found it. Then push your repository up
to github and issue a pull request. Describe your changes in the request,
if Travis isn't too annoyed someone will review it, and eventually merge
it back.
.. _flake8: http://flake8.readthedocs.org/
.. _nose: http://nose.readthedocs.org/
.. _sphinx: http://sphinx-doc.org/
.. _detox: http://testrun.org/tox/
.. _tox: http://testrun.org/tox/
.. _virtualenv: http://virtualenv.pypa.io/

52
docs/examples.rst
View file

@ -1,54 +1,4 @@
Examples
========
The following example demonstrates how to use the Mix-in to handle media
type validation and serialization.
.. code:: python
from tornado import web, gen
from sprockets.mixins import media_type
class MyRequestHandler(media_type.MediaTypeMixin, web.RequestHandler):
@gen.coroutine
def post(self, **kwargs):
# Validate the Content-Type header using the Mix-in
if not self.is_valid_content_type():
self.set_status(415, 'Unsupported content type')
self.finish()
return
# Deserialize your request payload
data = self.decode_request()
# Ensure that you get some data out of it!
if not data:
self.set_status(400)
self.finish()
return
# Manipulate your data and do business stuff with it
data.pop('the_key')
self.set_status(200)
# Automatically serialize your data using the HTTP Accept headers
self.write(data)
@gen.coroutine
def get(self, some_id):
# Validate the Accept headers using the Mix-in
if not self.is_valid_accept_header():
self.set_status(406, 'Invalid Accept header')
self.finish()
return
# Maybe do some lookups from the database or get some data from somewhere
data = {'some_id': some_id}
self.set_status(200)
# Automatically serialize your data using the HTTP Accept headers
self.write(data)
.. literalinclude:: ../examples.py

65
docs/index.rst
View file

@ -1,61 +1,16 @@
sprockets.mixins.media_type
===========================
A mixin that performs Content-Type negotiation and request/response (de)serialization.
.. include:: ../README.rst
|Version| |Downloads| |Status| |Coverage| |License|
Installation
------------
``sprockets.mixins.media_type`` is available on the
`Python Package Index <https://pypi.python.org/pypi/sprockets.mixins.media_type>`_
and can be installed via ``pip`` or ``easy_install``:
.. code:: bash
pip install sprockets.mixins.media_type
API Documentation
-----------------
.. toctree::
:maxdepth: 2
api
examples
Version History
---------------
See :doc:`history`
Issues
------
Please report any issues to the Github project at `https://github.com/sprockets/sprockets.mixins.media_type/issues <https://github.com/sprockets/sprockets.mixins.media_type/issues>`_
Source
------
``sprockets.mixins.media_type`` source is available on Github at `https://github.com/sprockets/sprockets.mixins.media_type <https://github.com/sprockets/sprockets.mixins.media_type>`_
Examples
--------
.. literalinclude:: ../examples.py
License
-------
``sprockets.mixins.media_type`` is released under the `3-Clause BSD license <https://github.com/sprockets/sprockets.mixins.media_type/blob/master/LICENSE>`_.
.. include:: ../LICENSE
Indices and tables
------------------
.. toctree::
:hidden:
* :ref:`genindex`
* :ref:`modindex`
* :ref:`search`
.. |Version| image:: https://img.shields.io/pypi/v/sprockets.mixins.media_type.svg?
:target: http://badge.fury.io/py/sprockets.mixins.media_type
.. |Status| image:: https://img.shields.io/travis/sprockets/sprockets.mixins.media_type.svg?
:target: https://travis-ci.org/sprockets/sprockets.mixins.media_type
.. |Coverage| image:: https://img.shields.io/codecov/c/github/sprockets/sprockets.mixins.media_type.svg?
:target: https://codecov.io/github/sprockets/sprockets.mixins.media_type?branch=master
.. |Downloads| image:: https://img.shields.io/pypi/dm/sprockets.mixins.media_type.svg?
:target: https://pypi.python.org/pypi/sprockets.mixins.media_type
.. |License| image:: http://img.shields.io/:license-mit-blue.svg
:target: http://doge.mit-license.org
api
contributing
history

4
docs/static/custom.css vendored Normal file
View file

@ -0,0 +1,4 @@
@import url("alabaster.css");
h1.logo {
font-size: 12pt;
}

43
examples.py Normal file
View file

@ -0,0 +1,43 @@
import json
import logging
import signal
from sprockets.mixins import media_type
from tornado import ioloop, web
import msgpack
class SimpleHandler(media_type.ContentMixin, web.RequestHandler):
def post(self):
body = self.get_request_body()
self.set_status(200)
self.send_response(body)
self.finish()
def make_application(**settings):
application = web.Application([web.url(r'/', SimpleHandler)], **settings)
media_type.set_default_content_type(application, 'application/json',
encoding='utf-8')
media_type.add_binary_content_type(application, 'application/msgpack',
msgpack.packb, msgpack.unpackb)
media_type.add_text_content_type(application, 'application/json', 'utf-8',
json.dumps, json.loads)
return application
def _signal_handler(signo, _):
logging.info('received signal %d, stopping application', signo)
iol = ioloop.IOLoop.instance()
iol.add_callback_from_signal(iol.stop)
if __name__ == '__main__':
logging.basicConfig(level=logging.DEBUG,
format='%(levelname)1.1s - %(name)s: %(message)s')
application = make_application(debug=True)
application.listen(8000)
signal.signal(signal.SIGINT, _signal_handler)
signal.signal(signal.SIGTERM, _signal_handler)
ioloop.IOLoop.instance().start()

2
requirements.txt
View file

@ -1,2 +0,0 @@
tornado
msgpack-python

4
requires/development.txt Normal file
View file

@ -0,0 +1,4 @@
-r testing.txt
-r installation.txt
sphinx>=1.2,<2
sphinxcontrib-httpdomain>=1.3,<2

2
requires/installation.txt Normal file
View file

@ -0,0 +1,2 @@
ietfparse>=1.2.2,<2
tornado>=3.2,<5

4
requires/testing.txt Normal file
View file

@ -0,0 +1,4 @@
coverage>=3.7,<3.99 # prevent installing 4.0b on ALL pip versions
mock>=1.3,<2
msgpack-python>=0.4,<0.5
nose>=1.3,<2

43
setup.py Normal file → Executable file
View file

@ -1,10 +1,37 @@
#!/usr/bin/env python
#
import os
import setuptools
def read_requirements(file_name):
requirements = []
try:
with open(os.path.join('requires', file_name)) as req_file:
for req_line in req_file:
req_line = req_line.strip()
if '#' in req_line:
req_line = req_line[0:req_line.find('#')].strip()
if req_line.startswith('-r'):
req_line = req_line[2:].strip()
requirements.extend(read_requirements(req_line))
else:
requirements.append(req_line)
except IOError:
pass
return requirements
install_requires = read_requirements('install.txt')
setup_requires = read_requirements('setup.txt')
tests_require = read_requirements('testing.txt')
setuptools.setup(
name='sprockets.mixins.media_type',
version='0.0.0',
description='A mixin for reporting handling content-type/accept headers',
long_description=open('test-requirements.txt', 'r').read(),
long_description='\n' + open('README.rst').read(),
url='https://github.com/sprockets/sprockets.mixins.media_type',
author='AWeber Communications',
author_email='api@aweber.com',
@ -27,12 +54,10 @@ setuptools.setup(
'Topic :: Software Development :: Libraries',
'Topic :: Software Development :: Libraries :: Python Modules'
],
packages=['sprockets',
'sprockets.mixins',
'sprockets.mixins.media_type'],
package_data={'': ['LICENSE', 'README.md']},
include_package_data=True,
install_requires=['tornado'],
namespace_packages=['sprockets',
'sprockets.mixins'],
packages=setuptools.find_packages(),
install_requires=install_requires,
setup_requires=setup_requires,
tests_require=tests_require,
namespace_packages=['sprockets', 'sprockets.mixins'],
test_suite='nose.collector',
zip_safe=False)

263
sprockets/mixins/media_type.py Normal file
View file

@ -0,0 +1,263 @@
"""
sprockets.mixins.media_type
===========================
"""
import logging
from ietfparse import algorithms, errors, headers
from tornado import escape, web
version_info = (0, 0, 0)
__version__ = '.'.join(str(v) for v in version_info)
logger = logging.getLogger(__name__)
class ContentSettings(object):
"""
Content selection settings.
An instance of this class is stashed as the ``_content_settings``
attribute on the application object. It 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.
This class acts as a mapping from content-type string to the
appropriate handler instance. Add new content types and find
handlers using the common ``dict`` syntax:
.. code-block:: python
class SomeHandler(web.RequestHandler):
def get(self):
settings = ContentSettings.from_application(self.application)
response_body = settings['application/msgpack'].to_bytes(
response_dict, encoding='utf-8')
self.write(response_body)
self.finish()
def make_application():
app = web.Application([web.url('/', 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)
return app
Of course, that is quite tedious, so use the :class:`.ContentMixin`
instead.
"""
def __init__(self):
self._handlers = {}
self._available_types = []
self.default_content_type = None
self.default_encoding = None
def __getitem__(self, content_type):
return self._handlers[content_type]
def __setitem__(self, content_type, handler):
if content_type in self._handlers:
logger.warning('handler for %s already set to %r',
content_type, self._handers[content_type])
return
self._available_types.append(headers.parse_content_type(content_type))
self._handlers[content_type] = handler
def get(self, content_type, default=None):
return self._handlers.get(content_type, default)
@classmethod
def from_application(cls, application):
"""Retrieve the content settings from an application."""
if not hasattr(application, '_content_settings'):
setattr(application, '_content_settings', cls())
return application._content_settings
@property
def available_content_types(self):
"""
List of the content types that are registered.
This is a sequence of :class:`ietfparse.datastructures.ContentType`
instances.
"""
return self._available_types
def add_binary_content_type(application, content_type, pack, unpack):
"""
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 pack: function that packs a dictionary to a byte string.
``pack(dict) -> bytes``
:param unpack: function that takes a byte string and returns a
dictionary. ``unpack(bytes) -> dict``
"""
settings = ContentSettings.from_application(application)
settings[content_type] = _BinaryContentHandler(content_type, pack, unpack)
def add_text_content_type(application, content_type, default_encoding,
dumps, loads):
"""
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 dumps: function that dumps a dictionary to a string.
``dumps(dict, encoding:str) -> str``
:param loads: function that loads a dictionary from a string.
``loads(str, encoding:str) -> dict``
"""
settings = ContentSettings.from_application(application)
settings[content_type] = _TextContentHandler(content_type, dumps, loads,
default_encoding)
def set_default_content_type(application, content_type, encoding=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
"""
settings = ContentSettings.from_application(application)
settings.default_content_type = content_type
settings.default_encoding = encoding
class ContentMixin(object):
"""
Mix this in to add some content handling methods.
.. code-block:: python
class MyHandler(ContentMixin, web.RequestHandler):
def post(self):
body = self.get_request_body()
# do stuff --> response_dict
self.send_response(response_dict)
self.finish()
:meth:`get_request_body` will deserialize the request data into
a dictionary based on the :http:header:`Content-Type` request
header. Similarly, :meth:`send_response` takes a dictionary,
serializes it based on the :http:header:`Accept` request header
and the application :class:`ContentSettings`, and writes it out,
using ``self.write()``.
"""
def initialize(self):
super(ContentMixin, self).initialize()
self._request_body = None
self._best_response_match = None
def get_response_content_type(self):
"""Figure out what content type will be used in the response."""
if self._best_response_match is None:
settings = ContentSettings.from_application(self.application)
acceptable = headers.parse_http_accept_header(
self.request.headers.get('Accept', '*/*'))
try:
selected, _ = algorithms.select_content_type(
acceptable, settings.available_content_types)
self._best_response_match = '/'.join(
[selected.content_type, selected.content_subtype])
except errors.NoMatch:
self._best_response_match = settings.default_content_type
return self._best_response_match
def get_request_body(self):
"""
Fetch (and cache) the request body as a dictionary.
:raise web.HTTPError: if the content type cannot be decoded.
The status code is set to 415 Unsupported Media Type
"""
if self._request_body is None:
settings = ContentSettings.from_application(self.application)
content_type_header = headers.parse_content_type(
self.request.headers.get('Content-Type',
settings.default_content_type))
content_type = '/'.join([content_type_header.content_type,
content_type_header.content_subtype])
try:
handler = settings[content_type]
self._request_body = handler.from_bytes(self.request.body)
except KeyError:
raise web.HTTPError(415, 'cannot decode body of type %s',
content_type)
return self._request_body
def send_response(self, body, set_content_type=True):
"""
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`
header be set? Defaults to :data:`True`
"""
settings = ContentSettings.from_application(self.application)
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.write(data_bytes)
class _BinaryContentHandler(object):
def __init__(self, content_type, pack, unpack):
self._pack = pack
self._unpack = unpack
self.content_type = content_type
def to_bytes(self, data_dict, encoding=None):
return self.content_type, self._pack(data_dict)
def from_bytes(self, data, encoding=None):
return self._unpack(data)
class _TextContentHandler(object):
def __init__(self, content_type, dumps, loads, default_encoding):
self._dumps = dumps
self._loads = loads
self.content_type = content_type
self.default_encoding = default_encoding
def to_bytes(self, data_dict, encoding=None):
selected = encoding or self.default_encoding
content_type = '{0}; charset="{1}"'.format(self.content_type, selected)
dumped = self._dumps(escape.recursive_unicode(data_dict))
return content_type, dumped.encode(selected)
def from_bytes(self, data, encoding=None):
return self._loads(data.decode(encoding or self.default_encoding))

8
sprockets/mixins/media_type/__init__.py
View file

@ -1,8 +0,0 @@
"""
Sprockets.Mixins.Media_type
===========================
"""
version_info = (0, 0, 0)
__version__ = '.'.join(str(v) for v in version_info)

5
test-requirements.txt
View file

@ -1,5 +0,0 @@
coverage>=3.7,<4
codecov
nose>=1.3,<2
mock
-r requirements.txt

61
tests.py
View file

@ -0,0 +1,61 @@
import json
from tornado import testing
import msgpack
import examples
class SendResponseTests(testing.AsyncHTTPTestCase):
def get_app(self):
return examples.make_application(debug=True)
def test_that_content_type_default_works(self):
response = self.fetch('/', method='POST', body='{}',
headers={'Content-Type': 'application/json'})
self.assertEqual(response.code, 200)
self.assertEqual(response.headers['Content-Type'],
'application/json; charset="utf-8"')
def test_that_missing_content_type_uses_default(self):
response = self.fetch('/', method='POST', body='{}',
headers={'Accept': 'application/xml',
'Content-Type': 'application/json'})
self.assertEqual(response.code, 200)
self.assertEqual(response.headers['Content-Type'],
'application/json; charset="utf-8"')
def test_that_accept_header_is_obeyed(self):
response = self.fetch('/', method='POST', body='{}',
headers={'Accept': 'application/msgpack',
'Content-Type': 'application/json'})
self.assertEqual(response.code, 200)
self.assertEqual(response.headers['Content-Type'],
'application/msgpack')
class GetRequestBodyTests(testing.AsyncHTTPTestCase):
def get_app(self):
return examples.make_application(debug=True)
def test_that_request_with_unhandled_type_results_in_415(self):
response = self.fetch(
'/', method='POST', headers={'Content-Type': 'application/xml'},
body=(u'<request><name>value</name>'
u'<embedded><utf8>\u2731</utf8></embedded>'
u'</request>').encode('utf-8'))
self.assertEqual(response.code, 415)
def test_that_msgpack_request_returns_default_type(self):
body = {
'name': 'value',
'embedded': {
'utf8': u'\u2731'
}
}
response = self.fetch('/', method='POST', body=msgpack.packb(body),
headers={'Content-Type': 'application/msgpack'})
self.assertEqual(response.code, 200)
self.assertEqual(json.loads(response.body.decode('utf-8')), body)

11
tox.ini Normal file
View file

@ -0,0 +1,11 @@
[tox]
envlist = py27,py34
indexserver =
default = https://pypi.python.org/simple
toxworkdir = build/tox
[testenv]
deps =
-rrequires/installation.txt
-rrequires/testing.txt
commands = nosetests []