Thuban/Model/transientdb.py

# Copyright (C) 2003 by Intevation GmbH
# Authors:
# Bernhard Herzog <[email protected]>
#
# This program is free software under the GPL (>=v2)
# Read the file COPYING coming with the software for details.

"""Database for transient data

This database is intended for data representations needed during the
lifetime of a Thuban session but which is not permanent. Examples of
this are for instance a join of two DBF files where the DBF files are
the permanent representation of the data and the join only exists in the
Thuban session and is reconstructed when the session is opened.
"""

__version__ = "$Revision$"
# $Source$
# $Id$

import os
import weakref
from sqlite import connect

import table

sql_type_map = {
    table.FIELDTYPE_INT: "INTEGER",
    table.FIELDTYPE_STRING: "VARCHAR",
    table.FIELDTYPE_DOUBLE: "FLOAT",
    }

type_converter_map = {
    table.FIELDTYPE_INT: int,
    table.FIELDTYPE_STRING: str,
    table.FIELDTYPE_DOUBLE: float,
    }

class TransientDatabase:

    def __init__(self, filename):
        self.filename = filename
        self.conn = connect(filename)
        # Counters to produce unique table and column names
        self.num_tables = 0
        self.num_cols = 0
        # Since there's only once process using the SQLite database, we
        # might be able to get a tad more speed with default_synchronous
        # OFF. So far I haven't seen any measurable speedup, though.
        #self.execute("PRAGMA default_synchronous = OFF")

    def __del__(self):
        self.close()

    def close(self):
        if self.conn is not None:
            self.conn.close()
            self.conn = None

    def new_table_name(self):
        self.num_tables += 1
        return "Table%03d" % self.num_tables

    def new_column_name(self):
        self.num_cols += 1
        return "Col%03d" % self.num_cols

    def execute(self, *args):
        """execute the SQL statement in the database and return the result"""
        cursor = self.conn.cursor()
        cursor.execute(*args)
        result = cursor.fetchone()
        self.conn.commit()
        return result

    def cursor(self):
        return self.conn.cursor()


class ColumnReference:

    def __init__(self, name, type, internal_name):
        self.name = name
        self.type = type
        self.internal_name = internal_name


