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	bh	765	# 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	bh	777	if self.conn is not None:
57	bh	765	self.conn.close()
58	bh	777	self.conn = None
59	bh	765
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	bh	818	class TransientTableBase(table.OldTableInterfaceMixin):
89	bh	765
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	bh	785	self.read_record_last_result = None
100	bh	765
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	bh	818	self.column_map = {}
108	bh	765
109			# Create the column objects and fill various maps and lists
110	bh	818	for index in range(len(self.columns)):
111			col = self.columns[index]
112	bh	765	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	bh	818	self.column_map[col.name] = col
117			self.column_map[index] = col
118	bh	765
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	bh	818	def NumColumns(self):
146	bh	765	return len(self.columns)
147
148	bh	818	def NumRows(self):
149	bh	765	result = self.db.execute("SELECT count(*) FROM %s;" % self.tablename)
150			return int(result[0])
151
152	bh	818	def Columns(self):
153			return self.columns
154
155			def Column(self, col):
156			return self.column_map[col]
157
158	bh	839	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	bh	818	def ReadRowAsDict(self, index):
164	bh	765	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	bh	785	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	bh	818	result = self.read_record_last_result
180	bh	785	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	bh	765	self.read_record_last_row = index
185	bh	785	return dict(zip(self.orig_names, result))
186	bh	765
187	bh	818	def ValueRange(self, col):
188			col = self.column_map[col]
189	bh	765	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	bh	818	return (converter(min), converter(max))
194	bh	765
195	bh	818	def UniqueValues(self, col):
196			iname = self.column_map[col].internal_name
197	bh	765	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	bh	818	for col in table.Columns():
225			columns.append(ColumnReference(col.name, col.type,
226	bh	765	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	bh	818	for i in range(table.NumRows()):
238			cursor.execute(insert_template, table.ReadRowAsDict(i))
239	bh	765	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	bh	818	class AutoTransientTable(table.OldTableInterfaceMixin):
303	bh	765
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	bh	818	def Columns(self):
316			return self.table.Columns()
317	bh	765
318	bh	818	def Column(self, col):
319			return self.table.Column(col)
320	bh	765
321	bh	839	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	bh	818	def NumRows(self):
327			return self.table.NumRows()
328	bh	765
329	bh	818	def NumColumns(self):
330			return self.table.NumColumns()
331	bh	765
332	bh	818	def ReadRowAsDict(self, record):
333	bh	765	"""Return the record no. record as a dict mapping field names to values
334			"""
335			if self.t_table is not None:
336	bh	839	return self.t_table.ReadRowAsDict(record)
337	bh	765	else:
338	bh	818	return self.table.ReadRowAsDict(record)
339	bh	765
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	bh	818	def ValueRange(self, col):
353	bh	765	if self.t_table is None:
354			self.copy_to_transient()
355	bh	839	return self.t_table.ValueRange(col)
356	bh	765
357	bh	839	def UniqueValues(self, col):
358	bh	765	if self.t_table is None:
359			self.copy_to_transient()
360	bh	839	return self.t_table.UniqueValues(col)
Name	Value
svn:eol-style	native
svn:keywords	Author Date Id Revision