Source code for tkp.testutil.db_queries

"""
A collection of back end db query subroutines used for unittesting
"""
from tkp.db import execute
from tkp.db.generic import get_db_rows_as_dicts


def dataset_images(dataset_id, database=None):
    q = "SELECT id FROM image WHERE dataset=%(dataset)s LIMIT 1"
    args = {'dataset': dataset_id}
    cursor = execute(q, args)
    image_ids = [x[0] for x in cursor.fetchall()]
    return image_ids


def convert_to_cartesian(conn, ra, decl):
[docs] """Returns tuple (x,y,z)""" qry = """SELECT x,y,z FROM cartesian(%s, %s)""" curs = conn.cursor() curs.execute(qry, (ra, decl)) return curs.fetchone() def get_assoc_entries(db, runcat_id):
[docs] """ Return the full history of variability indices for a runcat entry, ordered by time. """ query = """\ select a.runcat ,a.xtrsrc ,x.extract_type ,i.taustart_ts ,a.v_int ,a.eta_int ,a.f_datapoints ,a.type ,r.mon_src from assocxtrsource a ,extractedsource x ,image i ,runningcatalog r where a.xtrsrc = x.id and x.image = i.id and a.runcat = r.id and r.id = %(runcat_id)s order by i.taustart_ts """ db.cursor.execute(query, {'runcat_id': runcat_id}) return get_db_rows_as_dicts(db.cursor)