[Repoze-dev] Repoze CVS: obob_helper.py root.py
Chris McDonough
chrism at agendaless.com
Thu Sep 13 05:47:55 UTC 2007
Update of /home/repoze/cvs/repoze.zope2/repoze/zope2
In directory laguna.palladion.com:/tmp/cvs-serv17421/repoze/zope2
Modified Files:
obob_helper.py
Added Files:
root.py
Log Message:
Implement replacement root object.
Factor out dummy modules into base.
More tests for obob_helper, and some refactoring.
--- NEW FILE: root.py ---
from App.ZApplication import connection_open_hooks
class Root:
""" A class which is much like App.ZApplication.ZApplicationWrapper,
but which:
- does not support ZODB versions (they are deprecated)
- does not allow for an alternate top-level object based on the 'name'
argument to __bobo_traverse__ (we ignore this argument).
- registers a Closer in the wsgi environment rather than in request._held
so we close the connection at the right time
"""
def __init__(self, db, name, txn):
# we pass txn in for testability
from OFS.Application import Application
conn = db.open()
root = conn.root()
if not root.has_key(name):
root[name] = Application()
txn.commit()
conn.close()
self._db = db
self._name = name
def __bobo_traverse__(self, REQUEST):
conn = self._db.open()
if connection_open_hooks:
for hook in connection_open_hooks:
hook(conn)
# close the connection when the txn ends
closer = Closer(conn)
cleanup = REQUEST.environ.setdefault('tm.cleanup', {})
cleanup['closeconn'] = closer
conn.setDebugInfo(REQUEST.environ, REQUEST.other)
return conn.root()[self._name]
def __call__(self):
conn = self._db.open()
# conn will go out of scope and thus be closed, we needn't close it
# manually
return conn.root()[self._name]
class Closer:
def __init__(self, jar):
self._jar = jar
def __del__(self):
try:
self._jar.close()
except:
import traceback
traceback.print_exc()
raise
Index: obob_helper.py
===================================================================
RCS file: /home/repoze/cvs/repoze.zope2/repoze/zope2/obob_helper.py,v
retrieving revision 1.4
retrieving revision 1.5
diff -u -d -r1.4 -r1.5
--- obob_helper.py 13 Sep 2007 00:58:05 -0000 1.4
+++ obob_helper.py 13 Sep 2007 05:47:52 -0000 1.5
@@ -1,74 +1,33 @@
+import transaction
+
from paste import httpexceptions
from zope.security.management import newInteraction
from zope.security.management import endInteraction
+from ZPublisher.BaseRequest import UNSPECIFIED_ROLES
+from ZPublisher.BaseRequest import quote
+from ZPublisher.BaseRequest import RequestContainer
from ZPublisher.HTTPResponse import HTTPResponse
from ZPublisher.HTTPRequest import HTTPRequest
from ZPublisher.Publish import call_object
from ZPublisher.Publish import missing_name
from ZPublisher.Publish import dont_publish_class
+from ZPublisher import xmlrpc
from ZPublisher.mapply import mapply
-db = None
-
-def parse_config_file(config):
- from Zope2.Startup import options, handlers
- opts = options.ZopeOptions()
- opts.configfile = config['zope.conf']
- opts.realize(doc="Sorry, no option docs yet.", raise_getopt_errs=0)
- handlers.handleConfig(opts.configroot, opts.confighandlers)
- return opts
-
-def configure(config):
- from Zope2 import Startup
- starter = Startup.get_starter()
- opts = parse_config_file(config)
- import App.config
- App.config.setConfiguration(opts.configroot)
- starter.setConfiguration(opts.configroot)
- starter.setupInitialLogging()
- #starter.setupLocale() # this is rude
- starter.setupSecurityOptions()
- starter.setupPublisher()
- starter.makeLockFile()
- starter.makePidFile()
- starter.startZope() # this sets up Globals
- starter.setupFinalLogging()
- import Globals
- return Globals.DB
-
-def get_connection(helper, configure_fn):
- # we pass in a configure function to make testing this function a
- # little easier
- global db
- if not db:
- db = configure_fn(helper.config)
- conn = db.open()
- closer = Closer(conn)
- environ = helper.environ
- cleanup = environ.setdefault('tm.cleanup', {})
- cleanup['closeconn'] = closer # close the connection when the txn ends
- conn.setDebugInfo(environ)
- return conn
-
-class Closer:
- def __init__(self, jar):
- self._jar = jar
+from AccessControl.ZopeSecurityPolicy import getRoles
- def __del__(self):
- try:
- self._jar.close()
- except:
- import traceback
- traceback.print_exc()
- raise
+from repoze.zope2.root import Root
def makeRequest(environ):
response = HTTPResponse()
stdin = environ['wsgi.input']
- request = HTTPRequest(stdin, environ, response)
+ # The clean = True argument is very important. If we don't pass it,
+ # our WSGI environment is copied and replaced, and thus we won't be
+ # able to mutate it successfully within the application.
+ request = HTTPRequest(stdin, environ, response, clean=True)
return request
class Zope2ObobHelper:
@@ -76,7 +35,8 @@
self.environ = environ
self.config = config
self.request = None
- self.browser_path = None
+ self.no_acquire_flag = False
+ self.root = None
def setup(self):
self.request = makeRequest(self.environ)
@@ -86,8 +46,10 @@
def teardown(self):
endInteraction()
- def path_elements(self, path):
+ def path_elements(self):
# remember path for later use
+ path = self.environ['PATH_INFO']
+ self.request['PARENTS'] = [self.root]
self.browser_path = path
# Clean up the path list
@@ -102,15 +64,84 @@
# Make sure that certain things that dont make sense
# cannot be traversed.
if item in ('REQUEST', 'aq_self', 'aq_base'):
- raise httpexceptions.HTTPNotFound(path)
+ # ZPublisher did NotFound but that's just wrong
+ raise httpexceptions.HTTPForbidden(path)
if not item or item=='.':
continue
elif item == '..':
del clean[-1]
else: clean.append(item)
+ self._before_traversal(clean, self.browser_path)
return clean
+ def _before_traversal(self, clean, browser_path):
+ # we only break this into a separate method for easier testing
+
+ request = self.request
+ request.path = browser_path
+
+ self.req_method = request.get('REQUEST_METHOD', 'GET').upper()
+ self.method = self.req_method
+ self.no_acquire_flag = False
+
+ if self.req_method in ('GET', 'POST'):
+ # ZPublisher checked the response to see if it was an XML-RPC
+ # response, but it was pointless.
+ if request.maybe_webdav_client:
+ self.no_acquire_flag = True
+ else:
+ # index_html is still the default method, only any object can
+ # override it by implementing its own __browser_default__ method
+ self.method = 'index_html'
+
+ URL = request['URL']
+ parents = request['PARENTS']
+ object = parents[-1]
+ del parents[:]
+
+ request.roles = getRoles(None, None, object, UNSPECIFIED_ROLES)
+
+ # if the top object has a __bobo_traverse__ method, then use it
+ # to possibly traverse to an alternate top-level object.
+ if hasattr(object,'__bobo_traverse__'):
+ object = object.__bobo_traverse__(request)
+ request.roles = getRoles(None, None, object, UNSPECIFIED_ROLES)
+
+ if not clean and not self.method:
+ raise httpexceptions.HTTPForbidden(request['URL'])
+
+ # Traverse the URL to find the object:
+ if hasattr(object, '__of__'):
+ # Try to bind the top-level object to the request
+ # This is how you get 'self.REQUEST'
+ object = object.__of__(RequestContainer(REQUEST=request))
+
+ parents.append(object)
+
+ steps = request.steps
+ request._steps = map(quote, steps)
+ path = clean[:]
+ path.reverse()
+
+ request['TraversalRequestNameStack'] = request.path = path
+ request['ACTUAL_URL'] = request['URL'] + quote(browser_path)
+
+ # Set the posttraverse for duration of the traversal here
+ request._post_traverse = post_traverse = []
+
+ def before_traverse(self, current, name):
+ # XXX implement
+ pass
+
+ def traverse(self, current, name):
+ # XXX implement
+ pass
+
+ def before_invoke(self, published):
+ # XXX implement
+ pass
+
def invoke(self, published):
return mapply(published,
positional = self.request.args,
@@ -122,27 +153,45 @@
context = self.request,
bind=1)
- def before_traverse(self, current, name):
+ def map_result(self, result):
# XXX implement
- pass
+ result = [str(result)]
+ return '200 OK', [('Content-Type', 'text/html')], result
- def traverse(self, current, name):
- # XXX implement
- pass
+ def make_root(self):
+ from Zope2.Startup import options, handlers, get_starter
+ import App.config
+ import Zope2
+ starter = get_starter()
+ opts = options.ZopeOptions()
+ opts.configfile = self.config['zope.conf']
+ opts.realize(args=[], doc="", raise_getopt_errs=0)
+ handlers.handleConfig(opts.configroot, opts.confighandlers)
+ App.config.setConfiguration(opts.configroot)
+ starter.setConfiguration(opts.configroot)
+ starter.setupInitialLogging()
+ #starter.setupLocale() # this is rude
+ starter.setupSecurityOptions()
+ starter.setupPublisher()
+ starter.makeLockFile()
+ starter.makePidFile()
+ starter.startZope() # this calls Zope2.App.startup.startup
+ starter.setupFinalLogging()
+ Zope2.zpublisher_transactions_manager = None # middleware does this
+ db = Zope2.DB
+ # replace the ZApplicationWrapper with a saner instance (the "Root")
+ root = Root(db, 'Application', transaction)
+ Zope2.bobo_application = root
+ self.root = root
+ return root
- def before_invoke(self, published):
- # XXX implement
- pass
+root = None
+
+def get_root(helper):
+ global root
+ if root is None:
+ # only once
+ root = helper.make_root()
+ return root
- def map_result(self, request, result):
- # XXX implement
- pass
-def get_root(helper, configure_fn=configure):
- # we allow the override of configure_fn here to make testing a bit
- # easier, not because it's used by the API
- conn = get_connection(helper, configure_fn)
- root = conn.root()
- app = root['Application']
- return app
-
_______________________________________________
Repoze-dev mailing list
Repoze-dev at lists.repoze.org
http://lists.repoze.org/mailman/listinfo/repoze-dev
More information about the Repoze-dev
mailing list