class TransientTableBase(table.OldTableInterfaceMixin):

    """Base class for tables in the transient database"""

    def __init__(self, transient_db):
        """Initialize the table for use with the given transient db"""
        self.db = transient_db
        self.tablename = self.db.new_table_name()
        self.indexed_columns = {}
        self.read_record_cursor = None
        self.read_record_last_row = None
        self.read_record_last_result = None

    def create(self, columns):
        self.columns = columns
        self.name_to_column = {}
        self.orig_names = []
        self.internal_to_orig = {}
        self.orig_to_internal = {}
        self.column_map = {}

        # Create the column objects and fill various maps and lists
        for index in range(len(self.columns)):
            col = self.columns[index]
            self.name_to_column[col.name] = col
            self.orig_names.append(col.name)
            self.internal_to_orig[col.internal_name] = col.name
            self.orig_to_internal[col.name] = col.internal_name
            self.column_map[col.name] = col
            self.column_map[index] = col

        # Build the CREATE TABLE statement and create the table in the
        # database
        table_types = ["id INTEGER PRIMARY KEY"]
        for col in self.columns:
            table_types.append("%s %s" % (col.internal_name,
                                          sql_type_map[col.type]))
        table_stmt = "CREATE TABLE %s (\n    %s\n);" % (self.tablename,
                                                   ",\n    ".join(table_types))
        self.db.execute(table_stmt)

    def transient_table(self):
        """
        Return a table whose underlying implementation is in the transient db
        """
        return self

    def ensure_index(self, column):
        """Ensure that there's an index on the given column"""
        if not column in self.indexed_columns:
            internal_name = self.orig_to_internal[column]
            indexname = "Index_%s_%s" % (self.tablename, internal_name)
            stmt = "CREATE INDEX %s ON %s (%s);" % (indexname, self.tablename,
                                                    internal_name)
            self.db.execute(stmt)
            self.indexed_columns[column] = 1

    def NumColumns(self):
        return len(self.columns)

    def NumRows(self):
        result = self.db.execute("SELECT count(*) FROM %s;" % self.tablename)
        return int(result[0])

    def Columns(self):
        return self.columns

    def Column(self, col):
        return self.column_map[col]

    def HasColumn(self, col):
        """Return whether the table has a column with the given name or index
        """
        return self.column_map.has_key(col)

    def ReadRowAsDict(self, index):
        if self.read_record_cursor is None or index <self.read_record_last_row:
            stmt = ("SELECT %s FROM %s;"
                    % (", ".join([c.internal_name for c in self.columns]),
                       self.tablename))
            self.read_record_cursor = self.db.cursor()
            self.read_record_cursor.execute(stmt)
            self.read_record_last_row = -1
            self.read_record_last_result = None

        # Now we should have a cursor at a position less than or equal
        # to the index so the following if statement will always set
        # result to a suitable value
        assert index >= self.read_record_last_row

        if index == self.read_record_last_row:
            result = self.read_record_last_result
        else:
            for i in range(index - self.read_record_last_row):
                result = self.read_record_cursor.fetchone()
                self.read_record_last_result = result
        self.read_record_last_row = index
        return dict(zip(self.orig_names, result))

    def ValueRange(self, col):
        col = self.column_map[col]
        iname = col.internal_name
        min, max = self.db.execute("SELECT min(%s), max(%s) FROM %s;"
                                   % (iname, iname, self.tablename))
        converter = type_converter_map[col.type]
        return (converter(min), converter(max))

    def UniqueValues(self, col):
        iname = self.column_map[col].internal_name
        cursor = self.db.cursor()
        cursor.execute("SELECT %s FROM %s GROUP BY %s;"
                       % (iname, self.tablename, iname))
        result = []
        while 1:
            row = cursor.fetchone()
            if row is None:
                break
            result.append(row[0])
        return result

    def SimpleQuery(self, left, comparison, right):
        """Return the indices of all rows that matching a condition.

        Parameters:
           left -- The column object for the left side of the comparison

           comparison -- The comparison operator as a string. It must be
                         one of '==', '!=', '<', '<=', '>=', '>'

           right -- The right hand side of the comparison. It must be
                    either a column object or a value, i.e. a string,
                    int or float.

        The return value is a sorted list of the indices of the rows
        where the condition is true.
        """
        if comparison not in ("==", "!=", "<", "<=", ">=", ">"):
            raise ValueError("Comparison operator %r not allowed" % comparison)

        if hasattr(right, "internal_name"):
            right_template = right.internal_name
            params = ()
        else:
            right_template = "%s"
            params = (right,)

        query = "SELECT id FROM %s WHERE %s %s %s ORDER BY id;" \
                % (self.tablename, left.internal_name, comparison,
                   right_template)

        cursor = self.db.cursor()
        cursor.execute(query, params)
        result = []
        while 1:
            row = cursor.fetchone()
            if row is None:
                break
            result.append(row[0])
        return result


class TransientTable(TransientTableBase):

    """A Table in a transient DB that starts as the copy of a Thuban Table."""

    def __init__(self, transient_db, table):
        """Create a new table in the given transient DB as a copy of table

        The table argument can be any object implementing the Table
        interface.
        """
        TransientTableBase.__init__(self, transient_db)
        self.create(table)

    def create(self, table):
        columns = []
        for col in table.Columns():
            columns.append(ColumnReference(col.name, col.type,
                                           self.db.new_column_name()))
        TransientTableBase.create(self, columns)

        # copy the input table to the transient db

        # A key to insert to use for the formatting of the insert
        # statement. The key must not be equal to any of the column
        # names so we construct one by building a string of x's that is
        # longer than any of the column names
        id_key = max([len(col.name) for col in self.columns]) * "x"

        insert_template = "INSERT INTO %s (id, %s) VALUES (%%(%s)s, %s);" \
                               % (self.tablename,
                                  ", ".join([col.internal_name
                                             for col in self.columns]),
                                  id_key,
                                  ", ".join(["%%(%s)s" % col.name
                                             for col in self.columns]))
        cursor = self.db.cursor()
        for i in range(table.NumRows()):
            row = table.ReadRowAsDict(i)
            row[id_key] = i
            cursor.execute(insert_template, row)
        self.db.conn.commit()


