File Manager

Hackfut Security File Manager

Current Path: /opt/alt/python27/lib/python2.7/site-packages/paste/util

opt / alt / python27 / lib / python2.7 / site-packages / paste / util /

📁 ..
📄 PySourceColor.py(83.85 KB)
📄 PySourceColor.pyc(54.31 KB)
📄 PySourceColor.pyo(54.31 KB)
📄 UserDict24.py(5.39 KB)
📄 UserDict24.pyc(10.27 KB)
📄 UserDict24.pyo(10.27 KB)
📄 __init__.py(86 B)
📄 __init__.pyc(257 B)
📄 __init__.pyo(257 B)
📄 classinit.py(1.81 KB)
📄 classinit.pyc(1.89 KB)
📄 classinit.pyo(1.89 KB)
📄 classinstance.py(1.34 KB)
📄 classinstance.pyc(2.27 KB)
📄 classinstance.pyo(2.11 KB)
📄 converters.py(863 B)
📄 converters.pyc(1.15 KB)
📄 converters.pyo(1.15 KB)
📄 dateinterval.py(2.36 KB)
📄 dateinterval.pyc(3.08 KB)
📄 dateinterval.pyo(3.08 KB)
📄 datetimeutil.py(10.91 KB)
📄 datetimeutil.pyc(10.01 KB)
📄 datetimeutil.pyo(10.01 KB)
📄 doctest24.py(97.09 KB)
📄 doctest24.pyc(82.42 KB)
📄 doctest24.pyo(82.14 KB)
📄 filemixin.py(1.4 KB)
📄 filemixin.pyc(2.01 KB)
📄 filemixin.pyo(2.01 KB)
📄 finddata.py(3.75 KB)
📄 finddata.pyc(2.89 KB)
📄 finddata.pyo(2.89 KB)
📄 findpackage.py(786 B)
📄 findpackage.pyc(914 B)
📄 findpackage.pyo(914 B)
📄 import_string.py(3.04 KB)
📄 import_string.pyc(3.27 KB)
📄 import_string.pyo(3.27 KB)
📄 intset.py(18.68 KB)
📄 intset.pyc(19.07 KB)
📄 intset.pyo(19.07 KB)
📄 ip4.py(9.05 KB)
📄 ip4.pyc(9.02 KB)
📄 ip4.pyo(9.02 KB)
📄 killthread.py(1.2 KB)
📄 killthread.pyc(1.4 KB)
📄 killthread.pyo(1.4 KB)
📁 looper
📄 looper.py.tmpta(3.88 KB)
📄 mimeparse.py(6.44 KB)
📄 mimeparse.pyc(6.99 KB)
📄 mimeparse.pyo(6.99 KB)
📄 multidict.py(11.14 KB)
📄 multidict.pyc(16.33 KB)
📄 multidict.pyo(16.33 KB)
📄 quoting.py(2.42 KB)
📄 quoting.pyc(2.94 KB)
📄 quoting.pyo(2.94 KB)
📄 scgiserver.py(5.5 KB)
📄 scgiserver.pyc(5.98 KB)
📄 scgiserver.pyo(5.87 KB)
📄 string24.py(16.35 KB)
📄 string24.pyc(18.25 KB)
📄 string24.pyo(18.25 KB)
📁 template
📄 template.py.tmpta(23.73 KB)
📄 threadedprint.py(8.02 KB)
📄 threadedprint.pyc(10.15 KB)
📄 threadedprint.pyo(8.84 KB)
📄 threadinglocal.py(1.45 KB)
📄 threadinglocal.pyc(2.01 KB)
📄 threadinglocal.pyo(2.01 KB)

Editing: threadedprint.py

# (c) 2005 Ian Bicking and contributors; written for Paste (http://pythonpaste.org)
# Licensed under the MIT license: http://www.opensource.org/licenses/mit-license.php

"""
threadedprint.py
================

:author: Ian Bicking
:date: 12 Jul 2004

Multi-threaded printing; allows the output produced via print to be
separated according to the thread.

To use this, you must install the catcher, like::

threadedprint.install()

The installation optionally takes one of three parameters:

default
    The default destination for print statements (e.g., ``sys.stdout``).
factory
    A function that will produce the stream for a thread, given the
    thread's name.
paramwriter
    Instead of writing to a file-like stream, this function will be
    called like ``paramwriter(thread_name, text)`` for every write.

The thread name is the value returned by
``threading.currentThread().getName()``, a string (typically something
like Thread-N).

You can also submit file-like objects for specific threads, which will
override any of these parameters.  To do this, call ``register(stream,
[threadName])``.  ``threadName`` is optional, and if not provided the
stream will be registered for the current thread.

If no specific stream is registered for a thread, and no default has
been provided, then an error will occur when anything is written to
``sys.stdout`` (or printed).

Note: the stream's ``write`` method will be called in the thread the
text came from, so you should consider thread safety, especially if
multiple threads share the same writer.

Note: if you want access to the original standard out, use
``sys.__stdout__``.

You may also uninstall this, via::

threadedprint.uninstall()

TODO
----

* Something with ``sys.stderr``.
* Some default handlers.  Maybe something that hooks into `logging`.
* Possibly cache the results of ``factory`` calls.  This would be a
  semantic change.

"""

import threading
import sys
from paste.util import filemixin

