Skip to content

Exomiserdb

DBConnection

Source code in src/pheval/infra/exomiserdb.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class DBConnection:
    connection = None

    def __init__(self, connection):
        DBConnection.connection = connection

    @classmethod
    def get_connection(cls) -> jaydebeapi.Connection:
        """Creates return new Singleton database connection"""
        return DBConnection.connection

    def close(self):
        return self.connection.close()

    @classmethod
    def get_cursor(cls) -> jaydebeapi.Cursor:
        connection = cls.get_connection()
        return connection.cursor()

get_connection() classmethod

Creates return new Singleton database connection

Source code in src/pheval/infra/exomiserdb.py 
47
48
49
50
@classmethod
def get_connection(cls) -> jaydebeapi.Connection:
    """Creates return new Singleton database connection"""
    return DBConnection.connection

DBConnector

Source code in src/pheval/infra/exomiserdb.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class DBConnector:
    def __init__(self, jar: Path, driver: str, server: str, database: str, user: str, password: str):
        self.jar = jar
        self.driver = driver
        self.server = server
        self.database = database
        self.user = user
        self.password = password
        self.dbconn = None

    def create_connection(self) -> jaydebeapi.Connection:
        """creates h2 database connection"""
        return jaydebeapi.connect(
            self.driver,
            f"{self.server}{self.database}",
            [self.user, self.password],
            self.jar,
        )

    def __enter__(self) -> jaydebeapi.Connection:
        self.dbconn = self.create_connection()
        return self.dbconn

    def __exit__(self, *other):
        self.dbconn.close()

create_connection()

creates h2 database connection

Source code in src/pheval/infra/exomiserdb.py 
24
25
26
27
28
29
30
31
def create_connection(self) -> jaydebeapi.Connection:
    """creates h2 database connection"""
    return jaydebeapi.connect(
        self.driver,
        f"{self.server}{self.database}",
        [self.user, self.password],
        self.jar,
    )

ExomiserDB

Source code in src/pheval/infra/exomiserdb.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
class ExomiserDB:
    def __init__(self, db_path: Path):
        try:
            self.connector = DBConnector(
                jar=os.path.join(os.path.dirname(__file__), "../../../lib/h2-1.4.199.jar"),
                driver="org.h2.Driver",
                server=f"jdbc:h2:{db_path}",
                user="sa",
                password="",
                database="",
            )
        except Exception as e:
            print("An exception occurred", e)

    def import_from_semsim_file(self, input_file: Path, subject_prefix: str, object_prefix: str):
        """imports semsim tsv profile into exomiser phenotype database

        Args:
            input_file (Path): semsim profile
            subject_prefix (str): Subject Prefix. e.g HP
            object_prefix (str): Object Prefix. e.g MP
        """
        with self.connector as cnn:
            conn = DBConnection(cnn)
            reader = pl.read_csv_batched(input_file, separator="\t")
            batch_length = 5
            batches = reader.next_batches(batch_length)
            cursor = conn.get_cursor()
            # # TODO: Refactor this
            with open(input_file) as f:
                total = sum(1 for line in f)
            pbar = tqdm(total=total - 1)
            mapping_id = 1
            while batches:
                input_data = pl.concat(batches)
                sql = _semsim2h2(input_data, object_prefix, subject_prefix, mapping_id=mapping_id)
                cursor.execute(sql)
                len_input_data = len(input_data)
                mapping_id += len_input_data
                pbar.update(len_input_data)

                batches = reader.next_batches(batch_length)

import_from_semsim_file(input_file, subject_prefix, object_prefix)

imports semsim tsv profile into exomiser phenotype database

Parameters:

Name Type Description Default
input_file Path

semsim profile

 required
subject_prefix str

Subject Prefix. e.g HP

 required
object_prefix str

Object Prefix. e.g MP

 required
Source code in src/pheval/infra/exomiserdb.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def import_from_semsim_file(self, input_file: Path, subject_prefix: str, object_prefix: str):
    """imports semsim tsv profile into exomiser phenotype database

    Args:
        input_file (Path): semsim profile
        subject_prefix (str): Subject Prefix. e.g HP
        object_prefix (str): Object Prefix. e.g MP
    """
    with self.connector as cnn:
        conn = DBConnection(cnn)
        reader = pl.read_csv_batched(input_file, separator="\t")
        batch_length = 5
        batches = reader.next_batches(batch_length)
        cursor = conn.get_cursor()
        # # TODO: Refactor this
        with open(input_file) as f:
            total = sum(1 for line in f)
        pbar = tqdm(total=total - 1)
        mapping_id = 1
        while batches:
            input_data = pl.concat(batches)
            sql = _semsim2h2(input_data, object_prefix, subject_prefix, mapping_id=mapping_id)
            cursor.execute(sql)
            len_input_data = len(input_data)
            mapping_id += len_input_data
            pbar.update(len_input_data)

            batches = reader.next_batches(batch_length)