class TransientJoinedTable(TransientTableBase):

    """A Table in the transient DB that contains a join of two tables"""

    def __init__(self, transient_db, left_table, left_field,
                 right_table, right_field = None):
        """Create a new table in the transient DB as a join of two tables.

        Both input tables, left_table and right_table must have a
        transient_table method that returns a table object for a table
        in the trnsient database. The join is performed on the condition
        that the value of the left_field column the the left table is
        equal to the value of the right_field in the right_table.

        The joined table contains all columns of the input tables with
        one exception: Any column in the right_table with the same name
        as one of the columns in the left_table will be omitted. This is
        somewhat of an implementation detail, but is done so that the
        column names of the joined table can be the same as the column
        names of the input tables without having to create prefixes.
        """
        TransientTableBase.__init__(self, transient_db)
        self.left_table = left_table.transient_table()
        self.left_field = left_field
        self.right_table = right_table.transient_table()
        if right_field:
            self.right_field = right_field
        else:
            self.right_field = self.left_field
        self.create()

    def create(self):
        """Internal: Create the table with the joined data"""
        self.tablename = self.db.new_table_name()

        self.right_table.ensure_index(self.right_field)

        # Coalesce the column information
        visited = {}
        columns = []
        for col in self.left_table.columns + self.right_table.columns:
            if col.name in visited:
                # We can't allow multiple columns with the same original
                # name, so omit this one. FIXME: There should be a
                # better solution.
                continue
            columns.append(col)
        TransientTableBase.create(self, columns)

        # Copy the joined data to the table.
        internal_names = [col.internal_name for col in self.columns]
        stmt = ("INSERT INTO %s (id, %s) SELECT %s.id, %s FROM %s"
                " JOIN %s ON %s = %s;"
                % (self.tablename,
                   ", ".join(internal_names),
                   self.left_table.tablename,
                   ", ".join(internal_names),
                   self.left_table.tablename,
                   self.right_table.tablename,
                   self.orig_to_internal[self.left_field],
                   self.orig_to_internal[self.right_field]))
        self.db.execute(stmt)


class AutoTransientTable(table.OldTableInterfaceMixin):

    """Table that copies data to a transient table on demand.

    The AutoTransientTable takes another table as input and copies data
    to a table in a TransientDatabase instance on demand.
    """

    def __init__(self, transient_db, table):
        self.transient_db = transient_db
        self.table = table
        self.t_table = None

    def Columns(self):
        return self.table.Columns()

    def Column(self, col):
        return self.table.Column(col)

    def HasColumn(self, col):
        """Return whether the table has a column with the given name or index
        """
        return self.table.HasColumn(col)

    def NumRows(self):
        return self.table.NumRows()

    def NumColumns(self):
        return self.table.NumColumns()

    def ReadRowAsDict(self, record):
        """Return the record no. record as a dict mapping field names to values
        """
        if self.t_table is not None:
            return self.t_table.ReadRowAsDict(record)
        else:
            return self.table.ReadRowAsDict(record)

    def copy_to_transient(self):
        """Internal: Create a transient table and copy the data into it"""
        self.t_table = TransientTable(self.transient_db, self)

    def transient_table(self):
        """
        Return a table whose underlying implementation is in the transient db
        """
        if self.t_table is None:
            self.copy_to_transient()
        return self.t_table

    def ValueRange(self, col):
        if self.t_table is None:
            self.copy_to_transient()
        return self.t_table.ValueRange(col)

    def UniqueValues(self, col):
        if self.t_table is None:
            self.copy_to_transient()
        return self.t_table.UniqueValues(col)