class PrintCatcher(filemixin.FileMixin):

def __init__(self, default=None, factory=None, paramwriter=None,
                 leave_stdout=False):
        assert len(filter(lambda x: x is not None,
                          [default, factory, paramwriter])) <= 1, (
            "You can only provide one of default, factory, or paramwriter")
        if leave_stdout:
            assert not default, (
                "You cannot pass in both default (%r) and "
                "leave_stdout=True" % default)
            default = sys.stdout
        if default:
            self._defaultfunc = self._writedefault
        elif factory:
            self._defaultfunc = self._writefactory
        elif paramwriter:
            self._defaultfunc = self._writeparam
        else:
            self._defaultfunc = self._writeerror
        self._default = default
        self._factory = factory
        self._paramwriter = paramwriter
        self._catchers = {}

def write(self, v, currentThread=threading.currentThread):
        name = currentThread().getName()
        catchers = self._catchers
        if not catchers.has_key(name):
            self._defaultfunc(name, v)
        else:
            catcher = catchers[name]
            catcher.write(v)

def seek(self, *args):
        # Weird, but Google App Engine is seeking on stdout
        name = threading.currentThread().getName()
        catchers = self._catchers
        if not name in catchers:
            self._default.seek(*args)
        else:
            catchers[name].seek(*args)

def read(self, *args):
        name = threading.currentThread().getName()
        catchers = self._catchers
        if not name in catchers:
            self._default.read(*args)
        else:
            catchers[name].read(*args)

def _writedefault(self, name, v):
        self._default.write(v)

def _writefactory(self, name, v):
        self._factory(name).write(v)

def _writeparam(self, name, v):
        self._paramwriter(name, v)

def _writeerror(self, name, v):
        assert False, (
            "There is no PrintCatcher output stream for the thread %r"
            % name)

def register(self, catcher, name=None,
                 currentThread=threading.currentThread):
        if name is None:
            name = currentThread().getName()
        self._catchers[name] = catcher

def deregister(self, name=None,
                   currentThread=threading.currentThread):
        if name is None:
            name = currentThread().getName()
        assert self._catchers.has_key(name), (
            "There is no PrintCatcher catcher for the thread %r" % name)
        del self._catchers[name]

_printcatcher = None
_oldstdout = None

def install(**kw):
    global _printcatcher, _oldstdout, register, deregister
    if (not _printcatcher or sys.stdout is not _printcatcher):
        _oldstdout = sys.stdout
        _printcatcher = sys.stdout = PrintCatcher(**kw)
        register = _printcatcher.register
        deregister = _printcatcher.deregister

def uninstall():
    global _printcatcher, _oldstdout, register, deregister
    if _printcatcher:
        sys.stdout = _oldstdout
        _printcatcher = _oldstdout = None
        register = not_installed_error
        deregister = not_installed_error

def not_installed_error(*args, **kw):
    assert False, (
        "threadedprint has not yet been installed (call "
        "threadedprint.install())")

class StdinCatcher(filemixin.FileMixin):

def __init__(self, default=None, factory=None, paramwriter=None):
        assert len(filter(lambda x: x is not None,
                          [default, factory, paramwriter])) <= 1, (
            "You can only provide one of default, factory, or paramwriter")
        if default:
            self._defaultfunc = self._readdefault
        elif factory:
            self._defaultfunc = self._readfactory
        elif paramwriter:
            self._defaultfunc = self._readparam
        else:
            self._defaultfunc = self._readerror
        self._default = default
        self._factory = factory
        self._paramwriter = paramwriter
        self._catchers = {}

def read(self, size=None, currentThread=threading.currentThread):
        name = currentThread().getName()
        catchers = self._catchers
        if not catchers.has_key(name):
            return self._defaultfunc(name, size)
        else:
            catcher = catchers[name]
            return catcher.read(size)

def _readdefault(self, name, size):
        self._default.read(size)

def _readfactory(self, name, size):
        self._factory(name).read(size)

def _readparam(self, name, size):
        self._paramreader(name, size)

def _readerror(self, name, size):
        assert False, (
            "There is no StdinCatcher output stream for the thread %r"
            % name)

def deregister(self, catcher, name=None,
                   currentThread=threading.currentThread):
        if name is None:
            name = currentThread().getName()
        assert self._catchers.has_key(name), (
            "There is no StdinCatcher catcher for the thread %r" % name)
        del self._catchers[name]

_stdincatcher = None
_oldstdin = None

def install_stdin(**kw):
    global _stdincatcher, _oldstdin, register_stdin, deregister_stdin
    if not _stdincatcher:
        _oldstdin = sys.stdin
        _stdincatcher = sys.stdin = StdinCatcher(**kw)
        register_stdin = _stdincatcher.register
        deregister_stdin = _stdincatcher.deregister

def uninstall():
    global _stdincatcher, _oldstin, register_stdin, deregister_stdin
    if _stdincatcher:
        sys.stdin = _oldstdin
        _stdincatcher = _oldstdin = None
        register_stdin = deregister_stdin = not_installed_error_stdin

def not_installed_error_stdin(*args, **kw):
    assert False, (
        "threadedprint has not yet been installed for stdin (call "
        "threadedprint.install_stdin())")

Hackfut Security File Manager

Editing: threadedprint.py

Upload File

Create Folder