from peewee import * from playhouse.fields import CompressedField as _CompressedField from playhouse.hybrid import * from playhouse.sqlite_ext import SqliteExtDatabase import datetime from enum import Enum, unique from pkg_resources import resource_filename from .prefixes import query_pseudo_prefix from .peewee_ext import ClosureTable, EnumField db = SqliteExtDatabase('test.db', pragmas=[('foreign_keys', 'ON')]) db.load_extension(resource_filename(__name__, 'sqlext/closure')) __tables__ = [] __all__ = ['create_tables', 'drop_tables'] def table(cls): __tables__.append(cls) __all__.append(cls.__name__) return cls def create_tables(): db.create_tables(__tables__, True) def drop_tables(): db.drop_tables(__tables__, True) class BaseModel(Model): class Meta: database = db class CompressedField(_CompressedField): def db_value(self, value): return value if value is None else self.compress(value) def python_value(self, value): return value if value is None else self.decompress(value) @table class Document(BaseModel): @unique class Direction(Enum): IN = 0 OUT = 1 content = CompressedField() created = DateField(default=datetime.date.today) inserted = DateTimeField(default=datetime.datetime.now) description = CharField(null=True) original_path = CharField(null=True) file_type = CharField(null=True) direction = EnumField(Direction, null=True) @classmethod def matches(cls, prefix, value): return query_pseudo_prefix(prefix, value) or Document.id << ( DocumentTag.select(DocumentTag.document) .join(Tag, on=Tag.matches(prefix, value))) @table class Prefix(BaseModel): name = CharField(primary_key = True) virtual = BooleanField(default = False) description = CharField(null=True) def __str__ (self): return self.name def __repr__ (self): ext = ' (virt)' if self.virtual else '' return "<%s %s%s>" % (self.__class__.__name__, self.name, ext) @table class Tag(BaseModel): name = CharField() prefix = ForeignKeyField(Prefix, null=True, related_name = 'tag', db_column = 'prefix') description = CharField(null=True) @hybrid_method def matches(self, prefix, name): if isinstance(prefix, Prefix): prefix = prefix.name return (self.prefix_id == prefix) & (self.name == name) class Meta: indexes = [ (('name', 'prefix'), True) ] @property def prefixed_name(self): if self.prefix_id: return "%s:%s" % (self.prefix_id, self.name) else: return self.name def __str__(self): return self.prefixed_name def __repr__(self): return "<%s (#%d) %s>" % (self.__class__.__name__, self.id, self.prefixed_name) @table class DocumentTag(BaseModel): document = ForeignKeyField(Document, related_name = 'tags') tag = ForeignKeyField(Tag) class Meta: primary_key = CompositeKey('document', 'tag') @table class TagImplications(BaseModel): tag = ForeignKeyField(Tag, related_name = 'implications') implies_tag = ForeignKeyField(Tag, related_name = '_implied_by') class Meta: primary_key = CompositeKey('tag', 'implies_tag') def __repr__(self): return "<%s %d --> %d>" % (self.__class__.__name__, self.tag_id, self.implies_tag_id) TagClosure = ClosureTable(Tag, TagImplications, TagImplications.implies_tag, TagImplications.tag) table(TagClosure)