python类proxy()的实例源码

def __init__(self, path):
        package_name = '_jinja2_module_templates_%x' % id(self)

        # create a fake module that looks for the templates in the
        # path given.
        mod = _TemplateModule(package_name)
        if isinstance(path, string_types):
            path = [path]
        else:
            path = list(path)
        mod.__path__ = path

        sys.modules[package_name] = weakref.proxy(mod,
            lambda x: sys.modules.pop(package_name, None))

        # the only strong reference, the sys.modules entry is weak
        # so that the garbage collector can remove it once the
        # loader that created it goes out of business.
        self.module = mod
        self.package_name = package_name

def __init__(self, path):
        package_name = '_jinja2_module_templates_%x' % id(self)

        # create a fake module that looks for the templates in the
        # path given.
        mod = _TemplateModule(package_name)
        if isinstance(path, string_types):
            path = [path]
        else:
            path = list(path)
        mod.__path__ = path

        sys.modules[package_name] = weakref.proxy(mod,
            lambda x: sys.modules.pop(package_name, None))

        # the only strong reference, the sys.modules entry is weak
        # so that the garbage collector can remove it once the
        # loader that created it goes out of business.
        self.module = mod
        self.package_name = package_name

def __init__(self, *args, **kwds):
        '''Initialize an ordered dictionary.  The signature is the same as
        regular dictionaries, but keyword arguments are not recommended because
        their insertion order is arbitrary.

        '''
        if len(args) > 1:
            raise TypeError('expected at most 1 arguments, got %d' % len(args))
        try:
            self.__root
        except AttributeError:
            self.__hardroot = _Link()
            self.__root = root = _proxy(self.__hardroot)
            root.prev = root.next = root
            self.__map = {}
        self.__update(*args, **kwds)

def test_proxy_ref(self):
        o = C()
        o.bar = 1
        ref1 = weakref.proxy(o, self.callback)
        ref2 = weakref.proxy(o, self.callback)
        del o
        gc_collect()

        def check(proxy):
            proxy.bar

        self.assertRaises(ReferenceError, check, ref1)
        self.assertRaises(ReferenceError, check, ref2)
        ref3 = weakref.proxy(C())
        gc_collect()
        self.assertRaises(ReferenceError, bool, ref3)
        self.assertEqual(self.cbcalled, 2)

def test_ref_reuse(self):
        o = C()
        ref1 = weakref.ref(o)
        # create a proxy to make sure that there's an intervening creation
        # between these two; it should make no difference
        proxy = weakref.proxy(o)
        ref2 = weakref.ref(o)
        self.assertTrue(ref1 is ref2,
                     "reference object w/out callback should be re-used")

        o = C()
        proxy = weakref.proxy(o)
        ref1 = weakref.ref(o)
        ref2 = weakref.ref(o)
        self.assertTrue(ref1 is ref2,
                     "reference object w/out callback should be re-used")
        self.assertTrue(weakref.getweakrefcount(o) == 2,
                     "wrong weak ref count for object")
        del proxy
        gc_collect()
        self.assertTrue(weakref.getweakrefcount(o) == 1,
                     "wrong weak ref count for object after deleting proxy")

