Project

General

Profile

1 13 aaronmk
# XML-database conversion
2
3 2418 aaronmk
import copy
4 16 aaronmk
import re
5 13 aaronmk
from xml.dom import Node
6
7 3718 aaronmk
import dicts
8 446 aaronmk
import exc
9 2436 aaronmk
import Parser
10 46 aaronmk
import sql
11 3077 aaronmk
import sql_io
12 2268 aaronmk
import sql_gen
13 84 aaronmk
import strings
14 133 aaronmk
import util
15 46 aaronmk
import xml_dom
16 2113 aaronmk
import xml_func
17 2436 aaronmk
import xpath
18 13 aaronmk
19 139 aaronmk
def name_of(node): return re.sub(r'^.*\.', r'', node.tagName)
20 16 aaronmk
21 13 aaronmk
ptr_suffix = '_id'
22
23 138 aaronmk
def is_ptr(node_name): return node_name.lower().endswith(ptr_suffix)
24 13 aaronmk
25 172 aaronmk
def ptr_type_guess(node_name):
26 13 aaronmk
    assert is_ptr(node_name)
27
    return node_name[:-len(ptr_suffix)]
28
29
def ptr_target(node):
30
    assert is_ptr(name_of(node))
31 2113 aaronmk
    return xml_dom.value_node(node)
32 13 aaronmk
33
def find_by_name(node, name):
34 46 aaronmk
    for parent in xml_dom.NodeParentIter(node):
35 13 aaronmk
        if name_of(parent) == name: return parent
36
        else:
37 46 aaronmk
            for child in xml_dom.NodeElemIter(parent):
38 16 aaronmk
                child_name = name_of(child)
39 13 aaronmk
                if is_ptr(child_name):
40 172 aaronmk
                    target = ptr_target(child)
41
                    if target.tagName == name: return target
42 13 aaronmk
                elif child_name == name: return child
43
    return None
44
45 2039 aaronmk
class ColRef:
46
    '''A reference to a table column'''
47
    def __init__(self, name, idx):
48
        self.name = name
49
        self.idx = idx
50
51
    def __str__(self): return self.name
52
53 3652 aaronmk
put_special_funcs = set(['_simplifyPath'])
54 2549 aaronmk
55 3660 aaronmk
def put(db, node, row_ins_ct_ref=None, on_error=exc.reraise,
56 3649 aaronmk
    col_defaults={}, in_table=None, parent_ids_loc=None, next=None):
57 3676 aaronmk
    '''
58
    @param node To use an entire XML document, pass root.firstChild.
59
    '''
60
    if node == None: return None # when no rows, root.firstChild == None
61
62 3685 aaronmk
    def put_(node):
63
        return put(db, node, row_ins_ct_ref, on_error, col_defaults, in_table,
64
            parent_ids_loc, next)
65 3621 aaronmk
66 3677 aaronmk
    def augment_error(e): exc.add_msg(e, 'node:\n'+str(node))
67 3646 aaronmk
    def on_error_(e):
68 3677 aaronmk
        augment_error(e)
69 3646 aaronmk
        on_error(e)
70
71 3719 aaronmk
    def wrap_e(e):
72
        augment_error(e)
73
        raise xml_func.SyntaxError(e)
74 3718 aaronmk
75 2551 aaronmk
    is_func = xml_func.is_func(node)
76 2005 aaronmk
    out_table = name_of(node)
77 2432 aaronmk
78
    # Divide children into fields and children with fkeys to parent
79 3718 aaronmk
    row = dicts.OnceOnlyDict()
80 2005 aaronmk
    children = []
81 3718 aaronmk
    try:
82
        for child in xml_dom.NodeElemIter(node):
83
            child_name = name_of(child)
84
            if xml_dom.is_empty(child): row[child_name] = None
85
            elif xml_dom.is_text(child):
86
                row[child_name] = strings.to_unicode(xml_dom.value(child))
87
            else:
88
                child_value = xml_dom.value_node(child)
89
                if (is_func or is_ptr(child_name)
90
                    or xml_func.is_func(child_value)):
91
                    row[child_name] = child_value
92
                else: children.append(child)
93
    except dicts.KeyExistsError, e: wrap_e(e)
94 2434 aaronmk
95
    # Special handling for structural XML functions
96
    if out_table == '_simplifyPath':
97 2436 aaronmk
        # Parse args
98 2434 aaronmk
        try:
99 3650 aaronmk
            next = row['next'] # modifies outer next var used by put_()
100 2434 aaronmk
            require = row['require']
101
            path = row['path']
102 2436 aaronmk
        except KeyError, e: wrap_e(e)
103
        try: next = xpath.parse(next)
104
        except Parser.SyntaxError, e: wrap_e(e)
105
        try: next = next[0].name
106
        except IndexError, e: wrap_e(e)
107 2434 aaronmk
108 3650 aaronmk
        return put_(path)
109 2434 aaronmk
110 3687 aaronmk
    is_literals = in_table == None
111 3624 aaronmk
    in_tables = []
112
    no_empty = set()
113 3687 aaronmk
    if not is_literals:
114 3624 aaronmk
        in_tables.append(in_table)
115
        no_empty.add(in_table)
116 2966 aaronmk
117 3107 aaronmk
    def pkey(table): return sql.pkey(db, table, True)
118
119 2177 aaronmk
    # Add fkey to parent
120
    if parent_ids_loc != None:
121 3601 aaronmk
        if sql_gen.is_table_col(parent_ids_loc):
122
            no_empty.add(parent_ids_loc.table)
