"""Utilities to format SQL with correct parameters.""" from sound_recordings import features from sound_recordings.constants.parameters import ALL_TIME def format_with_permissions_filter( sql, permissions_filter, use_old_format=False, start_date=None, end_date=None, extra_format={}, allow_null_feed=False, ): """Format SQL with permission filters. Args: sql (str): SQL containing a subaccount clause to be filled in permissions_filter (dict): dict containing resources users can access use_old_format (bool): use old format for label id and subaccount id start_date (date | str | None): start of the daterange or ALL_TIME end_date (date | None): end of the daterange Returns: str: SQL formatted with appropriate subaccount information """ clause = _compose_permissions_filter(permissions_filter) if start_date and end_date: if start_date == ALL_TIME: clause += " AND download_activity_date <= :end_date " else: clause += ( " AND (download_activity_date BETWEEN" " :start_date AND :end_date) " ) if use_old_format: clause += " AND feedid IN (:feed_ids) " else: if allow_null_feed: clause += " AND (feed_id IS NULL OR feed_id IN (:feed_ids)) " else: clause += " AND feed_id IN (:feed_ids) " return sql.format(filter_clause=clause, **extra_format) def _compose_permissions_filter(permissions_filter): """Compose product SQL filter.""" if _has_full_access(permissions_filter): return " true " if _has_no_access(permissions_filter): return " false " dim_release_filter = _dim_release_filter(permissions_filter) filters = [] if dim_release_filter: filters.append( """ SELECT product_id FROM dim_release WHERE {dim_release_filter} """.format( dim_release_filter=dim_release_filter ) ) if permissions_filter.get("label_participant_ids"): filters.append( """ SELECT product_id FROM label_participant_participated_in_orchard_product WHERE label_participant_id IN (:label_participant_ids) AND participated_as in ('performer', 'featuring', 'remixer') """ ) if filters: return """ product_id IN ( {filters} ) """.format( filters=" UNION ".join(filters) ) def _has_full_access(permissions_filter): """Check if full access is allowed.""" return ( permissions_filter.get("label_ids") == [] and permissions_filter.get("subaccount_ids") == [] and permissions_filter.get("artist_ids") == [] and permissions_filter.get("label_participant_ids") == [] ) def _has_no_access(permissions_filter): return not any( [ permissions_filter.get("label_ids"), permissions_filter.get("subaccount_ids"), permissions_filter.get("artist_ids"), permissions_filter.get("label_participant_ids"), ] ) def _dim_release_filter(permissions_filter): """Compose SQL filter from dim_product table.""" filters = [] if permissions_filter.get("label_ids"): filters.append("labelid IN (:label_ids)") if permissions_filter.get("subaccount_ids"): filters.append("subaccountid IN (:subaccount_ids)") if permissions_filter.get("artist_ids"): filters.append("artistid IN (:artist_ids)") return " OR ".join(filters)