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

from typing import List, Dict
from allensdk.internal.api import PostgresQueryMixin
from allensdk.internal.api.queries.utils import (
    build_in_list_selector_query)
from allensdk import OneResultExpectedError


def wkf_path_from_attachable(
        lims_connection: PostgresQueryMixin,
        wkf_type_name: List[str],
        attachable_type: str,
        attachable_id: int) -> Dict[str, str]:
    """
    Get the path to well known files, selecting files of a specific
    type with a specified attachable ID and attachable_type.

    Parameters
    ----------
    lims_connection: PostgresQueryMixin

    wkf_type_name: List[str]
        i.e. 'StimulusPickle' or 'RawEyeTrackingVideoMetadata' etc.

    attachable_type: str
        the value of well_known_file.attachable_type to look for

    attachable_id: int
        the value of well_known_file.attachable_id to look for

    Returns
    -------
    wkf_path_lookup: Dict[str, str]
        a dict mapping attachable_type to absolute file path

    Notes
    -----
    Will raise an error if more than one result is returned
    for a single type
    """

    query = """
    SELECT
      wkft.name as type_name
      ,wkf.storage_directory || wkf.filename as filepath
    FROM
      well_known_files wkf
    JOIN
      well_known_file_types wkft
    ON
      wkf.well_known_file_type_id=wkft.id
    """

    query += build_in_list_selector_query(
                col="wkft.name",
                valid_list=wkf_type_name,
                operator="WHERE",
                valid=True)

    query += build_in_list_selector_query(
                col="wkf.attachable_type",
                valid_list=[f"'{attachable_type}'", ],
                operator="AND",
                valid=True)

    query += build_in_list_selector_query(
                col="wkf.attachable_id",
                valid_list=[f"'{attachable_id}'", ],
                operator="AND",
                valid=True)

    query_result = lims_connection.select(query)

    wkf_path_lookup = dict()

    if len(query_result) == 0:
        return wkf_path_lookup

    for type_name, filepath in zip(query_result.type_name,
                                   query_result.filepath):
        if type_name in wkf_path_lookup:
            raise OneResultExpectedError(
                f"More than one result returned for {type_name}")
        wkf_path_lookup[type_name] = filepath

    return wkf_path_lookup