Project

General

Profile

1
# XML-database conversion
2

    
3
import copy
4
import re
5
from xml.dom import Node
6

    
7
import exc
8
import Parser
9
import sql
10
import sql_io
11
import sql_gen
12
import strings
13
import util
14
import xml_dom
15
import xml_func
16
import xpath
17

    
18
def name_of(node): return re.sub(r'^.*\.', r'', node.tagName)
19

    
20
ptr_suffix = '_id'
21

    
22
def is_ptr(node_name): return node_name.lower().endswith(ptr_suffix)
23

    
24
def ptr_type_guess(node_name):
25
    assert is_ptr(node_name)
26
    return node_name[:-len(ptr_suffix)]
27

    
28
def ptr_target(node):
29
    assert is_ptr(name_of(node))
30
    return xml_dom.value_node(node)
31

    
32
def find_by_name(node, name):
33
    for parent in xml_dom.NodeParentIter(node):
34
        if name_of(parent) == name: return parent
35
        else:
36
            for child in xml_dom.NodeElemIter(parent):
37
                child_name = name_of(child)
38
                if is_ptr(child_name):
39
                    target = ptr_target(child)
40
                    if target.tagName == name: return target
41
                elif child_name == name: return child
42
    return None
43

    
44
def get(db, node, limit=None, start=None):
45
    def pkey(table): return sql.pkey(db, table)
46
    
47
    node = node.firstChild
48
    table = name_of(node)
49
    pkey_ = pkey(table)
50
    
51
    fields = []
52
    conds = {}
53
    for child in xml_dom.NodeElemIter(node):
54
        child_name = name_of(child)
55
        if xml_dom.is_empty(child): fields.append(child_name)
56
        elif xml_dom.is_text(child): conds[child_name] = xml_dom.value(child)
57
        else: raise Exception('Joins not supported yet')
58
    id_ = xml_dom.get_id(node)
59
    if id_ != None: conds[pkey(table)] = id_ # replace any existing pkey value
60
    if fields == []: fields.append(pkey_)
61
    
62
    return sql.select(db, table, fields, conds, limit, start)
63

    
64
def put(db, node, row_ct_ref=None, on_error=exc.raise_, pool=None,
65
    store_ids=False, parent_id=None):
66
    '''store_ids enables searching the tree for missing fields'''
67
    def pkey(table): return sql.pkey(db, table, True)
68
    
69
    def put_(node, parent_id=None):
70
        args = (db, node, row_ct_ref, on_error, pool, store_ids, parent_id)
71
        if parent_id != None and pool != None: pool.apply_async(put, args)
72
        else: return put(*args)
73
    
74
    def on_error_(e):
75
        exc.add_msg(e, 'node:\n'+str(node))
76
        on_error(e)
77
    
78
    table = name_of(node)
79
    try: pkey_ = pkey(table)
80
    except sql.DatabaseErrors, e: on_error_(e); return None
81
    row = {}
82
    children = []
83
    
84
    # Divide children into fields and children with fkeys to parent
85
    for child in xml_dom.NodeElemIter(node):
86
        child_name = name_of(child)
87
        if xml_dom.is_empty(child): row[child_name] = None
88
        elif xml_dom.is_text(child):
89
            row[child_name] = strings.to_unicode(xml_dom.value(child))
90
        elif is_ptr(child_name): row[child_name] = put_(ptr_target(child))
91
        else: children.append(child)
92
    try: del row[pkey_]
93
    except KeyError: pass
94
    
95
    # Add fkey to parent
96
    if parent_id != None:
97
        parent_ptr = node.getAttribute('fkey')
98
        if parent_ptr == '': parent_ptr = pkey(name_of(node.parentNode))
99
        row[parent_ptr] = parent_id
100
    
101
    # Insert node
102
    try:
103
        for try_num in xrange(2):
104
            try:
105
                id_ = sql_io.put(db, table, row, pkey_, row_ct_ref)
106
                if store_ids: xml_dom.set_id(node, id_)
107
                break
108
            except sql.NullValueException, e:
109
                col = e.cols[0]
110
                if try_num > 0: raise # exception still raised after retry
111
                if store_ids and is_ptr(col):
112
                    # Search for required column in ancestors and their children
113
                    target = find_by_name(node, ptr_type_guess(col))
114
                    if target == None: raise
115
                    row[col] = xml_dom.get_id(target)
116
                else: raise
117
    except sql.DatabaseErrors, e: on_error_(e); return None
118
    
119
    # Insert children with fkeys to parent
120
    for child in children: put_(child, id_)
121
    
122
    return id_
123

    
124
class ColRef:
125
    '''A reference to a table column'''
126
    def __init__(self, name, idx):
127
        self.name = name
128
        self.idx = idx
129
    
130
    def __str__(self): return self.name
131

    
132
# Controls when and how put_table() will partition the input table
133
partition_size = 200000 # rows
134

    
135
input_col_prefix = '$'
136

    
137
put_table_special_funcs = set(['_simplifyPath'])
138

    
139
def put_table(db, node, in_table, in_row_ct_ref=None,
140
    row_ins_ct_ref=None, limit=None, start=0, on_error=exc.raise_,
141
    parent_ids_loc=None, next=None):
