home *** CD-ROM | disk | FTP | other *** search
# Source Generated with Decompyle++ # File: in.pyc (Python 2.6) __version__ = '3.1.2' from urlparse import urljoin as _urljoin class _AttributeDocstrings(type): def __init__(cls, name, bases, dct): if not cls.__doc__: pass newdoc = [ ''] dctnames = dct.keys() dctnames.sort() for name in dctnames: if name.endswith('__doc'): if hasattr(cls, name): delattr(cls, name) val = []([ ' ' + line.strip() for line in dct[name].split('\n') ]) attrname = name[:-5] try: attrval = getattr(cls, attrname) except AttributeError: [] [] '\n'.join attrval = 'missing' except: [] newdoc.append('%s [= %r]:\n%s' % (attrname, attrval, val)) continue [] cls.__doc__ = '\n\n'.join(newdoc) from cherrypy._cperror import HTTPError, HTTPRedirect, InternalRedirect from cherrypy._cperror import NotFound, CherryPyException, TimeoutError from cherrypy import _cpdispatch as dispatch from cherrypy import _cptools tools = _cptools.default_toolbox Tool = _cptools.Tool from cherrypy import _cprequest from cherrypy.lib import http as _http from cherrypy import _cptree tree = _cptree.Tree() from cherrypy._cptree import Application from cherrypy import _cpwsgi as wsgi from cherrypy import process try: from cherrypy.process import win32 engine = win32.Win32Bus() engine.console_control_handler = win32.ConsoleCtrlHandler(engine) del win32 except ImportError: engine = process.bus class _TimeoutMonitor(process.plugins.Monitor): def __init__(self, bus): self.servings = [] process.plugins.Monitor.__init__(self, bus, self.run) def acquire(self): self.servings.append((serving.request, serving.response)) def release(self): try: self.servings.remove((serving.request, serving.response)) except ValueError: pass def run(self): for req, resp in self.servings: resp.check_timeout() engine.timeout_monitor = _TimeoutMonitor(engine) engine.timeout_monitor.subscribe() engine.autoreload = process.plugins.Autoreloader(engine) engine.autoreload.subscribe() engine.thread_manager = process.plugins.ThreadManager(engine) engine.thread_manager.subscribe() engine.signal_handler = process.plugins.SignalHandler(engine) from cherrypy import _cpserver server = _cpserver.Server() server.subscribe() def quickstart(root = None, script_name = '', config = None): if config: _global_conf_alias.update(config) if root is not None: tree.mount(root, script_name, config) if hasattr(engine, 'signal_handler'): engine.signal_handler.subscribe() if hasattr(engine, 'console_control_handler'): engine.console_control_handler.subscribe() engine.start() engine.block() try: from threading import local as _local except ImportError: from cherrypy._cpthreadinglocal import local as _local class _Serving(_local): __metaclass__ = _AttributeDocstrings request = _cprequest.Request(_http.Host('127.0.0.1', 80), _http.Host('127.0.0.1', 1111)) request__doc = '\n The request object for the current thread. In the main thread,\n and any threads which are not receiving HTTP requests, this is None.' response = _cprequest.Response() response__doc = '\n The response object for the current thread. In the main thread,\n and any threads which are not receiving HTTP requests, this is None.' def load(self, request, response): self.request = request self.response = response def clear(self): self.__dict__.clear() serving = _Serving() class _ThreadLocalProxy(object): __slots__ = [ '__attrname__', '__dict__'] def __init__(self, attrname): self.__attrname__ = attrname def __getattr__(self, name): child = getattr(serving, self.__attrname__) return getattr(child, name) def __setattr__(self, name, value): if name in ('__attrname__',): object.__setattr__(self, name, value) else: child = getattr(serving, self.__attrname__) setattr(child, name, value) def __delattr__(self, name): child = getattr(serving, self.__attrname__) delattr(child, name) def _get_dict(self): child = getattr(serving, self.__attrname__) d = child.__class__.__dict__.copy() d.update(child.__dict__) return d __dict__ = property(_get_dict) def __getitem__(self, key): child = getattr(serving, self.__attrname__) return child[key] def __setitem__(self, key, value): child = getattr(serving, self.__attrname__) child[key] = value def __delitem__(self, key): child = getattr(serving, self.__attrname__) del child[key] def __contains__(self, key): child = getattr(serving, self.__attrname__) return key in child def __len__(self): child = getattr(serving, self.__attrname__) return len(child) def __nonzero__(self): child = getattr(serving, self.__attrname__) return bool(child) request = _ThreadLocalProxy('request') response = _ThreadLocalProxy('response') class _ThreadData(_local): pass thread_data = _ThreadData() def _cherrypy_pydoc_resolve(thing, forceload = 0): if isinstance(thing, _ThreadLocalProxy): thing = getattr(serving, thing.__attrname__) return _pydoc._builtin_resolve(thing, forceload) try: import pydoc as _pydoc _pydoc._builtin_resolve = _pydoc.resolve _pydoc.resolve = _cherrypy_pydoc_resolve except ImportError: pass from cherrypy import _cplogging class _GlobalLogManager(_cplogging.LogManager): def __call__(self, *args, **kwargs): try: log = request.app.log except AttributeError: log = self return log.error(*args, **kwargs) def access(self): try: return request.app.log.access() except AttributeError: return _cplogging.LogManager.access(self) log = _GlobalLogManager() log.screen = True log.error_file = '' log.access_file = '' def _buslog(msg, level): log.error(msg, 'ENGINE', severity = level) engine.subscribe('log', _buslog) def expose(func = None, alias = None): def expose_(func): func.exposed = True if alias is not None: if isinstance(alias, basestring): parents[alias.replace('.', '_')] = func else: for a in alias: parents[a.replace('.', '_')] = func return func import sys import types if isinstance(func, (types.FunctionType, types.MethodType)): if alias is None: func.exposed = True return func parents = sys._getframe(1).f_locals return expose_(func) isinstance(func, (types.FunctionType, types.MethodType)) if func is None: if alias is None: parents = sys._getframe(1).f_locals return expose_ parents = sys._getframe(1).f_locals return expose_ func is None parents = sys._getframe(1).f_locals alias = func return expose_ def url(path = '', qs = '', script_name = None, base = None, relative = None): if qs: qs = '?' + qs if request.app: if not path.startswith('/'): pi = request.path_info if request.is_index is True: if not pi.endswith('/'): pi = pi + '/' elif request.is_index is False: if pi.endswith('/') and pi != '/': pi = pi[:-1] if path == '': path = pi else: path = _urljoin(pi, path) if script_name is None: script_name = request.script_name if base is None: base = request.base newurl = base + script_name + path + qs elif base is None: base = server.base() if not script_name: pass path = '' + path newurl = base + path + qs if './' in newurl: atoms = [] for atom in newurl.split('/'): if atom == '.': continue if atom == '..': atoms.pop() continue atoms.append(atom) newurl = '/'.join(atoms) if relative is None: relative = getattr(request.app, 'relative_urls', False) if relative == 'server': newurl = '/' + '/'.join(newurl.split('/', 3)[3:]) elif relative: old = url().split('/')[:-1] new = newurl.split('/') while old and new: a = old[0] b = new[0] if a != b: break old.pop(0) new.pop(0) new = [ '..'] * len(old) + new newurl = '/'.join(new) return newurl from cherrypy import _cpconfig config = _global_conf_alias = _cpconfig.Config() from cherrypy import _cpchecker checker = _cpchecker.Checker() engine.subscribe('start', checker) 
(.txt)