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
|
from sqlalchemy import create_engine
from sqlalchemy import MetaData, Table, Column, ForeignKey, UniqueConstraint
from sqlalchemy import types as ty
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import as_declarative, declared_attr
from functools import partial
from . import Session
engine = create_engine('sqlite:///:memory:', echo=True)
Session.configure(bind=engine)
ReqColumn = partial(Column, nullable = False)
convention = {
'ix': "ix_%(column_0_label)s",
'uq': "uq_%(table_name)s_%(column_0_name)s",
'ck': "ck_%(table_name)s_%(column_0_name)s",
'fk': "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
'pk': "pk_%(table_name)s"
}
metadata = MetaData(naming_convention = convention)
try:
from sqlalchemy_repr import RepresentableBase as Base
except ImportError:
Base = object
class _QueryProperty(object):
def __get__(self, obj, type):
return Session().query(type)
@as_declarative(metadata = metadata)
class Model(Base):
id = Column(ty.Integer, primary_key=True)
@declared_attr
def __tablename__ (cls):
return cls.__name__.lower()
query = _QueryProperty()
@classmethod
def get_by(cls, *args, **kwargs):
return cls.query.filter_by(*args, **kwargs).first()
@classmethod
def get(cls, *args, **kwargs):
return cls.query.get(*args, **kwargs)
class Prefix(Model):
prefix = Column(ty.Unicode, index = True, unique = True)
builtin = ReqColumn(ty.Boolean, default = False)
pseudo = ReqColumn(ty.Boolean, default = False)
def __init__(self, name):
self.name = name
class Tag(Model):
name = ReqColumn(ty.Unicode)
prefix_id = Column(ty.Integer, ForeignKey(Prefix.id))
prefix = relationship('Prefix', backref='tag')
implications = relationship('Tag',
secondary = 'tag_x_tag',
primaryjoin = 'Tag.id == tag_x_tag.c.tag_id',
secondaryjoin = 'Tag.id == tag_x_tag.c.implies_tag_id',
backref='implied_by')
def __init__(self, name, prefix = None):
self.name = name
self.prefix = prefix
def implies(self, other):
self.implications.append(other)
def remove_implication(self, other):
self.implications.remove(other)
class Document(Model):
description = Column(ty.Unicode)
created = ReqColumn(ty.DateTime)
content = ReqColumn(ty.LargeBinary)
original_path = Column(ty.Unicode)
direction = Column(ty.Boolean)
tags = relationship('Tag', secondary = 'document_x_tag')
Table('document_x_tag', Model.metadata,
Column('document_id', ty.Integer, ForeignKey(Document.id)),
Column('tag_id', ty.Integer, ForeignKey(Tag.id)),
UniqueConstraint('document_id', 'tag_id', name = 'uq_document_x_tag')
)
Table('tag_x_tag', Model.metadata,
Column('tag_id', ty.Integer, ForeignKey(Tag.id)),
Column('implies_tag_id', ty.Integer, ForeignKey(Tag.id)),
UniqueConstraint('tag_id', 'implies_tag_id', name = 'uq_tag_x_tag')
)
|