142
    '''
143
    @param node The XML tree that transforms the input to the output. Similar to
144
        put()'s node param, but with the input column name prefixed by
145
        input_col_prefix in place of the column value.
146
    @return (table, col) Where the pkeys (from INSERT RETURNING) are made
147
        available
148
    '''
149
    in_table = sql_gen.as_Table(in_table)
150
    in_table.set_srcs([in_table], overwrite=False)
151
    
152
    def put_table_(node, in_row_ct_ref=None):
153
        return put_table(db, node, in_table, in_row_ct_ref, row_ins_ct_ref,
154
            on_error=on_error, parent_ids_loc=parent_ids_loc, next=next)
155
    
156
    # Subset and/or partition in_table if needed
157
    in_row_ct = sql.table_row_count(db, in_table)
158
    if limit != None or start != 0 or in_row_ct > partition_size:
159
        end = in_row_ct
160
        if limit != None: end = min(start + limit, end)
161
        
162
        full_in_table = in_table
163
        
164
        for start_ in xrange(start, end, partition_size):
165
            limit_ = min(end - start_, partition_size)
166
            
167
            # Row # is interally 0-based, but 1-based to the user
168
            db.log_debug('********** Partition: rows '+str(start_+1)+'-'
169
                +str(start_+limit_)+' **********', level=1.2)
170
            
171
            # Subset in_table
172
            in_table = copy.copy(full_in_table) # don't modify input!
173
            sql.run_query_into(db, sql.mk_select(db, full_in_table,
174
                limit=limit_, start=start_), into=in_table)
175
                # full_in_table will be shadowed (hidden) by created temp table
176
            sql.add_pkey(db, in_table)
177
            
178
            pkeys_loc = put_table_(node, in_row_ct_ref)
179
        
180
        return pkeys_loc
181
    
182
    is_func = xml_func.is_func(node)
183
    out_table = name_of(node)
184
    
185
    # Divide children into fields and children with fkeys to parent
186
    row = {}
187
    children = []
188
    for child in xml_dom.NodeElemIter(node):
189
        child_name = name_of(child)
190
        if xml_dom.is_empty(child): row[child_name] = None
191
        elif xml_dom.is_text(child):
192
            row[child_name] = strings.to_unicode(xml_dom.value(child))
193
        else:
194
            child_value = xml_dom.value_node(child)
195
            if is_func or is_ptr(child_name) or xml_func.is_func(child_value):
196
                row[child_name] = child_value
197
            else: children.append(child)
198
    
199
    # Special handling for structural XML functions
200
    if out_table == '_simplifyPath':
201
        # Parse args
202
        def wrap_e(e): raise xml_func.SyntaxError(e)
203
        try:
204
            next = row['next'] # modifies outer next var used by put_table_()
205
            require = row['require']
206
            path = row['path']
207
        except KeyError, e: wrap_e(e)
208
        try: next = xpath.parse(next)
209
        except Parser.SyntaxError, e: wrap_e(e)
210
        try: next = next[0].name
211
        except IndexError, e: wrap_e(e)
212
        
213
        return put_table_(path)
214
    
215
    no_empty = set([in_table])
216
    
217
    def pkey(table): return sql.pkey(db, table, True)
218
    
219
    # Remove any explicit pkey
220
    try: del row[pkey(out_table)]
221
    except KeyError: pass
222
    
223
    # Add fkey to parent
224
    if parent_ids_loc != None:
225
        no_empty.add(parent_ids_loc.table)
226
        parent_ptr = node.getAttribute('fkey')
227
        if parent_ptr == '': parent_ptr = pkey(name_of(node.parentNode))
228
        row[parent_ptr] = parent_ids_loc
229
    
230
    # Divide fields into input columns and literal values
231
    parent_ids_loc = None # applies to this section
232
    in_tables = [in_table]
233
    for out_col, value in row.iteritems():
234
        # Handle forward pointers
235
        if xml_dom.is_node(value): row[out_col] = value = put_table_(value)
236
        
237
        # Translate values
238
        if isinstance(value, sql_gen.Col): # value is temp table column
239
            in_tables.append(value.table)
240
        elif util.is_str(value) and value.startswith(input_col_prefix):
241
            # value is input column
242
            row[out_col] = sql_gen.Col(strings.remove_prefix(input_col_prefix,
243
                value), in_table)
244
        else: # value is literal value; should only be string or None
245
            assert util.is_str(value) or value == None
246
            row[out_col] = sql_gen.NamedCol(out_col, value)
247
    
248
    # Insert node
249
    pkeys_loc = sql_io.put_table(db, out_table, in_tables, row, row_ins_ct_ref,
250
        None, next, is_func, on_error)
251
    
252
    sql.empty_temp(db, set(in_tables) - no_empty)
253
    
254
    # Insert children with fkeys to parent
255
    parent_ids_loc = pkeys_loc # applies to this section
256
    for child in children: put_table_(child)
257
    
258
    # Count # rows and update in_row_ct_ref once all columns have been processed
259
    if in_row_ct_ref != None:
260
        in_row_ct_ref[0] += sql.table_row_count(db, in_table)
261
    
262
    return pkeys_loc
(9-9/37)