"""General SQL queries and utility functions.""" from argparse import ArgumentTypeError def _only_decimal(seq): """Check if all values of the sequence are numbers or integers. Args: seq (list(str|int)): list of strings which should be numbers or ints. Returns: bool: True if all values are numbers. """ return all([(type(e) is int) or e.isdecimal() for e in seq]) def _quote(val): """Enclose value in single quotes. Args: val (str): value to quote. Returns: str: single-quoted value. """ return "'{}'".format(val) def sql_upcs_condition_in(upcs, column_name='upc'): """Generate SQL condition for list of UPCs. Args: upcs (list(str)): list of upc values. column_name (str): name of column to use in condition. Returns: str: SQL IN condition if `upcs` contains any elements. Empty string otherwise. Raises: ArgumentTypeError: if upcs contains at least one non-numeric value. """ if upcs: if not _only_decimal(upcs): raise ArgumentTypeError('UPC is not in valid format') return '{col_name} IN ({upc_list}) AND '.format( col_name=column_name, upc_list=', '.join(map(_quote, upcs))) else: return '' def upc_vendor_id_lookup(upcs): """Generate SQL to get UPCs' vendor IDs. Args: upcs (list): UPCs to get vendor IDs of. Returns: str: sql to lookup vendor IDs for specified UPCs. """ sql_template = """ SELECT upc, vendor_id FROM art_relations.vw_product WHERE {upc_in_clause} TRUE LIMIT {limit_clause};""" upc_in_clause = sql_upcs_condition_in(upcs, column_name='upc') limit_clause = str(len(upcs)) sql = sql_template.format( upc_in_clause=upc_in_clause, limit_clause=limit_clause) return sql # Vendors and UPCs for film clients at The Orchard. The excluded vendor IDs in # the SQL are common test vendor accounts. serviced_vendor_id_upc_map_sql = """ SELECT vendor_id, upc FROM vw_product WHERE product_type_id = 3 AND vendor_id NOT IN (25824, 7123);"""