123 2177 aaronmk
        parent_ptr = node.getAttribute('fkey')
124
        if parent_ptr == '': parent_ptr = pkey(name_of(node.parentNode))
125
        row[parent_ptr] = parent_ids_loc
126
127 2060 aaronmk
    # Divide fields into input columns and literal values
128 2506 aaronmk
    parent_ids_loc = None # applies to this section
129 3718 aaronmk
    row = row.inner # now allow keys to be overwritten
130 2060 aaronmk
    for out_col, value in row.iteritems():
131 2432 aaronmk
        # Handle forward pointers
132 3650 aaronmk
        if xml_dom.is_node(value): row[out_col] = value = put_(value)
133 2432 aaronmk
134
        # Translate values
135 2268 aaronmk
        if isinstance(value, sql_gen.Col): # value is temp table column
136 3687 aaronmk
            assert sql_gen.is_table_col(value)
137 2268 aaronmk
            in_tables.append(value.table)
138 3687 aaronmk
        elif (not is_literals and util.is_str(value)
139
            and value.startswith(input_col_prefix)): # value is input column
140 2272 aaronmk
            row[out_col] = sql_gen.Col(strings.remove_prefix(input_col_prefix,
141 2313 aaronmk
                value), in_table)
142 3428 aaronmk
        else: # value is literal value
143 2323 aaronmk
            row[out_col] = sql_gen.NamedCol(out_col, value)
144 2060 aaronmk
145
    # Insert node
146 3657 aaronmk
    try: pkeys_loc = sql_io.put_table(db, out_table, in_tables, row,
147
        row_ins_ct_ref, next, col_defaults, on_error_)
148
    except Exception, e:
149 3677 aaronmk
        augment_error(e)
150
        raise
151 2060 aaronmk
152 2975 aaronmk
    sql.empty_temp(db, set(in_tables) - no_empty)
153 2966 aaronmk
154 2177 aaronmk
    # Insert children with fkeys to parent
155 2506 aaronmk
    parent_ids_loc = pkeys_loc # applies to this section
156 3650 aaronmk
    for child in children: put_(child)
157 2177 aaronmk
158 2133 aaronmk
    return pkeys_loc
159 3651 aaronmk
160
def get(db, node, limit=None, start=None):
161
    def pkey(table): return sql.pkey(db, table)
162
163
    node = node.firstChild
164
    table = name_of(node)
165
    pkey_ = pkey(table)
166
167
    fields = []
168
    conds = {}
169
    for child in xml_dom.NodeElemIter(node):
170
        child_name = name_of(child)
171
        if xml_dom.is_empty(child): fields.append(child_name)
172
        elif xml_dom.is_text(child): conds[child_name] = xml_dom.value(child)
173
        else: raise Exception('Joins not supported yet')
174
    id_ = xml_dom.get_id(node)
175
    if id_ != None: conds[pkey(table)] = id_ # replace any existing pkey value
176
    if fields == []: fields.append(pkey_)
177
178
    return sql.select(db, table, fields, conds, limit, start)
179
180
# Controls when and how put_table() will partition the input table
181
partition_size = 500000 # rows
182
183
input_col_prefix = '$'
184
185
def put_table(db, node, in_table, in_row_ct_ref=None, row_ins_ct_ref=None,
186 3660 aaronmk
    limit=None, start=0, on_error=exc.reraise, col_defaults={}):
187 3651 aaronmk
    '''
188
    @param node The XML tree that transforms the input to the output. Similar to
189
        put()'s node param, but with the input column name prefixed by
190
        input_col_prefix in place of the column value.
191
    @return sql_gen.Col Where the pkeys (from INSERT RETURNING) are made
192
        available
193
    '''
194
    in_table = sql_gen.as_Table(in_table)
195
    in_table.set_srcs([in_table], overwrite=False)
196
    db.src = str(in_table)
197
198
    db.autoanalyze = True # but don't do this in row-based import
199
    db.autoexplain = True # but don't do this in row-based import
200
201
    # Subset and partition in_table
202
    # OK to do even if table already the right size because it takes <1 sec.
203
    full_in_table = in_table
204
    total = 0
205
    while limit == None or total < limit:
206
        # Adjust partition size if last partition
207
        this_limit = partition_size
208
        if limit != None: this_limit = min(this_limit, limit - total)
209
210
        # Row # is interally 0-based, but 1-based to the user
211
        db.log_debug('********** Partition: rows '+str(start+1)+'-'
212
            +str(start+this_limit)+' **********', level=1.2)
213
214
        # Subset in_table
215
        in_table = copy.copy(full_in_table) # don't modify input!
216
        in_table.name = str(in_table) # prepend schema
217
        cur = sql.run_query_into(db, sql.mk_select(db, full_in_table,
218
            limit=this_limit, start=start), into=in_table, add_pkey_=True)
219
            # full_in_table will be shadowed (hidden) by created temp table
220
221
        this_ct = cur.rowcount
222
        total += this_ct
223
        start += this_ct # advance start to fetch next set
224
        if this_ct == 0: break # in_table size is multiple of partition_size
225
226
        # Recurse
227
        pkeys_loc = put(db, node, row_ins_ct_ref, on_error,
228
            col_defaults, in_table)
229
        if in_row_ct_ref != None: in_row_ct_ref[0] += this_ct
230
231
        sql.empty_temp(db, in_table)
232
233
        if this_ct < partition_size: break # partial partition = last
234
235
        # Work around PostgreSQL's temp table disk space leak
236
        db.reconnect()
237
238
    return pkeys_loc