""" open/dulcinea/lib/history.py """ from durus.persistent import PersistentObject from durus.utils import xrange from qp.lib.spec import add_getters, specify, require, string, proper, Mixin from qp.lib.spec import Specified, datetime_with_tz from qp.pub.user import User from qp.pub.common import get_users, site_now, get_hit, get_user class Event (Specified): """ Event objects should never be created outside of History.add_event(). """ timestamp_is = datetime_with_tz user_is = (User, None) event_code_is = string message_is = (string, None) def __init__(self, *args): if args: user, event_code, message = args specify(self, user=user, event_code=event_code, message=message, timestamp=site_now()) def __str__(self): return "%s %s" % (self.timestamp, self.event_code) def get_user(self): if self.user is None: return get_users().get('') return self.user add_getters(Event) class History (PersistentObject, Specified): events_is = [proper(Event)] def __init__(self): specify(self, events=[]) def __str__(self): return "history (%d events)" % len(self.events) def __len__(self): return len(self.events) def __getitem__(self, i): return self.events[i] def add_event(self, user, event_code, message): """(user : User, event_code : string, message : string) Create and append a new Event instance. """ if user is None: if get_hit(): user = get_user() else: user = get_users().get('') new_event = Event(user, event_code, message) self._p_note_change() self.events.append(new_event) def update_last_event(self, user, event_code, message): """(user : User, event_code : string, message : string) Create a new Event instance. If the last event on the list had the same event code and user, replace the last event with the new one. """ self._p_note_change() if self.events: if (self.events[-1].event_code == event_code and self.events[-1].user == user): self.events = self.events[:-1] self.events.append(Event(user, event_code, message)) def find_event(self, user=None, event_code=None, event_codes=None, latest=False, message=None): """(user : User = None, event_code : string = None, latest : bool = False, message : string = None) -> Event Find the earliest(or latest, if latest is true) event matching 'user' and/or 'event_code'. If either search criterion is not supplied or None, it does not affect the search, thus the default behaviour is to return the first event in the history list. Return None if no events match. """ if user is not None: require(user, User) if latest: indices = xrange(len(self.events)-1, -1, -1) else: indices = xrange(len(self.events)) for i in indices: event = self.events[i] if user and event.user is not user: continue if event_code is not None and event.event_code != event_code: continue if event_codes is not None and event.event_code not in event_codes: continue if message is not None and event.message != message: continue return event else: return None def find_events_after(self, time, event_code=None): return [event for event in self.events if event.get_timestamp() > time and ( event_code is None or event.get_event_code() == event_code)] def find_last_date(self, event_code): """(event_code:string) -> datetime | None""" event = self.find_event(event_code=event_code, latest=True) return event and event.get_timestamp() def __iter__(self): for event in self.events: yield event def gen_reversed(self): for event in reversed(self.events): yield event class Historical (Mixin): history_is = History def __init__(self): self.history = History() def get_history(self): return self.history def add_event(self, user, event_code, message): return self.get_history().add_event(user, event_code, message)