Mercurial > treepkg > treepkg
view treepkg/status.py @ 516:f2de1c162d30
read scm_type and rules_scm_type from config
author | Bjoern Ricks <bricks@intevation.de> |
---|---|
date | Wed, 10 Nov 2010 12:03:58 +0000 |
parents | 5fda6768bef6 |
children |
line wrap: on
line source
# Copyright (C) 2007, 2008, 2009 by Intevation GmbH # Authors: # Bernhard Herzog <bh@intevation.de> # # This program is free software under the GPL (>=v2) # Read the file COPYING coming with the software for details. import os import datetime import time import util # special object to indicate no default value nodefault = object() class FieldDesc(object): def __init__(self, default=nodefault): self.default = default def has_default(self): return self.default is not nodefault def set_default(self, value): self.default = value def serialize(self, value): raise NotImplementedError def deserialize(self, string): raise NotImplementedError class StringFieldDesc(FieldDesc): def serialize(self, value): return str(value) def deserialize(self, value): return value.strip() class DateFieldDesc(FieldDesc): date_format = "%Y-%m-%d %H:%M:%S" def serialize(self, value): return value.strftime(self.date_format) def deserialize(self, string): return datetime.datetime(*time.strptime(string.strip(), self.date_format)[:6]) class EnumValue(object): def __init__(self, name, description, finished=False, error=False): self.name = name self.description = description self.finished = finished self.error = error class EnumFieldDesc(FieldDesc): def __init__(self, *args, **kw): super(EnumFieldDesc, self).__init__(*args, **kw) self.values = {} def add(self, name, description, default=False, **kw): enum = EnumValue(name, description, **kw) self.values[enum.name] = enum if default: self.set_default(enum) def __iter__(self): return self.values.itervalues() def serialize(self, value): assert value.name is not None return value.name def deserialize(self, string): return self.values[string.strip()] def make_setter(fieldname, enum): def setter(self): setattr(self, fieldname, enum) setter.__name__ = enum.name return setter class StatusMetaClass(type): def __new__(cls, name, bases, clsdict): # Generate the _fields class variable from the field descriptors # in clsdict and remove the descriptors themselves. Also, add # one setter method for each enum. fields = dict() for key, value in clsdict.items(): if isinstance(value, FieldDesc): fields[key] = value del clsdict[key] if isinstance(value, EnumFieldDesc): for enum in value: clsdict[enum.name] = make_setter(key, enum) clsdict["_fields"] = fields return type.__new__(cls, name, bases, clsdict) class Status(object): __metaclass__ = StatusMetaClass # Overwrite in derived classes with a different magic string _magic = "Status 0.0\n" # Derived classes may extend a copy of this set with more instance # variables. _attrs = set(["_filename", "_values", "_after_setattr"]) def __init__(self, filename, after_setattr=None): self._values = dict() self._filename = filename self._after_setattr = after_setattr if self._filename is not None: assert os.path.isabs(self._filename) self.read() def deserialize(self, iter_lines, filename=None): magic = iter_lines.next() if magic != self._magic: message = ("Wrong magic: found %r but expected %r" % (magic, self._magic)) if filename is not None: message += " (read from %r)" % filename raise ValueError(message) values = dict() for line in iter_lines: field, value = line.split(":", 1) values[field] = self._fields[field].deserialize(value) self._values = values def serialize(self): lines = [self._magic] for field, desc in self._fields.items(): if field in self._values: lines.append("%s: %s\n" % (field, desc.serialize(self._values[field]))) return lines def read(self): if not os.path.exists(self._filename): return f = open(self._filename) try: self.deserialize(f, self._filename) finally: f.close() def write(self): if self._filename is not None: util.writefile(self._filename, "".join(self.serialize()), 0644) def __getattr__(self, attr): desc = self._fields.get(attr) if desc is not None: if attr in self._values: return self._values[attr] elif desc.has_default(): return desc.default raise AttributeError(attr) def __setattr__(self, attr, value): if attr in self._fields: self._values[attr] = value self.write() if self._after_setattr is not None: self._after_setattr(self, attr) elif attr in self._attrs: self.__dict__[attr] = value else: raise AttributeError(attr) class RevisionStatus(Status): _magic = "TreePackagerStatus 0.0\n" status = EnumFieldDesc() status.add("creating_source_package", "creating source package") status.add("source_package_created", "source package created") status.add("creating_binary_package", "building binary packages") status.add("binary_package_created", "build successful", finished=True) status.add("error", "error", error=True) status.add("unknown", "unknown", default=True) start = DateFieldDesc(default=None) stop = DateFieldDesc(default=None) tags = StringFieldDesc(default="") notification_mail = EnumFieldDesc() notification_mail.add("notification_sent", "notification mail has been sent", default=True) notification_mail.add("notification_pending", "notification mail still needs to be sent")