/usr/lib/python2.7/dist-packages/koji/context.py is in koji-common 1.10.0-1.
This file is owned by root:root, with mode 0o755.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | #!/usr/bin/python
# Copyright (c) 2005-2014 Red Hat, Inc.
#
# Koji is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation;
# version 2.1 of the License.
#
# This software is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this software; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#
# Authors:
# Mike McLean <mikem@redhat.com>
# This modules provides a thread-safe way of passing
# request context around in a global way
# - db connections
# - request data
# - auth data
import thread
class _data(object):
pass
class ThreadLocal(object):
def __init__(self):
object.__setattr__(self, '_tdict', {})
# should probably be getattribute, but easier to debug this way
def __getattr__(self, key):
id = thread.get_ident()
tdict = object.__getattribute__(self, '_tdict')
if not tdict.has_key(id):
raise AttributeError(key)
data = tdict[id]
return object.__getattribute__(data, key)
def __setattr__(self, key, value):
id = thread.get_ident()
tdict = object.__getattribute__(self, '_tdict')
if not tdict.has_key(id):
tdict[id] = _data()
data = tdict[id]
return object.__setattr__(data,key,value)
def __delattr__(self, key):
id = thread.get_ident()
tdict = object.__getattribute__(self, '_tdict')
if not tdict.has_key(id):
raise AttributeError(key)
data = tdict[id]
ret = object.__delattr__(data, key)
if len(data.__dict__) == 0:
del tdict[id]
return ret
def __str__(self):
id = thread.get_ident()
tdict = object.__getattribute__(self, '_tdict')
return "(current thread: %s) {" % id + \
", ".join([ "%s : %s" %(k,v.__dict__) for (k,v) in tdict.iteritems() ]) + \
"}"
def _threadclear(self):
id = thread.get_ident()
tdict = object.__getattribute__(self, '_tdict')
if not tdict.has_key(id):
return
del tdict[id]
context = ThreadLocal()
if __name__ == '__main__':
#testing
#context.foo = 1
#context.bar = 2
print context
#del context.bar
print context
import random
import time
def test():
context.foo=random.random()
time.sleep(1.5+random.random())
context._threadclear()
print context
for x in xrange(1,10):
thread.start_new_thread(test,())
time.sleep(4)
print
print context
context.foo = 1
context.bar = 2
print context.foo,context.bar
print context
context._threadclear()
print context
|