Source code for dvha.db.sql_to_python

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# db.sql_to_python.py
"""Query a DVHA SQL table and parse the return into a python object"""
# Copyright (c) 2016-2019 Dan Cutright
# This file is part of DVH Analytics, released under a BSD license.
#    See the file LICENSE included with this distribution, also
#    available at https://github.com/cutright/DVH-Analytics

from dvha.db.sql_connector import DVH_SQL
from dateutil.parser import parse as date_parser
from dvha.tools.errors import push_to_log


[docs]class QuerySQL: """Object to generically query a specified table. Each column is stored as a property of the object For example, if you query 'dvhs' with condition string of "mrn = 'some_mrn'" you can access any column name 'some_column' with QuerySQL.some_column which will return a list of values for 'some_column'. All properties contain lists with the order of their values synced, unless unique=True Parameters ---------- table_name : str Beams', 'DVHs', 'Plans', or 'Rxs' condition_str : str condition in SQL syntax unique : bool, optional If True, only unique values stored columns : list, optional A list of SQL column names to be included in the return. If left as ``None``, all columns will be returned group : int, optional either 1 or 2 """ def __init__( self, table_name, condition_str, unique=False, columns=None, group=1 ): table_name = table_name.lower() if table_name in {"beams", "dvhs", "plans", "rxs"}: self.table_name = table_name self.condition_str = condition_str with DVH_SQL(group=group) as cnx: all_columns = cnx.get_column_names(table_name) if columns is not None: columns = set(all_columns).intersection( columns ) # ensure provided columns exist in SQL table else: columns = all_columns for column in columns: if column not in { "roi_coord_string", "distances_to_ptv", }: # ignored for memory since not used here self.cursor = cnx.query( self.table_name, column, self.condition_str ) force_date = cnx.is_sqlite_column_datetime( self.table_name, column ) # returns False for pgsql rtn_list = self.cursor_to_list(force_date=force_date) if unique: rtn_list = get_unique_list(rtn_list) # create property of QuerySQL based on SQL column name setattr(self, column, rtn_list) else: push_to_log( msg="QuerySQL: Table name in valid. Please select from Beams, " "DVHs, Plans, or Rxs." )
[docs] def cursor_to_list(self, force_date=False): """Convert a cursor return into a list of values Parameters ---------- force_date : bool, optional Apply dateutil.parser to values Returns ------- list queried data """ rtn_list = [] for row in self.cursor: if force_date: try: if type(row[0]) is int: rtn_list.append(str(date_parser(str(row[0])))) else: rtn_list.append(str(date_parser(row[0]))) except Exception: rtn_list.append("None") elif isinstance(row[0], (int, float)): rtn_list.append(row[0]) else: rtn_list.append(str(row[0])) return rtn_list
[docs]def get_unique_list(input_list): """Remove duplicates in list and retain order Parameters ---------- input_list : list any list of objects Returns ------- list input_list without duplicates """ rtn_list_unique = [] for value in input_list: if value not in rtn_list_unique: rtn_list_unique.append(value) return rtn_list_unique
[docs]def get_database_tree(): """Query SQL to get all columns of each table Parameters ---------- Returns ------- dict column data sorted by table """ with DVH_SQL() as cnx: tree = {table: cnx.get_column_names(table) for table in cnx.tables} return tree