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 = []
        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


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
        insert_template = "INSERT INTO %s (%s) VALUES (%s);" \
                               % (self.tablename,
                                  ", ".join([col.internal_name
                                             for col in self.columns]),
                                  ", ".join(["%%(%s)s" % col.name
                                             for col in self.columns]))
        cursor = self.db.cursor()
        for i in range(table.NumRows()):
            cursor.execute(insert_template, table.ReadRowAsDict(i))
        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:
                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 (%s) SELECT %s FROM %s JOIN %s ON %s = %s;" \
               % (self.tablename,
                  ", ".join(internal_names),
                  ", ".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 = []
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
209	class TransientTable(TransientTableBase):
210
211	"""A Table in a transient DB that starts as the copy of a Thuban Table."""
212
213	def __init__(self, transient_db, table):
214	"""Create a new table in the given transient DB as a copy of table
215
216	The table argument can be any object implementing the Table
217	interface.
218	"""
219	TransientTableBase.__init__(self, transient_db)
220	self.create(table)
221
222	def create(self, table):
223	columns = []
224	for col in table.Columns():
225	columns.append(ColumnReference(col.name, col.type,
226	self.db.new_column_name()))
227	TransientTableBase.create(self, columns)
228
229	# copy the input table to the transient db
230	insert_template = "INSERT INTO %s (%s) VALUES (%s);" \
231	% (self.tablename,
232	", ".join([col.internal_name
233	for col in self.columns]),
234	", ".join(["%%(%s)s" % col.name
235	for col in self.columns]))
236	cursor = self.db.cursor()
237	for i in range(table.NumRows()):
238	cursor.execute(insert_template, table.ReadRowAsDict(i))
239	self.db.conn.commit()
240
241
242
243	class TransientJoinedTable(TransientTableBase):
244
245	"""A Table in the transient DB that contains a join of two tables"""
246
247	def __init__(self, transient_db, left_table, left_field,
248	right_table, right_field = None):
249	"""Create a new table in the transient DB as a join of two tables.
250
251	Both input tables, left_table and right_table must have a
252	transient_table method that returns a table object for a table
253	in the trnsient database. The join is performed on the condition
254	that the value of the left_field column the the left table is
255	equal to the value of the right_field in the right_table.
256
257	The joined table contains all columns of the input tables with
258	one exception: Any column in the right_table with the same name
259	as one of the columns in the left_table will be omitted. This is
260	somewhat of an implementation detail, but is done so that the
261	column names of the joined table can be the same as the column
262	names of the input tables without having to create prefixes.
263	"""
264	TransientTableBase.__init__(self, transient_db)
265	self.left_table = left_table.transient_table()
266	self.left_field = left_field
267	self.right_table = right_table.transient_table()
268	if right_field:
269	self.right_field = right_field
270	else:
271	self.right_field = self.left_field
272	self.create()
273
274	def create(self):
275	"""Internal: Create the table with the joined data"""
276	self.tablename = self.db.new_table_name()
277
278	self.right_table.ensure_index(self.right_field)
279
280	# Coalesce the column information
281	visited = {}
282	columns = []
283	for col in self.left_table.columns + self.right_table.columns:
284	if col.name in visited:
285	continue
286	columns.append(col)
287	TransientTableBase.create(self, columns)
288
289	# Copy the joined data to the table.
290	internal_names = [col.internal_name for col in self.columns]
291	stmt = "INSERT INTO %s (%s) SELECT %s FROM %s JOIN %s ON %s = %s;" \
292	% (self.tablename,
293	", ".join(internal_names),
294	", ".join(internal_names),
295	self.left_table.tablename,
296	self.right_table.tablename,
297	self.orig_to_internal[self.left_field],
298	self.orig_to_internal[self.right_field])
299	self.db.execute(stmt)
300
301
302	class AutoTransientTable(table.OldTableInterfaceMixin):
303
304	"""Table that copies data to a transient table on demand.
305
306	The AutoTransientTable takes another table as input and copies data
307	to a table in a TransientDatabase instance on demand.
308	"""
309
310	def __init__(self, transient_db, table):
311	self.transient_db = transient_db
312	self.table = table
313	self.t_table = None
314
315	def Columns(self):
316	return self.table.Columns()
317
318	def Column(self, col):
319	return self.table.Column(col)
320
321	def HasColumn(self, col):
322	"""Return whether the table has a column with the given name or index
323	"""
324	return self.table.HasColumn(col)
325
326	def NumRows(self):
327	return self.table.NumRows()
328
329	def NumColumns(self):
330	return self.table.NumColumns()
331
332	def ReadRowAsDict(self, record):
333	"""Return the record no. record as a dict mapping field names to values
334	"""
335	if self.t_table is not None:
336	return self.t_table.ReadRowAsDict(record)
337	else:
338	return self.table.ReadRowAsDict(record)
339
340	def copy_to_transient(self):
341	"""Internal: Create a transient table and copy the data into it"""
342	self.t_table = TransientTable(self.transient_db, self)
343
344	def transient_table(self):
345	"""
346	Return a table whose underlying implementation is in the transient db
347	"""
348	if self.t_table is None:
349	self.copy_to_transient()
350	return self.t_table
351
352	def ValueRange(self, col):
353	if self.t_table is None:
354	self.copy_to_transient()
355	return self.t_table.ValueRange(col)
356
357	def UniqueValues(self, col):
358	if self.t_table is None:
359	self.copy_to_transient()
360	return self.t_table.UniqueValues(col)
Name	Value
svn:eol-style	native
svn:keywords	Author Date Id Revision