1	# Copyright (C) 2003 by Intevation GmbH
2	# Authors:
3	# Bernhard Herzog <[email protected]>
4	#
5	# This program is free software under the GPL (>=v2)
6	# Read the file COPYING coming with the software for details.
7
8	"""Database for transient data
9
10	This database is intended for data representations needed during the
11	lifetime of a Thuban session but which is not permanent. Examples of
12	this are for instance a join of two DBF files where the DBF files are
13	the permanent representation of the data and the join only exists in the
14	Thuban session and is reconstructed when the session is opened.
15	"""
16
17	__version__ = "$Revision$"
18	# $Source$
19	# $Id$
20
21	import os
22	import weakref
23	from sqlite import connect
24
25	import table
26
27	sql_type_map = {
28	table.FIELDTYPE_INT: "INTEGER",
29	table.FIELDTYPE_STRING: "VARCHAR",
30	table.FIELDTYPE_DOUBLE: "FLOAT",
31	}
32
33	type_converter_map = {
34	table.FIELDTYPE_INT: int,
35	table.FIELDTYPE_STRING: str,
36	table.FIELDTYPE_DOUBLE: float,
37	}
38
39	class TransientDatabase:
40
41	def __init__(self, filename):
42	self.filename = filename
43	self.conn = connect(filename)
44	# Counters to produce unique table and column names
45	self.num_tables = 0
46	self.num_cols = 0
47	# Since there's only once process using the SQLite database, we
48	# might be able to get a tad more speed with default_synchronous
49	# OFF. So far I haven't seen any measurable speedup, though.
50	#self.execute("PRAGMA default_synchronous = OFF")
51
52	def __del__(self):
53	self.close()
54
55	def close(self):
56	if self.conn is not None:
57	self.conn.close()
58	self.conn = None
59
60	def new_table_name(self):
61	self.num_tables += 1
62	return "Table%03d" % self.num_tables
63
64	def new_column_name(self):
65	self.num_cols += 1
66	return "Col%03d" % self.num_cols
67
68	def execute(self, *args):
69	"""execute the SQL statement in the database and return the result"""
70	cursor = self.conn.cursor()
71	cursor.execute(*args)
72	result = cursor.fetchone()
73	self.conn.commit()
74	return result
75
76	def cursor(self):
77	return self.conn.cursor()
78
79
80	class ColumnReference:
81
82	def __init__(self, name, type, internal_name):
83	self.name = name
84	self.type = type
85	self.internal_name = internal_name
86
87
88	class TransientTableBase(table.OldTableInterfaceMixin):
89
90	"""Base class for tables in the transient database"""
91
92	def __init__(self, transient_db):
93	"""Initialize the table for use with the given transient db"""
94	self.db = transient_db
95	self.tablename = self.db.new_table_name()
96	self.indexed_columns = {}
97	self.read_record_cursor = None
98	self.read_record_last_row = None
99	self.read_record_last_result = None
100
101	def create(self, columns):
102	self.columns = columns
103	self.name_to_column = {}
104	self.orig_names = []
105	self.internal_to_orig = {}
106	self.orig_to_internal = {}
107	self.column_map = {}
108
109	# Create the column objects and fill various maps and lists
110	for index in range(len(self.columns)):
111	col = self.columns[index]
112	self.name_to_column[col.name] = col
113	self.orig_names.append(col.name)
114	self.internal_to_orig[col.internal_name] = col.name
115	self.orig_to_internal[col.name] = col.internal_name
116	self.column_map[col.name] = col
117	self.column_map[index] = col
118
119	# Build the CREATE TABLE statement and create the table in the
120	# database
121	table_types = ["id INTEGER PRIMARY KEY"]
122	for col in self.columns:
123	table_types.append("%s %s" % (col.internal_name,
124	sql_type_map[col.type]))
125	table_stmt = "CREATE TABLE %s (\n %s\n);" % (self.tablename,
126	",\n ".join(table_types))
127	self.db.execute(table_stmt)
128
129	def transient_table(self):
130	"""
131	Return a table whose underlying implementation is in the transient db
132	"""
133	return self
134
135	def ensure_index(self, column):
136	"""Ensure that there's an index on the given column"""
137	if not column in self.indexed_columns:
138	internal_name = self.orig_to_internal[column]
139	indexname = "Index_%s_%s" % (self.tablename, internal_name)
140	stmt = "CREATE INDEX %s ON %s (%s);" % (indexname, self.tablename,
141	internal_name)
142	self.db.execute(stmt)
143	self.indexed_columns[column] = 1
144
145	def NumColumns(self):
146	return len(self.columns)
147
148	def NumRows(self):
149	result = self.db.execute("SELECT count(*) FROM %s;" % self.tablename)
150	return int(result[0])
151
152	def Columns(self):
153	return self.columns
154
155	def Column(self, col):
156	return self.column_map[col]
157
158	def HasColumn(self, col):
159	"""Return whether the table has a column with the given name or index
160	"""
161	return self.column_map.has_key(col)
162
163	def ReadRowAsDict(self, index):
164	if self.read_record_cursor is None or index <self.read_record_last_row:
165	stmt = ("SELECT %s FROM %s;"
166	% (", ".join([c.internal_name for c in self.columns]),
167	self.tablename))
168	self.read_record_cursor = self.db.cursor()
169	self.read_record_cursor.execute(stmt)
170	self.read_record_last_row = -1
171	self.read_record_last_result = None
172
173	# Now we should have a cursor at a position less than or equal
174	# to the index so the following if statement will always set
175	# result to a suitable value
176	assert index >= self.read_record_last_row
177
178	if index == self.read_record_last_row:
179	result = self.read_record_last_result
180	else:
181	for i in range(index - self.read_record_last_row):
182	result = self.read_record_cursor.fetchone()
183	self.read_record_last_result = result
184	self.read_record_last_row = index
185	return dict(zip(self.orig_names, result))
186
187	def ValueRange(self, col):
188	col = self.column_map[col]
189	iname = col.internal_name
190	min, max = self.db.execute("SELECT min(%s), max(%s) FROM %s;"
191	% (iname, iname, self.tablename))
192	converter = type_converter_map[col.type]
193	return (converter(min), converter(max))
194
195	def UniqueValues(self, col):
196	iname = self.column_map[col].internal_name
197	cursor = self.db.cursor()
198	cursor.execute("SELECT %s FROM %s GROUP BY %s;"
199	% (iname, self.tablename, iname))
200	result = []
201	while 1:
202	row = cursor.fetchone()
203	if row is None:
204	break
205	result.append(row[0])
206	return result
207
208	def SimpleQuery(self, left, comparison, right):
209	"""Return the indices of all rows that matching a condition.
210
211	Parameters:
212	left -- The column object for the left side of the comparison
213
214	comparison -- The comparison operator as a string. It must be
215	one of '==', '!=', '<', '<=', '>=', '>'
216
217	right -- The right hand side of the comparison. It must be
218	either a column object or a value, i.e. a string,
219	int or float.
220
221	The return value is a sorted list of the indices of the rows
222	where the condition is true.
223	"""
224	if comparison not in ("==", "!=", "<", "<=", ">=", ">"):
225	raise ValueError("Comparison operator %r not allowed" % comparison)
226
227	if hasattr(right, "internal_name"):
228	right_template = right.internal_name
229	params = ()
230	else:
231	right_template = "%s"
232	params = (right,)
233
234	query = "SELECT id FROM %s WHERE %s %s %s ORDER BY id;" \
235	% (self.tablename, left.internal_name, comparison,
236	right_template)
237
238	cursor = self.db.cursor()
239	cursor.execute(query, params)
240	result = []
241	while 1:
242	row = cursor.fetchone()
243	if row is None:
244	break
245	result.append(row[0])
246	return result
247
248
249	class TransientTable(TransientTableBase):
250
251	"""A Table in a transient DB that starts as the copy of a Thuban Table."""
252
253	def __init__(self, transient_db, table):
254	"""Create a new table in the given transient DB as a copy of table
255
256	The table argument can be any object implementing the Table
257	interface.
258	"""
259	TransientTableBase.__init__(self, transient_db)
260	self.create(table)
261
262	def create(self, table):
263	columns = []
264	for col in table.Columns():
265	columns.append(ColumnReference(col.name, col.type,
266	self.db.new_column_name()))
267	TransientTableBase.create(self, columns)
268
269	# copy the input table to the transient db
270
271	# A key to insert to use for the formatting of the insert
272	# statement. The key must not be equal to any of the column
273	# names so we construct one by building a string of x's that is
274	# longer than any of the column names
275	id_key = max([len(col.name) for col in self.columns]) * "x"
276
277	insert_template = "INSERT INTO %s (id, %s) VALUES (%%(%s)s, %s);" \
278	% (self.tablename,
279	", ".join([col.internal_name
280	for col in self.columns]),
281	id_key,
282	", ".join(["%%(%s)s" % col.name
283	for col in self.columns]))
284	cursor = self.db.cursor()
285	for i in range(table.NumRows()):
286	row = table.ReadRowAsDict(i)
287	row[id_key] = i
288	cursor.execute(insert_template, row)
289	self.db.conn.commit()
290
291
292
293	class TransientJoinedTable(TransientTableBase):
294
295	"""A Table in the transient DB that contains a join of two tables"""
296
297	def __init__(self, transient_db, left_table, left_field,
298	right_table, right_field = None):
299	"""Create a new table in the transient DB as a join of two tables.
300
301	Both input tables, left_table and right_table must have a
302	transient_table method that returns a table object for a table
303	in the trnsient database. The join is performed on the condition
304	that the value of the left_field column the the left table is
305	equal to the value of the right_field in the right_table.
306
307	The joined table contains all columns of the input tables with
308	one exception: Any column in the right_table with the same name
309	as one of the columns in the left_table will be omitted. This is
310	somewhat of an implementation detail, but is done so that the
311	column names of the joined table can be the same as the column
312	names of the input tables without having to create prefixes.
313	"""
314	TransientTableBase.__init__(self, transient_db)
315	self.left_table = left_table.transient_table()
316	self.left_field = left_field
317	self.right_table = right_table.transient_table()
318	if right_field:
319	self.right_field = right_field
320	else:
321	self.right_field = self.left_field
322	self.create()
323
324	def create(self):
325	"""Internal: Create the table with the joined data"""
326	self.tablename = self.db.new_table_name()
327
328	self.right_table.ensure_index(self.right_field)
329
330	# Coalesce the column information
331	visited = {}
332	columns = []
333	for col in self.left_table.columns + self.right_table.columns:
334	if col.name in visited:
335	# We can't allow multiple columns with the same original
336	# name, so omit this one. FIXME: There should be a
337	# better solution.
338	continue
339	columns.append(col)
340	TransientTableBase.create(self, columns)
341
342	# Copy the joined data to the table.
343	internal_names = [col.internal_name for col in self.columns]
344	stmt = ("INSERT INTO %s (id, %s) SELECT %s.id, %s FROM %s"
345	" JOIN %s ON %s = %s;"
346	% (self.tablename,
347	", ".join(internal_names),
348	self.left_table.tablename,
349	", ".join(internal_names),
350	self.left_table.tablename,
351	self.right_table.tablename,
352	self.orig_to_internal[self.left_field],
353	self.orig_to_internal[self.right_field]))
354	self.db.execute(stmt)
355
356
357	class AutoTransientTable(table.OldTableInterfaceMixin):
358
359	"""Table that copies data to a transient table on demand.
360
361	The AutoTransientTable takes another table as input and copies data
362	to a table in a TransientDatabase instance on demand.
363	"""
364
365	def __init__(self, transient_db, table):
366	self.transient_db = transient_db
367	self.table = table
368	self.t_table = None
369
370	def Columns(self):
371	return self.table.Columns()
372
373	def Column(self, col):
374	return self.table.Column(col)
375
376	def HasColumn(self, col):
377	"""Return whether the table has a column with the given name or index
378	"""
379	return self.table.HasColumn(col)
380
381	def NumRows(self):
382	return self.table.NumRows()
383
384	def NumColumns(self):
385	return self.table.NumColumns()
386
387	def ReadRowAsDict(self, record):
388	"""Return the record no. record as a dict mapping field names to values
389	"""
390	if self.t_table is not None:
391	return self.t_table.ReadRowAsDict(record)
392	else:
393	return self.table.ReadRowAsDict(record)
394
395	def copy_to_transient(self):
396	"""Internal: Create a transient table and copy the data into it"""
397	self.t_table = TransientTable(self.transient_db, self)
398
399	def transient_table(self):
400	"""
401	Return a table whose underlying implementation is in the transient db
402	"""
403	if self.t_table is None:
404	self.copy_to_transient()
405	return self.t_table
406
407	def ValueRange(self, col):
408	if self.t_table is None:
409	self.copy_to_transient()
410	return self.t_table.ValueRange(col)
411
412	def UniqueValues(self, col):
413	if self.t_table is None:
414	self.copy_to_transient()
415	return self.t_table.UniqueValues(col)
Name	Value
svn:eol-style	native
svn:keywords	Author Date Id Revision