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	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			def ReadRowAsDict(self, index):
159	bh	765	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	bh	785	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	bh	818	result = self.read_record_last_result
175	bh	785	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	bh	765	self.read_record_last_row = index
180	bh	785	return dict(zip(self.orig_names, result))
181	bh	765
182	bh	818	def ValueRange(self, col):
183			col = self.column_map[col]
184	bh	765	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	bh	818	return (converter(min), converter(max))
189	bh	765
190	bh	818	def UniqueValues(self, col):
191			iname = self.column_map[col].internal_name
192	bh	765	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	bh	818	for col in table.Columns():
220			columns.append(ColumnReference(col.name, col.type,
221	bh	765	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	bh	818	for i in range(table.NumRows()):
233			cursor.execute(insert_template, table.ReadRowAsDict(i))
234	bh	765	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	bh	818	class AutoTransientTable(table.OldTableInterfaceMixin):
298	bh	765
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	bh	818	def Columns(self):
311			return self.table.Columns()
312	bh	765
313	bh	818	def Column(self, col):
314			return self.table.Column(col)
315	bh	765
316	bh	818	def NumRows(self):
317			return self.table.NumRows()
318	bh	765
319	bh	818	def NumColumns(self):
320			return self.table.NumColumns()
321	bh	765
322	bh	818	def ReadRowAsDict(self, record):
323	bh	765	"""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	bh	818	return self.table.ReadRowAsDict(record)
329	bh	765
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	bh	818	def ValueRange(self, col):
343	bh	765	if self.t_table is None:
344			self.copy_to_transient()
345	bh	818	(min, row), (max, row) = self.t_table.field_range(col)
346			return min, max
347	bh	765
348	bh	818	def UniqueValues(self, colname):
349	bh	765	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