def join_attributes_desc()

in experiments/google/cloud/ml/applied/attributes/attributes.py [0:0]

• 24 lines of code
• 3 McCabe index (conditional complexity)

def join_attributes_desc(ids: list[str]) -> dict[str:dict]:
    """
    Gets the attributes and description for given product IDs.

    Args:
        ids: The product IDs to get the attributes for.

    Returns
        dict mapping product IDs to attributes and descriptions. Each ID will
        map to a dict with the following keys:
            attributes: e.g. {'color':'green', 'pattern': striped}
            description: e.g. 'This is a description'
    """
    bq = Config.SECTION_BIG_QUERY
    table_name = Config.value(bq, "product_table")
    column_id = Config.value(bq, "product_id_column")
    column_attributes = Config.value(bq, "product_attributes_column")
    column_description = Config.value(bq, "product_description_column")

    query = f"""
    SELECT
        {column_id},
        {column_attributes},
        {column_description}
    FROM
        `{table_name}`
    WHERE
        {column_id} IN {str(ids).replace('[', '(').replace(']', ')')}
    """
    query_job = bq_client.query(query)
    rows = query_job.result()
    attributes = {}
    for row in rows:
        id_value = row[column_id]
        logging.info(row)
        if isinstance(row[column_attributes], str):
            attributes[id_value] = {
                "attributes": json.loads(row[column_attributes]),
                "description": row[column_description],
            }
        else:
            attributes[id_value] = {
                "attributes": {},
                "description": row[column_description],
            }

    return attributes