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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
|
# schema.py
# Copyright (C) 2005 Michael Bayer mike_mp@zzzcomputing.com
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
from sqlalchemy.util import *
import copy
__ALL__ = ['Table', 'Column', 'Sequence', 'ForeignKey',
'INT', 'CHAR', 'VARCHAR', 'TEXT', 'FLOAT', 'DECIMAL',
'TIMESTAMP', 'DATETIME', 'CLOB', 'BLOB', 'BOOLEAN', 'String', 'Integer', 'Numeric', 'DateTime', 'Binary',
]
class String:
def __init__(self, length):
self.length = length
class Integer:
"""integer datatype"""
pass
class Numeric:
def __init__(self, precision, length):
self.precision = precision
self.length = length
class DateTime:
pass
class Binary:pass
class Boolean:pass
class FLOAT(Numeric):pass
class TEXT(String): pass
class DECIMAL(Numeric):pass
class INT(Integer):pass
INTEGER = INT
class TIMESTAMP(DateTime): pass
class DATETIME(DateTime): pass
class CLOB(String): pass
class VARCHAR(String): pass
class CHAR(String):pass
class BLOB(Binary): pass
class BOOLEAN(Boolean): pass
class SchemaItem(object):
"""base class for items that define a database schema."""
def _init_items(self, *args):
for item in args:
if item is not None:
item._set_parent(self)
def accept_visitor(self, visitor):
raise NotImplementedError()
def _set_parent(self, parent):
"""a child item attaches itself to its parent via this method."""
raise NotImplementedError()
def hash_key(self):
"""returns a string that identifies this SchemaItem uniquely"""
return repr(self)
def __getattr__(self, key):
"""proxies method calls to an underlying implementation object for methods not found locally"""
return getattr(self._impl, key)
class TableSingleton(type):
def __call__(self, name, engine, *args, **kwargs):
try:
return engine.tables[name]
except:
table = type.__call__(self, name, engine, *args, **kwargs)
engine.tables[name] = table
# load column definitions from the database if 'autoload' is defined
# we do it after the table is in the singleton dictionary to support
# circular foreign keys
if kwargs.get('autoload', False):
engine.reflecttable(table)
return table
class Table(SchemaItem):
"""represents a relational database table."""
__metaclass__ = TableSingleton
def __init__(self, name, engine, *args, **kwargs):
self.name = name
self.columns = OrderedProperties()
self.c = self.columns
self.foreign_keys = OrderedProperties()
self.primary_keys = []
self.engine = engine
self._impl = self.engine.tableimpl(self)
self._init_items(*args)
def append_item(self, item):
self._init_items(item)
def _set_parent(self, schema):
schema.tables[self.name] = self
self.schema = schema
def accept_visitor(self, visitor):
for c in self.columns:
c.accept_visitor(visitor)
return visitor.visit_table(self)
class Column(SchemaItem):
"""represents a column in a database table."""
def __init__(self, name, type, key = None, primary_key = False, foreign_key = None, sequence = None, nullable = True):
self.name = name
self.type = type
self.sequence = sequence
self.foreign_key = foreign_key
self.key = key or name
self.primary_key = primary_key
if primary_key:
nullable = False
self.nullable = nullable
self._orig = None
original = property(lambda s: s._orig or s)
def _set_parent(self, table):
table.columns[self.key] = self
if self.primary_key:
table.primary_keys.append(self)
self.table = table
self.engine = table.engine
self._impl = self.engine.columnimpl(self)
if self.foreign_key is not None:
self._init_items(self.foreign_key)
table.foreign_keys[self.foreign_key.column.key] = self.foreign_key
def set_foreign_key(self, fk):
self.foreign_key = fk
self._init_items(self.foreign_key)
self.table.foreign_keys[self.foreign_key.column.key] = self.foreign_key
def _make_proxy(self, selectable, name = None):
"""creates a copy of this Column for use in a new selectable unit"""
# using copy.copy(c) seems to add a full second to the select.py unittest package
#c = copy.copy(self)
#if name is not None:
# c.name = name
# c.key = name
c = Column(name or self.name, self.type, key = name or self.key, primary_key = self.primary_key, foreign_key = self.foreign_key, sequence = self.sequence)
c.table = selectable
c.engine = self.engine
c._orig = self.original
selectable.columns[c.key] = c
c._impl = self.engine.columnimpl(c)
return c
def accept_visitor(self, visitor):
return visitor.visit_column(self)
def __lt__(self, other): return self._impl.__lt__(other)
def __le__(self, other): return self._impl.__le__(other)
def __eq__(self, other): return self._impl.__eq__(other)
def __ne__(self, other): return self._impl.__ne__(other)
def __gt__(self, other): return self._impl.__gt__(other)
def __ge__(self, other): return self._impl.__ge__(other)
def __str__(self): return self._impl.__str__()
class ForeignKey(SchemaItem):
def __init__(self, column):
self.column = column
def _set_parent(self, column):
self.parent = column
class Sequence(SchemaItem):
"""represents a sequence, which applies to Oracle and Postgres databases."""
def _set_parent(self, column, key):
self.column = column
def accept_visitor(self, visitor):
return visitor.visit_sequence(self)
class SchemaEngine(object):
"""a factory object used to create implementations for schema objects"""
def tableimpl(self, table):
raise NotImplementedError()
def columnimpl(self, column):
raise NotImplementedError()
def reflecttable(self, table):
raise NotImplementedError()
class SchemaVisitor(object):
"""base class for an object that traverses across Schema objects"""
def visit_schema(self, schema):pass
def visit_table(self, table):pass
def visit_column(self, column):pass
def visit_foreign_key(self, join):pass
def visit_index(self, index):pass
def visit_sequence(self, sequence):pass
|
