[Zope-Checkins] CVS: Zope/lib/python/Zope/App - ClassFactory.py:1.1.2.1 __init__.py:1.1.2.1 startup.py:1.1.2.1
Jim Fulton
jim@zope.com
Wed, 17 Jul 2002 17:15:06 -0400
Update of /cvs-repository/Zope/lib/python/Zope/App
In directory cvs.zope.org:/tmp/cvs-serv7961/lib/python/Zope/App
Added Files:
Tag: Zope-2_7-development-branch
ClassFactory.py __init__.py startup.py
Log Message:
Changed the Zope package to be a pure (well almost pure) container
package, in preparation for making some Zope 3 packages available in
Zope 2.7.
Now importing Zope has no side effects (other than adding an entry to
sys.modules).
To initialize the Zope application server, you need to call the
startup function in the Zope package:
import Zope # does nothing
Zope.startup() # initializes the application
Zope.startup populates the Zope package with the traditional
attributes, like DB and debug.
Note that calling startup additional times has no effect.
=== Added File Zope/lib/python/Zope/App/ClassFactory.py ===
##############################################################################
#
# Copyright (c) 2001 Zope Corporation 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
#
##############################################################################
"""Zope Framework Class Finder
"""
import OFS.Uninstalled
def ClassFactory(jar, module, name,
_silly=('__doc__',), _globals={},
):
try:
if module[:1]=='*':
# ZCLass! Yee ha!
return jar.root()['ZGlobals'][module]
else:
m=__import__(module, _globals, _globals, _silly)
return getattr(m, name)
except:
return OFS.Uninstalled.Broken(jar, None, (module, name))
=== Added File Zope/lib/python/Zope/App/__init__.py ===
##############################################################################
#
# Copyright (c) 2002 Zope Corporation 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.
#
##############################################################################
=== Added File Zope/lib/python/Zope/App/startup.py ===
##############################################################################
#
# Copyright (c) 2001 Zope Corporation 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
#
##############################################################################
"""Initialize the Zope Package and provide a published module
"""
import sys
import os
import imp
import Zope
import App.FindHomes
import ZODB
import ZODB.ZApplication
import ZODB.POSException
import Globals
import OFS.Application
import AccessControl.SecurityManagement
import AccessControl.User
import ZPublisher
import ExtensionClass
from zLOG import LOG, WARNING, INFO, BLATHER, log_time
def startup():
Globals.BobobaseName = os.path.join(Globals.data_dir, 'Data.fs')
Globals.DatabaseVersion='3'
# Import products
OFS.Application.import_products()
# Open the database
try:
# Try to use custom storage
m=imp.find_module('custom_zodb',[INSTANCE_HOME])
except:
import ZODB.FileStorage
DB=ZODB.FileStorage.FileStorage(Globals.BobobaseName)
DB=ZODB.DB(DB)
else:
m=imp.load_module('Zope.custom_zodb', m[0], m[1], m[2])
if hasattr(m,'DB'):
DB=m.DB
else:
DB=m.Storage
DB=ZODB.DB(DB)
Globals.BobobaseName = DB.getName()
sys.modules['Zope.custom_zodb']=m
if DB.getActivityMonitor() is None:
from ZODB.ActivityMonitor import ActivityMonitor
DB.setActivityMonitor(ActivityMonitor())
Globals.DB=DB # Ick, this is temporary until we come up with some registry
Zope.DB = DB
# Hook for providing multiple transaction object manager undo support:
Globals.UndoManager=DB
Globals.opened.append(DB)
import ClassFactory
DB.setClassFactory(ClassFactory.ClassFactory)
# "Log on" as system user
AccessControl.SecurityManagement.newSecurityManager(
None, AccessControl.User.system)
# Set up the "application" object that automagically opens
# connections
app = ZODB.ZApplication.ZApplicationWrapper(
DB, 'Application', OFS.Application.Application, (),
Globals.VersionNameName)
Zope.app = app
Zope.bobo_application = app
# Initialize products:
application = app()
OFS.Application.initialize(application)
if Globals.DevelopmentMode:
# Set up auto-refresh.
from App.RefreshFuncs import setupAutoRefresh
setupAutoRefresh(application._p_jar)
application._p_jar.close()
# "Log off" as system user
AccessControl.SecurityManagement.noSecurityManager()
global startup_time
startup_time = log_time()
Zope.debug = debug
Zope.zpublisher_transactions_manager = TransactionsManager()
Zope.zpublisher_validated_hook = (
AccessControl.SecurityManagement.newSecurityManager)
Zope.__bobo_before__ = AccessControl.SecurityManagement.noSecurityManager
def debug(*args, **kw):
return apply(ZPublisher.test,('Zope',)+args, kw)
class RequestContainer(ExtensionClass.Base):
def __init__(self,r): self.REQUEST=r
conflict_errors = 0
def zpublisher_exception_hook(
published, REQUEST, t, v, traceback,
# static
StringType=type(''),
ConflictError=ZODB.POSException.ConflictError,
ListType=type([]),
):
try:
if isinstance(t, StringType):
if t.lower() in ('unauthorized', 'redirect'):
raise
else:
if t is SystemExit:
raise
if issubclass(t, ConflictError):
# First, we need to close the current connection. We'll
# do this by releasing the hold on it. There should be
# some sane protocol for this, but for now we'll use
# brute force:
global conflict_errors
conflict_errors = conflict_errors + 1
method_name = REQUEST.get('PATH_INFO', '')
err = ('ZODB conflict error at %s '
'(%s conflicts since startup at %s)')
LOG(err % (method_name, conflict_errors, startup_time),
INFO, '')
LOG('Conflict traceback', BLATHER, '', error=sys.exc_info())
raise ZPublisher.Retry(t, v, traceback)
if t is ZPublisher.Retry: v.reraise()
if (getattr(REQUEST.get('RESPONSE', None), '_error_format', '')
!='text/html'): raise
if (published is None or published is app or
type(published) is ListType):
# At least get the top-level object
published=app.__bobo_traverse__(REQUEST).__of__(
RequestContainer(REQUEST))
get_transaction().begin() # Just to be sure.
published=getattr(published, 'im_self', published)
while 1:
f=getattr(published, 'raise_standardErrorMessage', None)
if f is None:
published=getattr(published, 'aq_parent', None)
if published is None: raise
else:
break
client=published
while 1:
if getattr(client, 'standard_error_message', None) is not None:
break
client=getattr(client, 'aq_parent', None)
if client is None: raise
if REQUEST.get('AUTHENTICATED_USER', None) is None:
REQUEST['AUTHENTICATED_USER']=AccessControl.User.nobody
f(client, REQUEST, t, v, traceback)
finally: traceback=None
class TransactionsManager:
def begin(self,
# Optimize global var lookups:
get_transaction=get_transaction):
get_transaction().begin()
def commit(self,
# Optimize global var lookups:
get_transaction=get_transaction):
get_transaction().commit()
def abort(self,
# Optimize global var lookups:
get_transaction=get_transaction):
get_transaction().abort()
def recordMetaData(self, object, request,
# Optimize global var lookups:
hasattr=hasattr, getattr=getattr,
get_transaction=get_transaction,
LOG=LOG, WARNING=WARNING,
):
request_get = request.get
if hasattr(object, 'getPhysicalPath'):
path = '/'.join(object.getPhysicalPath())
else:
# Try hard to get the physical path of the object,
# but there are many circumstances where that's not possible.
to_append = ()
if hasattr(object, 'im_self') and hasattr(object, '__name__'):
# object is a Python method.
to_append = (object.__name__,)
object = object.im_self
while object is not None and \
not hasattr(object, 'getPhysicalPath'):
if not hasattr(object, '__name__'):
object = None
break
to_append = (object.__name__,) + to_append
object = getattr(object, 'aq_inner', object)
object = getattr(object, 'aq_parent', None)
if object is not None:
path = '/'.join(object.getPhysicalPath() + to_append)
else:
# As Jim would say, "Waaaaaaaa!"
# This may cause problems with virtual hosts
# since the physical path is different from the path
# used to retrieve the object.
path = request_get('PATH_INFO')
T=get_transaction()
T.note(path)
auth_user=request_get('AUTHENTICATED_USER',None)
if auth_user is not None:
try:
auth_folder = auth_user.aq_parent
except AttributeError:
# Most likely some product forgot to call __of__()
# on the user object.
LOG('AccessControl', WARNING,
'A user object of type %s has no aq_parent.'
% str(type(auth_user)))
auth_path = request_get('AUTHENTICATION_PATH')
else:
auth_path = '/'.join(auth_folder.getPhysicalPath()[1:-1])
T.setUser(auth_user, auth_path)