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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
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)
|
