view gafa_datatypes.py @ 0:f86b4ae77475 draft default tip

planemo upload for repository https://github.com/TGAC/earlham-galaxytools/tree/master/tools/aequatus/ commit 6a1dfb73f2172182096724ce3bee400287fd9c2f-dirty

import logging

from galaxy.datatypes.binary import Binary, SQlite
from galaxy.util import sqlite

log = logging.getLogger(__name__)


class GAFASQLite(SQlite):
    """Class describing a GAFA SQLite database"""
    file_ext = 'gafa.sqlite'

    def set_meta(self, dataset, overwrite=True, **kwd):
        super(GAFASQLite, self).set_meta(dataset, overwrite=overwrite, **kwd)
        try:
            conn = sqlite.connect(dataset.file_name)
            c = conn.cursor()
            version_query = 'SELECT meta FROM version'
            results = c.execute(version_query).fetchall()
            if len(results) == 0:
                raise Exception('version not found in meta table')
            elif len(results) > 1:
                raise Exception('Multiple versions found in meta table')
            dataset.metadata.gafa_version = results[0][0]
        except Exception as e:
            log.warn("%s, set_meta Exception: %s", self, e)

    def sniff(self, filename):
        if super(GAFASQLite, self).sniff(filename):
            gafa_table_names = frozenset(['gene', 'gene_family', 'gene_family_member', 'meta', 'transcript'])
            conn = sqlite.connect(filename)
            c = conn.cursor()
            tables_query = "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
            results = c.execute(tables_query).fetchall()
            found_table_names = frozenset(_[0] for _ in results)
            return gafa_table_names <= found_table_names
        return False


Binary.register_sniffable_binary_format('gafa.sqlite', 'gafa.sqlite', GAFASQLite)