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:

    """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 = {}

        # Create the column objects and fill various maps and lists
        for col in self.columns:
            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

        # 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 field_count(self):
        return len(self.columns)

    def field_info(self, i):
        col = self.columns[i]
        return col.type, col.name, 0, 0

    def field_info_by_name(self, name):
        for col in self.columns:
            if col.name == name:
                return col.type, col.name, 0, 0
        else:
            return None

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

    def read_record(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 field_range(self, colname):
        col = self.name_to_column[colname]
        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), None), (converter(max), None))

    def GetUniqueValues(self, colname):
        iname = self.orig_to_internal[colname]
        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 i in range(table.field_count()):
            type, name = table.field_info(i)[:2]
            columns.append(ColumnReference(name, 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.record_count()):
            cursor.execute(insert_template, table.read_record(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 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 record_count(self):
        """Return the number of records"""
        return self.table.record_count()

    def field_count(self):
        """Return the number of fields in a record"""
        return self.table.field_count()

    def field_info(self, field):
        """Return a tuple (type, name, width, prec) for the field no. field

        type is the data type of the field, name the name, width the
        field width in characters and prec the decimal precision.
        """
        info = self.table.field_info(field)
        if info:
            info = info[:2] + (0, 0)
        return info

    def field_info_by_name(self, fieldName):
        info = self.table.field_info_by_name(fieldName)
        if info:
            info = info[:2] + (0, 0)
        return info

    def read_record(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.read_record(record)
        else:
            return self.table.read_record(record)

    def write_record(self, record, values):
        raise NotImplementedError

    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 field_range(self, colname):
        if self.t_table is None:
            self.copy_to_transient()
        return self.t_table.field_range(colname)

    def GetUniqueValues(self, colname):
        if self.t_table is None:
            self.copy_to_transient()
        return self.t_table.GetUniqueValues(colname)
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:
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
108	# Create the column objects and fill various maps and lists
109	for col in self.columns:
110	self.name_to_column[col.name] = col
111	self.orig_names.append(col.name)
112	self.internal_to_orig[col.internal_name] = col.name
113	self.orig_to_internal[col.name] = col.internal_name
114
115	# Build the CREATE TABLE statement and create the table in the
116	# database
117	table_types = []
118	for col in self.columns:
119	table_types.append("%s %s" % (col.internal_name,
120	sql_type_map[col.type]))
121	table_stmt = "CREATE TABLE %s (\n %s\n);" % (self.tablename,
122	",\n ".join(table_types))
123	self.db.execute(table_stmt)
124
125	def transient_table(self):
126	"""
127	Return a table whose underlying implementation is in the transient db
128	"""
129	return self
130
131	def ensure_index(self, column):
132	"""Ensure that there's an index on the given column"""
133	if not column in self.indexed_columns:
134	internal_name = self.orig_to_internal[column]
135	indexname = "Index_%s_%s" % (self.tablename, internal_name)
136	stmt = "CREATE INDEX %s ON %s (%s);" % (indexname, self.tablename,
137	internal_name)
138	self.db.execute(stmt)
139	self.indexed_columns[column] = 1
140
141	def field_count(self):
142	return len(self.columns)
143
144	def field_info(self, i):
145	col = self.columns[i]
146	return col.type, col.name, 0, 0
147
148	def field_info_by_name(self, name):
149	for col in self.columns:
150	if col.name == name:
151	return col.type, col.name, 0, 0
152	else:
153	return None
154
155	def record_count(self):
156	result = self.db.execute("SELECT count(*) FROM %s;" % self.tablename)
157	return int(result[0])
158
159	def read_record(self, index):
160	if self.read_record_cursor is None or index <self.read_record_last_row:
161	stmt = ("SELECT %s FROM %s;"
162	% (", ".join([c.internal_name for c in self.columns]),
163	self.tablename))
164	self.read_record_cursor = self.db.cursor()
165	self.read_record_cursor.execute(stmt)
166	self.read_record_last_row = -1
167	self.read_record_last_result = None
168
169	# Now we should have a cursor at a position less than or equal
170	# to the index so the following if statement will always set
171	# result to a suitable value
172	assert index >= self.read_record_last_row
173
174	if index == self.read_record_last_row:
175	result = self.read_record_last_result
176	else:
177	for i in range(index - self.read_record_last_row):
178	result = self.read_record_cursor.fetchone()
179	self.read_record_last_result = result
180	self.read_record_last_row = index
181	return dict(zip(self.orig_names, result))
182
183	def field_range(self, colname):
184	col = self.name_to_column[colname]
185	iname = col.internal_name
186	min, max = self.db.execute("SELECT min(%s), max(%s) FROM %s;"
187	% (iname, iname, self.tablename))
188	converter = type_converter_map[col.type]
189	return ((converter(min), None), (converter(max), None))
190
191	def GetUniqueValues(self, colname):
192	iname = self.orig_to_internal[colname]
193	cursor = self.db.cursor()
194	cursor.execute("SELECT %s FROM %s GROUP BY %s;"
195	% (iname, self.tablename, iname))
196	result = []
197	while 1:
198	row = cursor.fetchone()
199	if row is None:
200	break
201	result.append(row[0])
202	return result
203
204
205	class TransientTable(TransientTableBase):
206
207	"""A Table in a transient DB that starts as the copy of a Thuban Table."""
208
209	def __init__(self, transient_db, table):
210	"""Create a new table in the given transient DB as a copy of table
211
212	The table argument can be any object implementing the Table
213	interface.
214	"""
215	TransientTableBase.__init__(self, transient_db)
216	self.create(table)
217
218	def create(self, table):
219	columns = []
220	for i in range(table.field_count()):
221	type, name = table.field_info(i)[:2]
222	columns.append(ColumnReference(name, type,
223	self.db.new_column_name()))
224	TransientTableBase.create(self, columns)
225
226	# copy the input table to the transient db
227	insert_template = "INSERT INTO %s (%s) VALUES (%s);" \
228	% (self.tablename,
229	", ".join([col.internal_name
230	for col in self.columns]),
231	", ".join(["%%(%s)s" % col.name
232	for col in self.columns]))
233	cursor = self.db.cursor()
234	for i in range(table.record_count()):
235	cursor.execute(insert_template, table.read_record(i))
236	self.db.conn.commit()
237
238
239
240	class TransientJoinedTable(TransientTableBase):
241
242	"""A Table in the transient DB that contains a join of two tables"""
243
244	def __init__(self, transient_db, left_table, left_field,
245	right_table, right_field = None):
246	"""Create a new table in the transient DB as a join of two tables.
247
248	Both input tables, left_table and right_table must have a
249	transient_table method that returns a table object for a table
250	in the trnsient database. The join is performed on the condition
251	that the value of the left_field column the the left table is
252	equal to the value of the right_field in the right_table.
253
254	The joined table contains all columns of the input tables with
255	one exception: Any column in the right_table with the same name
256	as one of the columns in the left_table will be omitted. This is
257	somewhat of an implementation detail, but is done so that the
258	column names of the joined table can be the same as the column
259	names of the input tables without having to create prefixes.
260	"""
261	TransientTableBase.__init__(self, transient_db)
262	self.left_table = left_table.transient_table()
263	self.left_field = left_field
264	self.right_table = right_table.transient_table()
265	if right_field:
266	self.right_field = right_field
267	else:
268	self.right_field = self.left_field
269	self.create()
270
271	def create(self):
272	"""Internal: Create the table with the joined data"""
273	self.tablename = self.db.new_table_name()
274
275	self.right_table.ensure_index(self.right_field)
276
277	# Coalesce the column information
278	visited = {}
279	columns = []
280	for col in self.left_table.columns + self.right_table.columns:
281	if col.name in visited:
282	continue
283	columns.append(col)
284	TransientTableBase.create(self, columns)
285
286	# Copy the joined data to the table.
287	internal_names = [col.internal_name for col in self.columns]
288	stmt = "INSERT INTO %s (%s) SELECT %s FROM %s JOIN %s ON %s = %s;" \
289	% (self.tablename,
290	", ".join(internal_names),
291	", ".join(internal_names),
292	self.left_table.tablename,
293	self.right_table.tablename,
294	self.orig_to_internal[self.left_field],
295	self.orig_to_internal[self.right_field])
296	self.db.execute(stmt)
297
298
299	class AutoTransientTable:
300
301	"""Table that copies data to a transient table on demand.
302
303	The AutoTransientTable takes another table as input and copies data
304	to a table in a TransientDatabase instance on demand.
305	"""
306
307	def __init__(self, transient_db, table):
308	self.transient_db = transient_db
309	self.table = table
310	self.t_table = None
311
312	def record_count(self):
313	"""Return the number of records"""
314	return self.table.record_count()
315
316	def field_count(self):
317	"""Return the number of fields in a record"""
318	return self.table.field_count()
319
320	def field_info(self, field):
321	"""Return a tuple (type, name, width, prec) for the field no. field
322
323	type is the data type of the field, name the name, width the
324	field width in characters and prec the decimal precision.
325	"""
326	info = self.table.field_info(field)
327	if info:
328	info = info[:2] + (0, 0)
329	return info
330
331	def field_info_by_name(self, fieldName):
332	info = self.table.field_info_by_name(fieldName)
333	if info:
334	info = info[:2] + (0, 0)
335	return info
336
337	def read_record(self, record):
338	"""Return the record no. record as a dict mapping field names to values
339	"""
340	if self.t_table is not None:
341	return self.t_table.read_record(record)
342	else:
343	return self.table.read_record(record)
344
345	def write_record(self, record, values):
346	raise NotImplementedError
347
348	def copy_to_transient(self):
349	"""Internal: Create a transient table and copy the data into it"""
350	self.t_table = TransientTable(self.transient_db, self)
351
352	def transient_table(self):
353	"""
354	Return a table whose underlying implementation is in the transient db
355	"""
356	if self.t_table is None:
357	self.copy_to_transient()
358	return self.t_table
359
360	def field_range(self, colname):
361	if self.t_table is None:
362	self.copy_to_transient()
363	return self.t_table.field_range(colname)
364
365	def GetUniqueValues(self, colname):
366	if self.t_table is None:
367	self.copy_to_transient()
368	return self.t_table.GetUniqueValues(colname)
Name	Value
svn:eol-style	native
svn:keywords	Author Date Id Revision