### Navigation

- index
- modules |
- versuchung 0.1-0 documentation »
- Module code »

# Source code for versuchung.database

    # This file is part of versuchung.
    #
    # versuchung is free software: you can redistribute it and/or modify it under the
    # terms of the GNU General Public License as published by the Free Software
    # Foundation, either version 3 of the License, or (at your option) any later
    # version.
    #
    # versuchung is distributed in the hope that it will be useful, but WITHOUT ANY
    # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
    # PARTICULAR PURPOSE.  See the GNU General Public License for more details.
    #
    # You should have received a copy of the GNU General Public License along with
    # versuchung.  If not, see <http://www.gnu.org/licenses/>.

    from versuchung.types import Type, InputParameter, OutputParameter
    import logging
    import sqlite3
    import os, stat

    # Import mysql handler
    try:
        import MySQLdb
        import _mysql_exceptions
    except:
        pass

    class Database_Abstract:
        def values(self, table_name, filter_expr = "where experiment = ?", *args):
            """Get the contets of a table in the database. It takes
            addtional to the table name, a filter expression and applies
            all args to the excute command. An example::

               (cols, rows) = database.values("metadata", "")
               for row in rows:
                    print cols, rows
            """
            cur = self.handle.cursor()
            cur.execute('select * from ' + table_name + filter_expr,
                        args)

            cols = [x[0] for x in cur.description]
            def generator():
                while True:
                    row = cur.fetchone()
                    if row == None:
                        cur.close()
                        return
                    yield row

            index = cols.index("experiment")
            return cols, generator()

    class Database_MySQL(InputParameter, OutputParameter, Type, Database_Abstract):
        """Can be used as **input parameter** and **output parameter**

        A database backend class for a MySQL database."""

        def __init__(self, database = None, host = "localhost", user = None, password = None):
            InputParameter.__init__(self)
            OutputParameter.__init__(self)
            Type.__init__(self)

            assert database != None, "Please give a database name to database connection"

            self.__database_name = database
            self.__database_host = os.environ.get("MYSQL_HOST", None) or host
            self.__database_user = os.environ.get("MYSQL_USER", None) or user
            self.__database_password = os.environ.get("MYSQL_PASSWORD", None) or password
            self.__database_connection = None

        def inp_setup_cmdline_parser(self, parser):
            self.inp_parser_add(parser, "database", self.__database_name)
            self.inp_parser_add(parser, "host", self.__database_host)
            self.inp_parser_add(parser, "user", self.__database_user)
            self.inp_parser_add(parser, "password", self.__database_password)

        def inp_extract_cmdline_parser(self, opts, args):
            self.__database_name = self.inp_parser_extract(opts, "database")
            self.__database_host = self.inp_parser_extract(opts, "host")
            self.__database_user = self.inp_parser_extract(opts, "user")
            self.__database_password = self.inp_parser_extract(opts, "password")

        def inp_metadata(self):
            return {self.name + "-database": self.__database_name,
                    self.name + "-host": self.__database_host}

        def install_my_cnf(self):
            """Creates a my.cnf file and sets the environment variable MYSQL_HOME"""
            directory = self.tmp_directory.new_directory(self.name)
            path = os.path.join(directory.path, "my.cnf")
            logging.debug("MYSQL_HOME=%s", path)
            with os.fdopen(os.open(path, os.O_WRONLY | os.O_CREAT, 0600), 'w') as handle:
                handle.write("""[client]
    host=%s
    user=%s
    password=%s
    database=%s
    """ %(self.__database_host, self.__database_user, self.__database_password, self.__database_name))

            # Set the environment variable to the directory the my.cnf is located
            os.environ["MYSQL_HOME"] = directory.path

            if os.path.exists(os.path.join(os.environ["HOME"], ".my.cnf")):
                logging.warning("~/.my.cnf does overwrite versuchung's file")

        def before_experiment_run(self, parameter_type):
            Type.before_experiment_run(self, parameter_type)
            assert parameter_type in ["input", "output"]

            args = {"db": self.__database_name,
                    "host": self.__database_host}
            if self.__database_user:
                args["user"] = self.__database_user
            if self.__database_password:
                args["passwd"] = self.__database_password

            default_file = os.path.join(os.environ["HOME"], ".my.cnf")
            if os.path.exists(default_file):
                args["read_default_file"] = default_file

            self.__database_connection = MySQLdb.connect(**args)

            if parameter_type == "output":
                try:
                    self.create_table("metadata", [("experiment", "varchar(256)"), ("metadata", "text")],
                                      keys = ["experiment"],
                                      conflict_strategy = "REPLACE")
                except _mysql_exceptions.OperationalError as e:
                    # Metadata table was already generated
                    pass

                self.execute("REPLACE INTO metadata(experiment, metadata) values(?, ?)",
                             self.dynamic_experiment.experiment_identifier,
                             str(self.dynamic_experiment.metadata))

        @property
        def handle(self):
            """:return: handle -- MySQLdb database handle"""
            assert self.__database_connection
            return self.__database_connection

        def execute(self, command, *args):
            """Execute command including the arguments on the sql
            handle. Question marks in the command are replaces by the ``*args``::

            >>> database.execute("SELECT * FROM metadata WHERE experiment = ?", identifer)
            """
            logging.debug("mysql: %s %s", str(command), str(args))
            c = self.__database_connection.cursor()
            c.execute(command.replace("?", "%s"), args)
            self.__database_connection.commit()
            return c

        def create_table(self, name, fields = [("key", "text"), ("value", "text")],
                         keys = None, conflict_strategy = None):
            """Creates a new table in the database. ``name`` is the name
            of newly created table. The ``fields`` are a list of
            columns. A column is a tuple, its first entry is the name, its
            second entry the column type. If primary is the name of a
            column this column is marked as the primary key for the
            table.

            conflict_strategy is ignored!
            """

            CT = "CREATE TABLE " + name + " ("
            CT += ", ".join([ "%s %s" % x for x in fields])
            if keys:
                assert set(keys).issubset(set([x[0] for x in fields]))
                CT += ", UNIQUE(" + (", ".join(keys)) + ")"
            CT += ")"

            return self.execute(CT)

    class Database_SQLite(InputParameter, OutputParameter, Type, Database_Abstract):
        """Can be used as **input parameter** and **output parameter**

        A database backend class for sqlite3 database."""

        # Static cache of all database connections open in system
        # Map from path -> tuple(db_handle, use_count)
        database_connections = {}

        def __init__(self, path = "sqlite3.db"):
            InputParameter.__init__(self)
            OutputParameter.__init__(self)
            Type.__init__(self)

            self.__database_path = path
            self.__database_connection = None

        def inp_setup_cmdline_parser(self, parser):
            self.inp_parser_add(parser, None, self.__database_path)

        def inp_extract_cmdline_parser(self, opts, args):
            self.__database_path = self.inp_parser_extract(opts, None)

        def inp_metadata(self):
            return {self.name: self.__database_path}

        def before_experiment_run(self, parameter_type):
            Type.before_experiment_run(self, parameter_type)
            assert parameter_type in ["input", "output"]
            if parameter_type == "input":
                # Ensure the path does exist
                if not os.path.exists(self.path):
                    raise RuntimeError("Database not found: %s" % self.path)
            self.__database_connection = self.__connect(self.path)

            if parameter_type == "output":
                try:
                    self.create_table("metadata", [("experiment", "text"), ("metadata", "text")],
                                      keys = ["experiment"],
                                      conflict_strategy = "REPLACE")
                except sqlite3.OperationalError as e:
                    # Metadata table was already generated
                    pass

                self.execute("INSERT INTO metadata(experiment, metadata) values(?, ?)",
                             self.dynamic_experiment.experiment_identifier,
                             str(self.dynamic_experiment.metadata))

        def after_experiment_run(self, parameter_type):
            Type.before_experiment_run(self, parameter_type)
            assert parameter_type in ["input", "output"]
            self.__database_connection = None
            self.__disconnect(self.path)
            if parameter_type == "output":
                # Remove execute and write permissions for file
                new_mode = os.stat(self.path).st_mode & (stat.S_IROTH | stat.S_IRGRP | stat.S_IRUSR)
                os.chmod(self.path, new_mode)

        @staticmethod
        def __connect(path):
            # Do reference counting on database connections
            if path in Database_SQLite.database_connections:
                (db, count) = Database_SQLite.database_connections[path]
                Database_SQLite.database_connections[path] = (db, count + 1)
                return db
            conn = sqlite3.connect(path)
            Database_SQLite.database_connections[path] = (conn, 1)
            return conn

        @staticmethod
        def __disconnect(path):
            (db, count) = Database_SQLite.database_connections[path]
            db.commit()
            if count == 1:
                db.close()
                del Database_SQLite.database_connections[path]
                return
            Database_SQLite.database_connections[path] = (db, count - 1)

        @property
        def path(self):
            """:return: string -- path to the sqlite database file"""
            return os.path.join(self.base_directory, self.__database_path)

        @property
        def handle(self):
            """:return: handle -- sqlite3 database handle"""
            assert self.__database_connection
            return self.__database_connection

        def execute(self, command, *args):
            """Execute command including the arguments on the sql
            handle. Question marks in the command are replaces by the ``*args``::

            >>> database.execute("SELECT * FROM metadata WHERE experiment = ?", identifer)
            """
            logging.debug("sqlite: %s %s", str(command), str(args))
            return self.__database_connection.execute(command, args)

        def create_table(self, name, fields = [("key", "text"), ("value", "text")],
                         keys = None, conflict_strategy = "REPLACE"):
            """Creates a new table in the database. ``name`` is the name
            of newly created table. The ``fields`` are a list of
            columns. A column is a tuple, its first entry is the name, its
            second entry the column type. If primary is the name of a
            column this column is marked as the primary key for the
            table."""

            CT = "CREATE TABLE " + name + " ("
            CT += ", ".join([ "%s %s" % x for x in fields])
            if keys:
                assert set(keys).issubset(set([x[0] for x in fields]))
                CT += ", UNIQUE(" + (", ".join(keys)) + ")"
                CT += " ON CONFLICT "  + conflict_strategy
            CT += ")"

            return self.execute(CT)

    def Database( database_type = "sqlite", *args, **kwargs):
        """This is a just a wrapper around the supported database
        abstraction classes. Every other argument and paramater than
        ``database_type`` is forwared directly to those classes.

        Supported database_type abstractions are at the moment:

        - ``sqlite`` -- :class:`~versuchung.database.Database_SQLite`
        - ``mysql`` -- :class:`~versuchung.database.Database_MySQL`
        """
        if database_type == "sqlite":
            if not "path" in kwargs:
                kwargs["path"] = "sqlite3.db"
            return Database_SQLite(*args, **kwargs)
        if database_type == "mysql":
            return Database_MySQL(*args, **kwargs)
        assert False, "Database type %s is not implemented yet" % database_type

    class Table(InputParameter, OutputParameter, Type):
        """Can be used as **input parameter** and **output parameter**

        A versuchung table is a table that is aware of experiments. It
        stores for each dataset the experiment it originates from. The
        field list consists either of plain strings, then the column type
        is text. If it's a tuple the first entry is the name and the second its type::

        >>> [("foo", "integer"), "barfoo"]

        This will result in two columns, one with type integer and one
        with type text. If a db is given this one is used instead of a
        default sqlite database named ``sqlite3.db``

        To make a set of field the index keys (UNIQUE), give it as a list
        of string as keys argument. The conflict_strategy gives the SQL
        strategy what to do on conflict. If you want to merge databases from
        multiple experiments without triggering a conflict if the given key
        set is equal (i.e., if you want the same values in the columns given
        as keys to be treated different when coming from different
        experiments), add ``experiment`` to the key set.
        """
        def __init__(self, fields, keys = None, db = None, conflict_strategy = "FAIL" ):
            self.read_only = True
            InputParameter.__init__(self)
            OutputParameter.__init__(self)
            Type.__init__(self)

            self.__keys = keys
            self.__fields = self.__field_typify(["experiment"] + fields)
            self.__conflict_strategy = conflict_strategy

            if not db:
                self.__db = Database()
            else:
                self.__db = db

        def __field_typify(self, fields):
            real_fields = []
            for f in fields:
                if type(f) in [tuple, list]:
                    real_fields.append(tuple(f))
                else:
                    # the default field type is text
                    assert type(f) == str
                    real_fields.append(tuple([f, 'text']))
            return real_fields

        def before_experiment_run(self, parameter_type):
            # Add database object as an
            self.subobjects["database"] = self.__db
            Type.before_experiment_run(self, parameter_type)

            if parameter_type == "output":
                self.read_only = False
                self.__db.create_table(self.table_name, self.__fields,
                                       keys = self.__keys,
                                       conflict_strategy = self.__conflict_strategy)
        @property
        def database(self):
            """:return: :class:`~versuchung.database.Database` -- the database the table is located in"""
            return self.__db

        @property
        def table_name(self):
            """:return: string -- return the name of the table in the database"""
            assert self.static_experiment
            name = self.name
            try:
                i = self.name.rindex("-")
                name = name[i+1:]
            except:
                pass
            return self.static_experiment.title + "__" + name

        def insert(self, data=None, **kwargs):
            """Insert a dict of data into the database table"""
            assert self.read_only == False
            if data:
                kwargs.update(data)
            kwargs["experiment"] = self.dynamic_experiment.experiment_identifier
            assert set(kwargs.keys()) == set([f for f, t in self.__fields])

            items = kwargs.items()
            insert_statement = "INSERT INTO %s(%s) values(%s)" % (
                self.table_name,
                ", ".join([f for f, t in items]),
                ", ".join(["?" for _ in items]))
            self.__db.execute(insert_statement, *[v for k,v in items])

        def clear(self):
            """Remove all entries associated with the current running experiment"""

            self.__db.execute("DELETE FROM " + self.table_name +" WHERE experiment = ?",
                              self.dynamic_experiment.experiment_identifier)

        @property
        def value(self):
            """The value of the table. It returns a tuple. The first entry
            is a tuple of column headings. The second entry is a list of
            rows, in the same order as the column headings. The column
            that associates the entry with the experiment is stripped
            apart and only data for the static enclosing experiment is
            shown."""
            (cols, rows) = self.__db.values(self.table_name, ' where experiment = ?',
                                            self.static_experiment.experiment_identifier)

            index = cols.index("experiment")
            table = []
            for row in rows:
                l = list(row)
                del l[index]
                table.append(tuple(l))
            del cols[index]

            return tuple(cols), table

    class TableDict(Table, dict):
        """Can be used as **input parameter** and **output parameter**

        This uses a :class:`~versuchung.database.Table` as a backend for a
        python dict. This object can be used in the same way
        :class:`~versuchung.tex.PgfKeyDict` is used. Please be aware, that
        the storage and retrieval of keys from the associated table is
        done lazy. Therefore the data is only then visible if the
        experiment was successful.
        """
        def __init__(self, db=None):
            self.__key_name = "key"
            self.__value_name = "value"
            columns = [(self.__key_name, 'text'), (self.__value_name, 'text')]
            Table.__init__(self, columns, keys=[self.__key_name], db=db,
                           conflict_strategy = "REPLACE")
            dict.__init__(self)

        def insert(self, *args, **kwargs):
            raise NotImplementedError

        def flush(self):
            """Save the current dict content to the database."""
            Table.clear(self)
            for key, value in self.items():
                Table.insert(self,
                             {self.__key_name: key,
                              self.__value_name: value})

        def after_experiment_run(self, parameter_type):
            assert self.parameter_type == parameter_type
            if parameter_type == "output":
                self.flush()
            Table.after_experiment_run(self, parameter_type)

        def before_experiment_run(self, parameter_type):
            Table.before_experiment_run(self, parameter_type)
            if parameter_type == "input":
                (header, values) = self.value
                key_index = header.index(self.__key_name)
                value_index = header.index(self.__value_name)
                self.update([(x[key_index], x[value_index]) for x in values])

    class Database_SQlite_Merger:
        def log(self, msg, *args):
            if self.logging:
                print "merger: " + (msg % args)

        def __init__(self, target_path, source_paths = [], logging = True):
            self.target_path = target_path
            self.logging = logging
            self.source_paths = {}
            self.target = sqlite3.connect(target_path)

            db_counter = 0
            for source in source_paths:
                assert os.path.exists(source), "Path does not exist " + source
                name = "db_%d" % db_counter
                db_counter += 1
                self.target.execute("ATTACH DATABASE '%s' AS %s" %(source, name))
                self.log("attached %s as %s", source, name)
                self.source_paths[name] = source

        def collect_and_create_tables(self, drop = True):
            cur = self.target.cursor()
            self.tables = {}
            for db in self.source_paths:
                cur.execute("SELECT * FROM " + db + ".sqlite_master WHERE type = 'table'")
                header = [x[0] for x in cur.description]
                for table in cur.fetchall():
                    table = dict(zip(header, table))
                    name = table["name"]
                    if table["name"] in self.tables:
                        if self.tables[name]["sql"] != table["sql"]:
                            self.log("Two tables with different defintions found: %s" % name)
                            sys.exit(-1)
                        self.tables[name]["databases"].append(db)
                    else:
                        self.tables[name] = table
                        self.tables[name]["databases"] = [db]
            for name, table in self.tables.items():
                if drop:
                    try:
                        cur.execute("DROP TABLE %s" % name)
                    except:
                        pass
                    cur.execute(table["sql"])
                else:
                    cur.execute(table["sql"].replace('CREATE TABLE', 'CREATE TABLE IF NOT EXISTS'))

                self.log("created table %s", name)

            cur.close()

        def collect_data(self):
            cur = self.target.cursor()

            TableDictrows = set()

            for name in self.tables:
                rows = set()
                headers = None
                for db in self.tables[name]["databases"]:
                    cur.execute("SELECT * FROM %s.%s" % (db, name))
                    for i in cur.fetchall():
                        rows.add(i)
                    headers = [x[0] for x in cur.description]

                cur.executemany("INSERT INTO %s (%s) values(%s)" % (\
                        name,
                        ", ".join(headers),
                        ", ".join(["?" for x in headers])
                        ), rows)
                self.log("inserted %d rows into %s", len(rows), name)

                if headers == ["experiment", "key", "value"]:
                    TableDictrows.update(rows)

            cur.execute("CREATE TABLE IF NOT EXISTS TableDict (experiment text, key text, value text,"
                        "UNIQUE (key) ON CONFLICT REPLACE)")
            cur.executemany("INSERT INTO TableDict (experiment, key, value) values(?,?,?)",
                            TableDictrows)
            self.log("inserted %d key-value pairs into TableDict", len(TableDictrows))
            cur.close()
            self.target.commit()

        def merge(self, update = True):
            """Do the actual merge operation"""
            self.collect_and_create_tables(drop = not update)
            self.collect_data()
            self.target.close()

    if __name__ == '__main__':
        import sys
        if len(sys.argv) < 2:
            print sys.argv[0] + " <target-database-file> [<source-db1> <source-db2> ...]"
            print " -- merges different versuchung sqlite databases into a single one"
            sys.exit(-1)

        merger = Database_SQlite_Merger(sys.argv[1], sys.argv[2:])
        merger.merge()

### Navigation

- index
- modules |
- versuchung 0.1-0 documentation »
- Module code »