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 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 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.read_record(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()
        (min, row), (max, row) = self.t_table.field_range(col)
        return min, max

    def UniqueValues(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(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 ReadRowAsDict(self, index):
159	if self.read_record_cursor is None or index <self.read_record_last_row:
160	stmt = ("SELECT %s FROM %s;"
161	% (", ".join([c.internal_name for c in self.columns]),
162	self.tablename))
163	self.read_record_cursor = self.db.cursor()
164	self.read_record_cursor.execute(stmt)
165	self.read_record_last_row = -1
166	self.read_record_last_result = None
167
168	# Now we should have a cursor at a position less than or equal
169	# to the index so the following if statement will always set
170	# result to a suitable value
171	assert index >= self.read_record_last_row
172
173	if index == self.read_record_last_row:
174	result = self.read_record_last_result
175	else:
176	for i in range(index - self.read_record_last_row):
177	result = self.read_record_cursor.fetchone()
178	self.read_record_last_result = result
179	self.read_record_last_row = index
180	return dict(zip(self.orig_names, result))
181
182	def ValueRange(self, col):
183	col = self.column_map[col]
184	iname = col.internal_name
185	min, max = self.db.execute("SELECT min(%s), max(%s) FROM %s;"
186	% (iname, iname, self.tablename))
187	converter = type_converter_map[col.type]
188	return (converter(min), converter(max))
189
190	def UniqueValues(self, col):
191	iname = self.column_map[col].internal_name
192	cursor = self.db.cursor()
193	cursor.execute("SELECT %s FROM %s GROUP BY %s;"
194	% (iname, self.tablename, iname))
195	result = []
196	while 1:
197	row = cursor.fetchone()
198	if row is None:
199	break
200	result.append(row[0])
201	return result
202
203
204	class TransientTable(TransientTableBase):
205
206	"""A Table in a transient DB that starts as the copy of a Thuban Table."""
207
208	def __init__(self, transient_db, table):
209	"""Create a new table in the given transient DB as a copy of table
210
211	The table argument can be any object implementing the Table
212	interface.
213	"""
214	TransientTableBase.__init__(self, transient_db)
215	self.create(table)
216
217	def create(self, table):
218	columns = []
219	for col in table.Columns():
220	columns.append(ColumnReference(col.name, col.type,
221	self.db.new_column_name()))
222	TransientTableBase.create(self, columns)
223
224	# copy the input table to the transient db
225	insert_template = "INSERT INTO %s (%s) VALUES (%s);" \
226	% (self.tablename,
227	", ".join([col.internal_name
228	for col in self.columns]),
229	", ".join(["%%(%s)s" % col.name
230	for col in self.columns]))
231	cursor = self.db.cursor()
232	for i in range(table.NumRows()):
233	cursor.execute(insert_template, table.ReadRowAsDict(i))
234	self.db.conn.commit()
235
236
237
238	class TransientJoinedTable(TransientTableBase):
239
240	"""A Table in the transient DB that contains a join of two tables"""
241
242	def __init__(self, transient_db, left_table, left_field,
243	right_table, right_field = None):
244	"""Create a new table in the transient DB as a join of two tables.
245
246	Both input tables, left_table and right_table must have a
247	transient_table method that returns a table object for a table
248	in the trnsient database. The join is performed on the condition
249	that the value of the left_field column the the left table is
250	equal to the value of the right_field in the right_table.
251
252	The joined table contains all columns of the input tables with
253	one exception: Any column in the right_table with the same name
254	as one of the columns in the left_table will be omitted. This is
255	somewhat of an implementation detail, but is done so that the
256	column names of the joined table can be the same as the column
257	names of the input tables without having to create prefixes.
258	"""
259	TransientTableBase.__init__(self, transient_db)
260	self.left_table = left_table.transient_table()
261	self.left_field = left_field
262	self.right_table = right_table.transient_table()
263	if right_field:
264	self.right_field = right_field
265	else:
266	self.right_field = self.left_field
267	self.create()
268
269	def create(self):
270	"""Internal: Create the table with the joined data"""
271	self.tablename = self.db.new_table_name()
272
273	self.right_table.ensure_index(self.right_field)
274
275	# Coalesce the column information
276	visited = {}
277	columns = []
278	for col in self.left_table.columns + self.right_table.columns:
279	if col.name in visited:
280	continue
281	columns.append(col)
282	TransientTableBase.create(self, columns)
283
284	# Copy the joined data to the table.
285	internal_names = [col.internal_name for col in self.columns]
286	stmt = "INSERT INTO %s (%s) SELECT %s FROM %s JOIN %s ON %s = %s;" \
287	% (self.tablename,
288	", ".join(internal_names),
289	", ".join(internal_names),
290	self.left_table.tablename,
291	self.right_table.tablename,
292	self.orig_to_internal[self.left_field],
293	self.orig_to_internal[self.right_field])
294	self.db.execute(stmt)
295
296
297	class AutoTransientTable(table.OldTableInterfaceMixin):
298
299	"""Table that copies data to a transient table on demand.
300
301	The AutoTransientTable takes another table as input and copies data
302	to a table in a TransientDatabase instance on demand.
303	"""
304
305	def __init__(self, transient_db, table):
306	self.transient_db = transient_db
307	self.table = table
308	self.t_table = None
309
310	def Columns(self):
311	return self.table.Columns()
312
313	def Column(self, col):
314	return self.table.Column(col)
315
316	def NumRows(self):
317	return self.table.NumRows()
318
319	def NumColumns(self):
320	return self.table.NumColumns()
321
322	def ReadRowAsDict(self, record):
323	"""Return the record no. record as a dict mapping field names to values
324	"""
325	if self.t_table is not None:
326	return self.t_table.read_record(record)
327	else:
328	return self.table.ReadRowAsDict(record)
329
330	def copy_to_transient(self):
331	"""Internal: Create a transient table and copy the data into it"""
332	self.t_table = TransientTable(self.transient_db, self)
333
334	def transient_table(self):
335	"""
336	Return a table whose underlying implementation is in the transient db
337	"""
338	if self.t_table is None:
339	self.copy_to_transient()
340	return self.t_table
341
342	def ValueRange(self, col):
343	if self.t_table is None:
344	self.copy_to_transient()
345	(min, row), (max, row) = self.t_table.field_range(col)
346	return min, max
347
348	def UniqueValues(self, colname):
349	if self.t_table is None:
350	self.copy_to_transient()
351	return self.t_table.GetUniqueValues(colname)
Name	Value
svn:eol-style	native
svn:keywords	Author Date Id Revision