Well, thank the gods for unit testing. Here's the fah_data module with
fewer errors:
import copy, threading, observable
class FAHData(observable.Observable):
"""The data model for the [EMAIL PROTECTED] monitor."""
def __init__(self):
observable.Observable.__init__(self)
self.data = {}#this dict will hold all data
self.mutex = threading.RLock()
def get_all_data(self):
"""Returns a COPY of entire data dict."""
#not sure deepcopy() is really necessary here
#but using it for now
#might cause some weird synchronization problems...
try:
self.mutex.acquire()
return copy.deepcopy(self.data)
finally:
self.mutex.release()
def get_data(self, key):
"""Returns a COPY of <key> data element."""
try:
self.mutex.acquire()
return copy.deepcopy(self.data[key])
finally:
self.mutex.release()
#these three methods don't need a mutex because they are atomic (I
think):
#-------------------------------------------->
def set_value(self, key, value):
"""Sets value of <key> data element."""
self.data[key] = value
observable.Observable.notifyObservers(self, arg =
'ELEMENT_CHANGED')
def set_data(self, data):
"""Sets entire data dictionary."""
self.data = data
observable.Observable.notifyObservers(self, arg =
'DATA_CHANGED')
def clear_data(self):
"""Clears entire data dictionary."""
self.data = {}
observable.Observable.notifyObservers(self, arg =
'DATA_CHANGED')
#<---------------------------------------------
--
http://mail.python.org/mailman/listinfo/python-list
