[Checkins] SVN: bozo.reuse/branches/dev/src/bozo/reuse/ Began implementing requests. Still need more tests.
Jim Fulton
jim at zope.com
Tue Jun 2 06:08:26 EDT 2009
Log message for revision 100580:
Began implementing requests. Still need more tests.
Implemented application, but still need tests.
Changed:
A bozo.reuse/branches/dev/src/bozo/reuse/__init__.py
A bozo.reuse/branches/dev/src/bozo/reuse/application.py
A bozo.reuse/branches/dev/src/bozo/reuse/doc/
A bozo.reuse/branches/dev/src/bozo/reuse/doc/request.test
A bozo.reuse/branches/dev/src/bozo/reuse/tests.py
-=-
Added: bozo.reuse/branches/dev/src/bozo/reuse/__init__.py
===================================================================
--- bozo.reuse/branches/dev/src/bozo/reuse/__init__.py (rev 0)
+++ bozo.reuse/branches/dev/src/bozo/reuse/__init__.py 2009-06-02 10:08:26 UTC (rev 100580)
@@ -0,0 +1 @@
+#
Property changes on: bozo.reuse/branches/dev/src/bozo/reuse/__init__.py
___________________________________________________________________
Added: svn:keywords
+ Id
Added: svn:eol-style
+ native
Added: bozo.reuse/branches/dev/src/bozo/reuse/application.py
===================================================================
--- bozo.reuse/branches/dev/src/bozo/reuse/application.py (rev 0)
+++ bozo.reuse/branches/dev/src/bozo/reuse/application.py 2009-06-02 10:08:26 UTC (rev 100580)
@@ -0,0 +1,238 @@
+##############################################################################
+#
+# Copyright Zope Foundation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+
+import bobo
+import zope.event
+import zope.i18n.interfaces
+import zope.publisher.browser
+import zope.publisher.http
+import webob
+import webob.multidict
+
+class BozoMultiDict:
+
+ def __init__(self, data):
+ self.data = data
+
+ def __getitem__(self, k):
+ v = self.data[k]
+ if isinstance(v, list):
+ return v[-1]
+ return v
+
+ def getall(self, name):
+ value = self.data.get(name)
+ if value is None:
+ return []
+ if isinstance(value, list):
+ return value
+ return [value]
+
+ def getone(self, name):
+ value = self.data.get(name)
+ if value is None:
+ raise KeyError(name)
+ if isinstance(value, list):
+ if len(value) != 1:
+ raise KeyError(name)
+ value = value[0]
+ return value
+
+ def mixed(self):
+ return self.data
+
+ def dict_of_lists(self):
+ return dict((k, self.getall(k)) for k in self.data)
+
+ def __contains__(self, k):
+ return k in self.data
+
+ def copy(self):
+ return self.__class__(self.data.copy())
+
+ def __len__(self):
+ return len(self.data)
+
+ def iteritems(self):
+ data = self.data
+ for k in sorted(data):
+ v = data[k]
+ if isinstance(v, list):
+ for x in v:
+ yield k, x
+ else:
+ yield k, v
+
+ def items(self):
+ return list(self.iteritems())
+
+ def iterkeys(self):
+ return (item[0] for item in self.iteritems())
+
+ __iter__ = iterkeys
+
+ def keys(self):
+ return list(self.iterkeys())
+
+ def itervalues(self):
+ return (item[1] for item in self.iteritems())
+
+ def values(self):
+ return list(self.itervalues())
+
+ def __repr__(self):
+ return "BozoMultiDict(%r)" % self.items()
+
+class Request(zope.publisher.browser.BrowserRequest):
+
+ __slots__ = ('_webob', '_form', '_charsets')
+
+ def __init__(self, *args):
+ zope.publisher.browser.BrowserRequest.__init__(self, *args)
+ self.processInputs()
+
+ @classmethod
+ def blank(class_, *args, **kw):
+ environ = webob.Request.blank(*args, **kw).environ
+ return class_(environ['wsgi.input'], environ)
+
+ @property
+ def charsets(self):
+ try:
+ return self._charsets
+ except AttributeError:
+ pass
+
+ adapter = zope.i18n.interfaces.IUserPreferredCharsets(self, None)
+ if adapter is None:
+ adapter = zope.publisher.http.HTTPCharsets(self)
+
+ self._charsets = adapter.getPreferredCharsets() or ['utf-8']
+
+ return self._charsets
+
+ @property
+ def charset(self):
+ return self.charsets[0]
+
+ @property
+ def environ(self):
+ return self._environ
+
+ @property
+ def webob(self):
+ try:
+ return self._webob
+ except AttributeError:
+ w = self._webob = webob.Request(self._environ)
+ if w.charset is None:
+ w.charset = self.charset
+ return w
+
+ @property
+ def headers(self):
+ return self.webob.headers
+
+ def __getattr__(self, name):
+ if name == '_webob':
+ raise AttributeError(name)
+ return getattr(self.webob, name)
+
+ def __setattr__(self, name, value):
+ try:
+ object.__setattr__(self, name, value)
+ except AttributeError:
+ if name == 'charsets':
+ if value is None:
+ return
+ raise
+ setattr(self.webob, name, value)
+
+ def __delattr__(self, name):
+ try:
+ object.__delattr__(self, name)
+ except AttributeError:
+ delattr(self.webob, name)
+
+ @property
+ def body_file(self):
+ return self.bodyStream
+
+ @property
+ def body(self):
+ stream = self.bodyStream
+ stream.seek(0)
+ return stream.read()
+
+ @property
+ def params(self):
+ return BozoMultiDict(self.form)
+
+ @property
+ def params(self):
+ return BozoMultiDict(self.form)
+
+ GET = params
+
+ @property
+ def POST(self):
+ if self.method in ('POST', 'PUT'):
+ content_type = self._environ.get('CONTENT_TYPE')
+ if content_type and (
+ content_type.startswith('application/x-www-form-urlencoded')
+ or
+ content_type.startswith('multipart/')
+ ):
+ return self.params
+ return webob.multidict.NoVars('Not a POST request')
+
+
+class Application(bobo.Application):
+
+ def __call__(self, environ, start_response):
+ request = Request(environ['wsgi.input'], environ)
+ zope.event.notify(BeginRequest(request))
+ try:
+ return self.bobo_response(
+ request, request.path_info, request.method
+ )(environ, start_response)
+ finally:
+ zope.event.notify(EndRequest(request))
+
+ def build_response(self, request, method, data):
+ content_type = data.content_type
+ response = request.response
+ if content_type:
+ response.setHeader('Content-Type', content_type)
+ for name, value in data.headers:
+ response.setHeader(name, value)
+ result = data.body
+ if result is not response:
+ response.setResult(result)
+
+ def wsgiresponse(environ, start_response):
+ start_response(response.getStatusString(), response.getHeaders())
+ return response.consumeBodyIter()
+
+ return wsgiresponse
+
+class RequestEvent:
+ def __init__(self, request):
+ self.request = request
+
+class BeginRequest(RequestEvent):
+ "A request has begun"
+
+class EndRequest(RequestEvent):
+ "A request has finished processing"
Property changes on: bozo.reuse/branches/dev/src/bozo/reuse/application.py
___________________________________________________________________
Added: svn:keywords
+ Id
Added: svn:eol-style
+ native
Added: bozo.reuse/branches/dev/src/bozo/reuse/doc/request.test
===================================================================
--- bozo.reuse/branches/dev/src/bozo/reuse/doc/request.test (rev 0)
+++ bozo.reuse/branches/dev/src/bozo/reuse/doc/request.test 2009-06-02 10:08:26 UTC (rev 100580)
@@ -0,0 +1,265 @@
+bozo.reuse.application.Request supports the WebOb requeest API
+==============================================================
+
+Sort of. There are some differences.
+
+- bozo requests support Zope form value marshalling
+- bozo requests make no distinction between POST body and query string
+ variables.
+- webob requests support mutation. You may be able to mutate webob
+ requests, but the effects of the mutation may not be visible.
+- a bozo request always has a charset and therefor request form data and
+ cookies are always unicode.
+
+
+The webob reference examples are reused below, with some modifications.
+
+
+ >>> from bozo.reuse.application import Request
+
+ >>> req = Request.blank('/article?id=1')
+ >>> from pprint import pprint
+ >>> pprint(dict(req.environ), width=1)
+ {'HTTP_HOST': 'localhost:80',
+ 'PATH_INFO': u'/article',
+ 'QUERY_STRING': 'id=1',
+ 'REQUEST_METHOD': 'GET',
+ 'SCRIPT_NAME': '',
+ 'SERVER_NAME': 'localhost',
+ 'SERVER_PORT': '80',
+ 'SERVER_PROTOCOL': 'HTTP/1.0',
+ 'wsgi.errors': <open file '<stderr>', mode 'w' at 0x3120c0>,
+ 'wsgi.input': <cStringIO.StringI object at 0x12f6410>,
+ 'wsgi.multiprocess': False,
+ 'wsgi.multithread': False,
+ 'wsgi.run_once': False,
+ 'wsgi.url_scheme': 'http',
+ 'wsgi.version': (1,
+ 0)}
+
+
+ >>> req.body
+ ''
+ >>> req.method
+ 'GET'
+ >>> req.scheme
+ 'http'
+ >>> req.script_name # The base of the URL
+ ''
+ >>> req.script_name = '/blog' # make it more interesting
+ >>> req.path_info # The yet-to-be-consumed part of the URL
+ u'/article'
+ >>> req.content_type # Content-Type of the request body
+ ''
+ >>> print req.remote_user # The authenticated user (there is none set)
+ None
+ >>> print req.remote_addr # The remote IP
+ None
+ >>> req.host
+ 'localhost:80'
+ >>> req.host_url
+ 'http://localhost'
+ >>> req.application_url
+ 'http://localhost/blog'
+ >>> req.path_url
+ 'http://localhost/blog/article'
+ >>> req.url
+ 'http://localhost/blog/article?id=1'
+ >>> req.path
+ '/blog/article'
+ >>> req.path_qs
+ '/blog/article?id=1'
+ >>> req.query_string
+ 'id=1'
+
+ >>> req.relative_url('archive')
+ 'http://localhost/blog/archive'
+
+ >>> req.path_info_peek() # Doesn't change request
+ u'article'
+ >>> req.path_info_pop() # Does change request!
+ u'article'
+ >>> req.script_name
+ u'/blog/article'
+ >>> req.path_info
+ ''
+
+ >>> from webob.compat import sorted
+ >>> req.headers['content-type'] = 'application/x-www-urlencoded'
+ >>> sorted(req.headers.items())
+ [('Content-Type', 'application/x-www-urlencoded'), ('Host', 'localhost:80')]
+ >>> req.environ['CONTENT_TYPE']
+ 'application/x-www-urlencoded'
+
+ >>> req = Request.blank('/test?check=a&check=b&name=Bob')
+ >>> req.GET
+ BozoMultiDict([(u'check', u'a'), (u'check', u'b'), (u'name', u'Bob')])
+ >>> req.GET['check']
+ u'b'
+ >>> req.GET.getall('check')
+ [u'a', u'b']
+ >>> req.GET.items()
+ [(u'check', u'a'), (u'check', u'b'), (u'name', u'Bob')]
+
+ >>> req.POST
+ <NoVars: Not a POST request>
+ >>> req.POST.items() # NoVars can be read like a dict, but not written
+ []
+
+ >>> from cStringIO import StringIO
+ >>> def post_environ(body, **kw):
+ ... return dict({'wsgi.input': StringIO(body)},
+ ... REQUEST_METHOD='POST',
+ ... CONTENT_TYPE='application/x-www-form-urlencoded',
+ ... CONTENT_LENGTH=len(body),
+ ... **kw)
+
+
+ >>> req = Request.blank('/test?check=a&check=b&name=Bob',
+ ... environ=post_environ('name=Joe&email=joe at example.com'))
+
+Note that, unlike webob, POST and GET returns all the form data,
+
+ >>> req.POST # doctest: +NORMALIZE_WHITESPACE
+ BozoMultiDict([(u'check', u'a'), (u'check', u'b'),
+ (u'email', u'joe at example.com'), (u'name', u'Joe'), (u'name', u'Bob')])
+
+ >>> req.POST['name']
+ u'Bob'
+
+ >>> req.params # doctest: +NORMALIZE_WHITESPACE
+ BozoMultiDict([(u'check', u'a'), (u'check', u'b'),
+ (u'email', u'joe at example.com'), (u'name', u'Joe'), (u'name', u'Bob')])
+
+ >>> req.params['name']
+ u'Bob'
+
+ >>> req.params.getall('name')
+ [u'Joe', u'Bob']
+ >>> for name, value in req.params.items():
+ ... print '%s: %r' % (name, value)
+ check: u'a'
+ check: u'b'
+ email: u'joe at example.com'
+ name: u'Joe'
+ name: u'Bob'
+
+ >>> req = Request.blank('/test?check=a&check=b&name=Bob',
+ ... environ=post_environ('var1=value1&var2=value2&rep=1&rep=2'))
+ >>> req.GET # doctest: +NORMALIZE_WHITESPACE
+ BozoMultiDict([(u'check', u'a'), (u'check', u'b'), (u'name', u'Bob'),
+ (u'rep', u'1'), (u'rep', u'2'), (u'var1', u'value1'),
+ (u'var2', u'value2')])
+
+ >>> req.POST # doctest: +NORMALIZE_WHITESPACE
+ BozoMultiDict([(u'check', u'a'), (u'check', u'b'), (u'name', u'Bob'),
+ (u'rep', u'1'), (u'rep', u'2'), (u'var1', u'value1'),
+ (u'var2', u'value2')])
+
+
+ >>> req = Request.blank('/test?check=a&check=b&name=Bob',
+ ... environ=post_environ('var1=value1&var2=value2&rep=1&rep=2',
+ ... HTTP_COOKIE='test=value'))
+
+
+ >>> req.headers['Cookie'] = 'test=value'
+ >>> req.cookies.items()
+ [(u'test', u'value')]
+
+ >>> 'text/html' in req.accept
+ True
+
+ >>> req.accept = 'text/html;q=0.5, application/xhtml+xml;q=1'
+ >>> req.accept
+ <MIMEAccept at 0x1342810 Accept: text/html;q=0.5, application/xhtml+xml>
+
+ >>> 'text/html' in req.accept
+ True
+
+ >>> req.accept.first_match(['text/html', 'application/xhtml+xml'])
+ 'text/html'
+
+ >>> req.accept.best_match(['text/html', 'application/xhtml+xml'])
+ 'application/xhtml+xml'
+
+ >>> req.accept.best_matches()
+ ['application/xhtml+xml', 'text/html']
+
+ >>> req.accept_language = 'es, pt-BR'
+ >>> req.accept_language.best_matches('en-US')
+ ['es', 'pt-BR', 'en-US']
+ >>> req.accept_language.best_matches('es')
+ ['es']
+
+ >>> server_token = 'opaque-token'
+ >>> server_token in req.if_none_match # You shouldn't return 304
+ False
+ >>> req.if_none_match = server_token
+ >>> req.if_none_match
+ <ETag opaque-token>
+ >>> server_token in req.if_none_match # You *should* return 304
+ True
+
+ >>> from webob import UTC
+ >>> from datetime import datetime
+ >>> req.if_modified_since = datetime(2006, 1, 1, 12, 0, tzinfo=UTC)
+ >>> req.headers['If-Modified-Since']
+ 'Sun, 01 Jan 2006 12:00:00 GMT'
+ >>> server_modified = datetime(2005, 1, 1, 12, 0, tzinfo=UTC)
+ >>> req.if_modified_since and req.if_modified_since >= server_modified
+ True
+
+ >>> req.if_range
+ <Empty If-Range>
+ >>> req.if_range.match(etag='some-etag', last_modified=datetime(2005, 1, 1, 12, 0))
+ True
+ >>> req.if_range = 'opaque-etag'
+ >>> req.if_range.match(etag='other-etag')
+ False
+ >>> req.if_range.match(etag='opaque-etag')
+ True
+
+ >>> from webob import Response
+ >>> res = Response(etag='opaque-etag')
+ >>> req.if_range.match_response(res)
+ True
+
+ >>> req.range = 'bytes=0-100'
+ >>> req.range
+ <Range ranges=(0, 99)>
+ >>> cr = req.range.content_range(length=1000)
+ >>> cr.start, cr.stop, cr.length
+ (0, 99, 1000)
+
+ >>> server_token in req.if_match # No If-Match means everything is ok
+ True
+ >>> req.if_match = server_token
+ >>> server_token in req.if_match # Still OK
+ True
+ >>> req.if_match = 'other-token'
+ >>> # Not OK, should return 412 Precondition Failed:
+ >>> server_token in req.if_match
+ False
+
+ >>> req = Request.blank('/')
+ >>> def wsgi_app(environ, start_response):
+ ... start_response('200 OK', [('Content-type', 'text/plain')])
+ ... return ['Hi!']
+ >>> req.call_application(wsgi_app)
+ ('200 OK', [('Content-type', 'text/plain')], ['Hi!'])
+
+ >>> res = req.get_response(wsgi_app)
+ >>> res
+ <Response at 0x13427d0 200 OK>
+
+ >>> res.status
+ '200 OK'
+ >>> res.headers
+ HeaderDict([('Content-type', 'text/plain')])
+ >>> res.body
+ 'Hi!'
+
+
+it also supports the zope.publisher browser requeest API
+========================================================
+
Property changes on: bozo.reuse/branches/dev/src/bozo/reuse/doc/request.test
___________________________________________________________________
Added: svn:eol-style
+ native
Added: bozo.reuse/branches/dev/src/bozo/reuse/tests.py
===================================================================
--- bozo.reuse/branches/dev/src/bozo/reuse/tests.py (rev 0)
+++ bozo.reuse/branches/dev/src/bozo/reuse/tests.py 2009-06-02 10:08:26 UTC (rev 100580)
@@ -0,0 +1,28 @@
+##############################################################################
+#
+# Copyright (c) Zope Foundation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE.
+#
+##############################################################################
+import re
+import unittest
+from zope.testing import doctest, renormalizing
+
+def test_suite():
+ return unittest.TestSuite((
+ doctest.DocFileSuite(
+ 'doc/request.test',
+ checker=renormalizing.RENormalizing([
+ (re.compile('0x[a-fA-F0-9]+'), 'ADDR'),
+ ]),
+ ),
+ ))
+
+
Property changes on: bozo.reuse/branches/dev/src/bozo/reuse/tests.py
___________________________________________________________________
Added: svn:keywords
+ Id
Added: svn:eol-style
+ native
More information about the Checkins
mailing list