def test_proxy_div(self):
        class C:
            def __floordiv__(self, other):
                return 42
            def __ifloordiv__(self, other):
                return 21
        o = C()
        p = weakref.proxy(o)
        self.assertEqual(p // 5, 42)
        p //= 5
        self.assertEqual(p, 21)

    # The PyWeakref_* C API is documented as allowing either NULL or
    # None as the value for the callback, where either means "no
    # callback".  The "no callback" ref and proxy objects are supposed
    # to be shared so long as they exist by all callers so long as
    # they are active.  In Python 2.3.3 and earlier, this guarantee
    # was not honored, and was broken in different ways for
    # PyWeakref_NewRef() and PyWeakref_NewProxy().  (Two tests.)

def test_callable_proxy(self):
        o = Callable()
        ref1 = weakref.proxy(o)

        self.check_proxy(o, ref1)

        self.assertTrue(type(ref1) is weakref.CallableProxyType,
                     "proxy is not of callable type")
        ref1('twinkies!')
        self.assertTrue(o.bar == 'twinkies!',
                     "call through proxy not passed through to original")
        ref1(x='Splat.')
        self.assertTrue(o.bar == 'Splat.',
                     "call through proxy not passed through to original")

        # expect due to too few args
        self.assertRaises(TypeError, ref1)

        # expect due to too many args
        self.assertRaises(TypeError, ref1, 1, 2, 3)

def check_proxy(self, o, proxy):
        o.foo = 1
        self.assertTrue(proxy.foo == 1,
                     "proxy does not reflect attribute addition")
        o.foo = 2
        self.assertTrue(proxy.foo == 2,
                     "proxy does not reflect attribute modification")
        del o.foo
        self.assertTrue(not hasattr(proxy, 'foo'),
                     "proxy does not reflect attribute removal")

        proxy.foo = 1
        self.assertTrue(o.foo == 1,
                     "object does not reflect attribute addition via proxy")
        proxy.foo = 2
        self.assertTrue(
            o.foo == 2,
            "object does not reflect attribute modification via proxy")
        del proxy.foo
        self.assertTrue(not hasattr(o, 'foo'),
                     "object does not reflect attribute removal via proxy")

def test_getweakrefcount(self):
        o = C()
        ref1 = weakref.ref(o)
        ref2 = weakref.ref(o, self.callback)
        self.assertTrue(weakref.getweakrefcount(o) == 2,
                     "got wrong number of weak reference objects")

        proxy1 = weakref.proxy(o)
        proxy2 = weakref.proxy(o, self.callback)
        self.assertTrue(weakref.getweakrefcount(o) == 4,
                     "got wrong number of weak reference objects")

        del ref1, ref2, proxy1, proxy2
        gc_collect()
        self.assertTrue(weakref.getweakrefcount(o) == 0,
                     "weak reference objects not unlinked from"
                     " referent when discarded.")

        # assumes ints do not support weakrefs
        self.assertTrue(weakref.getweakrefcount(1) == 0,
                     "got wrong number of weak reference objects for int")

def __init__(self, connection):
        from weakref import proxy

        self.connection = proxy(connection)
        self.description = None
        self.description_flags = None
        self.rowcount = -1
        self.arraysize = 1
        self._executed = None
        self.lastrowid = None
        self.messages = []
        self.errorhandler = connection.errorhandler
        self._result = None
        self._warnings = 0
        self._info = None
        self.rownumber = None

def __init__(self, path):
        package_name = '_jinja2_module_templates_%x' % id(self)

        # create a fake module that looks for the templates in the
        # path given.
        mod = _TemplateModule(package_name)
        if isinstance(path, string_types):
            path = [path]
        else:
            path = list(path)
        mod.__path__ = path

        sys.modules[package_name] = weakref.proxy(mod,
            lambda x: sys.modules.pop(package_name, None))

        # the only strong reference, the sys.modules entry is weak
        # so that the garbage collector can remove it once the
        # loader that created it goes out of business.
        self.module = mod
        self.package_name = package_name

def __init__(self, path):
        package_name = '_jinja2_module_templates_%x' % id(self)

        # create a fake module that looks for the templates in the
        # path given.
        mod = _TemplateModule(package_name)
        if isinstance(path, string_types):
            path = [path]
        else:
            path = list(path)
        mod.__path__ = path

        sys.modules[package_name] = weakref.proxy(mod,
            lambda x: sys.modules.pop(package_name, None))

        # the only strong reference, the sys.modules entry is weak
        # so that the garbage collector can remove it once the
        # loader that created it goes out of business.
        self.module = mod
        self.package_name = package_name

def __init__(self, parent_message):
    """Args:
      parent_message: The message whose _Modified() method we should call when
        we receive Modified() messages.
    """
    # This listener establishes a back reference from a child (contained) object
    # to its parent (containing) object.  We make this a weak reference to avoid
    # creating cyclic garbage when the client finishes with the 'parent' object
    # in the tree.
    if isinstance(parent_message, weakref.ProxyType):
      self._parent_message_weakref = parent_message
    else:
      self._parent_message_weakref = weakref.proxy(parent_message)

    # As an optimization, we also indicate directly on the listener whether
    # or not the parent message is dirty.  This way we can avoid traversing
    # up the tree in the common case.
    self.dirty = False

def __init__(self, path):
        package_name = '_jinja2_module_templates_%x' % id(self)

        # create a fake module that looks for the templates in the
        # path given.
        mod = _TemplateModule(package_name)
        if isinstance(path, string_types):
            path = [path]
        else:
            path = list(path)
        mod.__path__ = path

        sys.modules[package_name] = weakref.proxy(mod,
            lambda x: sys.modules.pop(package_name, None))

        # the only strong reference, the sys.modules entry is weak
        # so that the garbage collector can remove it once the
        # loader that created it goes out of business.
        self.module = mod
        self.package_name = package_name

def __init__(self, parent_message):
    """Args:
      parent_message: The message whose _Modified() method we should call when
        we receive Modified() messages.
    """
    # This listener establishes a back reference from a child (contained) object
    # to its parent (containing) object.  We make this a weak reference to avoid
    # creating cyclic garbage when the client finishes with the 'parent' object
    # in the tree.
    if isinstance(parent_message, weakref.ProxyType):
      self._parent_message_weakref = parent_message
    else:
      self._parent_message_weakref = weakref.proxy(parent_message)

    # As an optimization, we also indicate directly on the listener whether
    # or not the parent message is dirty.  This way we can avoid traversing
    # up the tree in the common case.
    self.dirty = False

def __init__(self, path):
        package_name = '_jinja2_module_templates_%x' % id(self)

        # create a fake module that looks for the templates in the
        # path given.
        mod = _TemplateModule(package_name)
        if isinstance(path, string_types):
            path = [path]
        else:
            path = list(path)
        mod.__path__ = path

        sys.modules[package_name] = weakref.proxy(mod,
            lambda x: sys.modules.pop(package_name, None))

        # the only strong reference, the sys.modules entry is weak
        # so that the garbage collector can remove it once the
        # loader that created it goes out of business.
        self.module = mod
        self.package_name = package_name

def __init__(self, ds, dataset_type='chombo_hdf5'):
        self.domain_left_edge = ds.domain_left_edge
        self.domain_right_edge = ds.domain_right_edge
        self.dataset_type = dataset_type
        self.field_indexes = {}
        self.dataset = weakref.proxy(ds)
        # for now, the index file is the dataset!
        self.index_filename = os.path.abspath(
            self.dataset.parameter_filename)
        self.directory = ds.fullpath
        self._handle = ds._handle

        self._levels = [
            key for key in self._handle.keys() if key.startswith('level')
        ]

        GridIndex.__init__(self, ds, dataset_type)

        self._read_particles()

def __init__(self, ds, dataset_type):
        ParallelAnalysisInterface.__init__(self)
        self.dataset = weakref.proxy(ds)
        self.ds = self.dataset

        self._initialize_state_variables()

        mylog.debug("Initializing data storage.")
        self._initialize_data_storage()

        mylog.debug("Setting up domain geometry.")
        self._setup_geometry()

        mylog.debug("Initializing data grid data IO")
        self._setup_data_io()

        # Note that this falls under the "geometry" object since it's
        # potentially quite expensive, and should be done with the indexing.
        mylog.debug("Detecting fields.")
        self._detect_output_fields()

def __init__(self, path):
        package_name = '_jinja2_module_templates_%x' % id(self)

        # create a fake module that looks for the templates in the
        # path given.
        mod = _TemplateModule(package_name)
        if isinstance(path, string_types):
            path = [path]
        else:
            path = list(path)
        mod.__path__ = path

        sys.modules[package_name] = weakref.proxy(mod,
            lambda x: sys.modules.pop(package_name, None))

        # the only strong reference, the sys.modules entry is weak
        # so that the garbage collector can remove it once the
        # loader that created it goes out of business.
        self.module = mod
        self.package_name = package_name

def __init__(self, connection):
        '''
        Do not create an instance of a Cursor yourself. Call
        connections.Connection.cursor().
        '''
        from weakref import proxy
        self.connection = proxy(connection)
        self.description = None
        self.rownumber = 0
        self.rowcount = -1
        self.arraysize = 1
        self._executed = None
        self.messages = []
        self.errorhandler = connection.errorhandler
        self._has_next = None
        self._rows = ()