Source code for allensdk.internal.api.queries.utils

from typing import Optional, List
from allensdk.core.typing import SupportsStr
from uuid import UUID


def build_in_list_selector_query(
        col: str,
        valid_list: Optional[List[SupportsStr]] = None,
        operator: str = "WHERE",
        valid: bool = True) -> str:
    """
    Filter for rows where the value of a column is contained in a list
    (or, if valid=False, where the value is not contained in that list).
    If no list is specified in `valid_list`, return an empty string.

    Parameters
    ----------
    col: str
        The name of the column being filtered on

    valid_list: Optional[SupportsStr]
        The list of values to test column on

    operator: str
        The SQL operator that starts the clause ("WHERE", "AND" or "OR")

    valid: bool
        If True, test for "col IN valid_list"; else, test for
        "col NOT IN valid_list"

    Returns
    -------
    session_query: str
        The clause performing the request filter
    """
    if operator not in ("AND", "OR", "WHERE"):
        msg = ("Operator must be 'AND', 'OR', or 'WHERE'; "
               f"you gave '{operator}'")
        raise ValueError(msg)

    if not valid_list:
        return ""

    if type(valid_list[0]) is str:
        valid_list = _convert_list_of_string_to_sql_safe_string(
            strings=valid_list)

    if valid:
        relation = "IN"
    else:
        relation = "NOT IN"

    session_query = (
        f"""{operator} {col} {relation} ({",".join(
            sorted(set(map(str, valid_list))))})""")
    return session_query


def build_where_clause(clauses: List[str]):
    if not clauses:
        return ''
    where_clause = ' AND '.join(clauses)
    if not where_clause[:5].lower() == 'where':
        where_clause = f'WHERE {where_clause}'
    return where_clause


def _sanitize_uuid_list(uuid_list: List[str]) -> List[str]:
    """
    Loop over a list of strings, removing any that cannot be cast
    into a valid UUID

    Parameters
    ----------
    uuid_list: List[str]
        List of strings that would ideally be cast into
        UUIDs

    Returns
    -------
    sanitized_list: List[str]
        A list containing all of the elements from uuid_list
        that could successfully be cast into a UUID

    Note
    ----
    This method is meant to be used as a pre-processing step
    for queries to MTRAIN. foraging_id values need to be valid
    string representations of UUIDs.
    """
    sanitized_list = []
    for val in uuid_list:
        try:
            UUID(val)
            sanitized_list.append(val)
        except ValueError:
            pass
    return sanitized_list


def _convert_list_of_string_to_sql_safe_string(
        strings: List[str]
) -> List[str]:
    """
    Given list of string ["A", "B"]
    converts to ["'A'", "'B'"]
    Parameters
    ----------
    strings: list of strings to convert

    Returns
    -------
    List of sql-safe strings
    """
    if len(strings) == 0:
        return strings
    if len(strings[0]) == 0:
        return strings

    # If the first element doesn't start with single quote, assume none of the
    # elements in the list do
    if strings[0][0] != "'":
        # Add single quotes to each element
        strings = [f"'{x}'" for x in strings]
    return strings