ClinicalTrials_DataLinkers/MatchingFormularies/formularies_db_interface.py

import psycopg2 as psyco
from psycopg2 import extras
from datetime import datetime

import click #used for cli commands. Not needed for what I am doing.
from flask import current_app, g

def get_db(**kwargs):

    if "db" not in g:
        g.db = psyco.connect(
            dbname=current_app.config["POSTGRES_DB"]
            ,user=current_app.config["POSTGRES_USER"]
            ,host=current_app.config["POSTGRES_HOST"]
            ,port=current_app.config["POSTGRES_PORT"]
            ,password=current_app.config["POSTGRES_PASSWORD"]
            ,**kwargs
        )
    return g.db

def close_db(e=None):
    db = g.pop('db', None)

    if db is not None:
        db.close()

def check_initialization(app):
    db = get_db()
    with db.cursor() as curse:
        curse.execute("select count(*) from \"DiseaseBurden\".trial_to_icd10")
    curse.fetchall()
    #just checking if everything is going to fail

def init_database(app):
    #check_initialization(app)
    app.teardown_appcontext(close_db)




def select_remaing_trials_to_analyze(db_conn):
    '''
    This will get the set of trials that need to be analyzed.
    '''
    sql = '''
    select distinct nct_id 
    from "DiseaseBurden".trial_to_icd10 tti 
    where tti.approved is null
    order by nct_id
    ;
    '''
    with db_conn.cursor() as cursor:
        cursor.execute(sql)
        return cursor.fetchall()


def select_analyzed_trials(db_conn):
    '''
    This will get the set of trials that have been analyzed.
    '''
    sql = '''
    select distinct nct_id, max(approval_timestamp)
    from "DiseaseBurden".trial_to_icd10 tti 
    where tti.approved in ('accepted','rejected')
    group by nct_id
    order by max(approval_timestamp) desc
    ;
    '''
    with db_conn.cursor() as cursor:
        cursor.execute(sql)
        return cursor.fetchall()

def select_unmatched_trials(db_conn):
    '''
    This will get the set of trials that have been analyzed.
    '''
    sql = '''
    select distinct nct_id
    from "DiseaseBurden".trial_to_icd10 tti 
    where tti.approved = 'unmatched'
    order by nct_id
    ;
    '''
    with db_conn.cursor() as cursor:
        cursor.execute(sql)
        return cursor.fetchall()


def get_trial_conditions_and_proposed_matches(db_conn, nct_id):
    sql = '''
        select *
        from "DiseaseBurden".trial_to_icd10 tti 
        where nct_id = %s
        '''
    with db_conn.cursor() as cursor:
        cursor.execute(sql,[nct_id])
        return cursor.fetchall()


def store_validation(db_conn, list_of_insert_data):
    sql = """
    update "DiseaseBurden".trial_to_icd10
    set approved=%s, approval_timestamp=%s
    where id=%s
    ; 
    """
    with db_conn.cursor() as cursor:
        for l in list_of_insert_data:
            cursor.execute(sql, l)
    db_conn.commit()

def get_trial_summary(db_conn,nct_id):
    sql_summary ="""
select 
    s.nct_id,
    brief_title ,
    official_title ,
    bs.description as brief_description,
    dd.description as detailed_description
from ctgov.studies s 
    left join ctgov.brief_summaries bs
        on bs.nct_id = s.nct_id 
    left join ctgov.detailed_descriptions dd 
        on dd.nct_id = s.nct_id 
where s.nct_id = %s
;
"""
    sql_conditions="""
--conditions mentioned
select * from ctgov.conditions c 
where c.nct_id = %s
;
"""
    sql_keywords="""
select nct_id ,downcase_name 
from ctgov.keywords k 
where k.nct_id = %s
;
"""
    with db_conn.cursor() as curse:
        curse.execute(sql_summary,[nct_id])
        summary = curse.fetchall()

        curse.execute(sql_keywords,[nct_id])
        keywords = curse.fetchall()

        curse.execute(sql_conditions,[nct_id])
        conditions = curse.fetchall()

    return {"summary":summary, "keywords":keywords, "conditions":conditions}

def get_list_icd10_codes(db_conn):
    sql = """
    select distinct code 
    from "DiseaseBurden".icd10_to_cause itc 
    order by code;
    """
    with db_conn.cursor() as curse:
        curse.execute(sql)
        codes = curse.fetchall()
    
    return [ x[0] for x in codes ]

def record_suggested_matches(db_conn, nct_id,condition,icd10_code):
    sql1 = """
    INSERT INTO "DiseaseBurden".trial_to_icd10 
    (nct_id,"condition",ui,"source",approved,approval_timestamp)
    VALUES (%s,%s,%s,'hand matched','accepted',%s)
    ;
    """


    with db_conn.cursor() as curse:
        curse.execute(sql1,[nct_id,condition,icd10_code,datetime.now()])
    db_conn.commit()