Source code for bigquery.bigquery_customer_notification_gateway

import logging

from exceptions.task_exception import BigQueryError, BigQueryNoItemsFoundError


class BigQueryCustomerNotificationGateway:

    def __init__(self, client, application_dependencies):
        """
        Class to manage the big query executions for edit order notifications
        Sets the client variable for use within the class

        Args:
            client (google.cloud.bigquery.client.Client): Initialised Bigquery Client Object
            application_dependencies (util.application_dependencies.ApplicationDependencies):
                Application dependency object used for DI
        """
        self.client = client
        self.application_dependencies = application_dependencies

    def get_all_customers_that_have_deliveries_tomorrow(self):
        """
        Method to get the json object from BigQuery and load it into memory as json object

        Returns:
            list: All returned data from the BigQuery table (orderid and retailerCustomerid) in a list

        Example:
            json list returned example:

            .. code-block:: python

                [{'orderId': '1937251515984', 'retailerCustomerId': '1005971340'},
                {'orderId': '1937251515984', 'retailerCustomerId': '1005971340'},
                {'orderId': '1608541365787', 'retailerCustomerId': '1001022062'},
                {'orderId': '1943775911778', 'retailerCustomerId': '1002105834'}]

        Raises:
            BigQueryError: If the client experiences an error when calling Google BigQuery
            BigQueryNoItemsFoundError: If the request to BigQuery returns no items
        """
        logger = self.application_dependencies.get_default_logger()
        data_set = self.application_dependencies.get_cut_off_tomorrow_data_set()
        table = self.application_dependencies.get_cut_off_tomorrow_table()
        query = """SELECT orderId, retailerCustomerId 
                FROM `{data_set}.{table}`""".format(data_set=data_set, table=table)
        try:
            logger.info("Retrieving customers from table using query: {query}".format(query=query))
            query_job = self.client.query(query)
            result_of_query = query_job.result()
        except Exception as e:
            logger.error("An Error occurred while making a query to BigQuery. Error: {error}".format(error=e))
            raise BigQueryError
        return self._validate_data_and_return_records(result_of_query)

    def _validate_data_and_return_records(self, result_of_query):
        """
        Args:
          (google.cloud.bigquery.table.RowIterator): The Row iterator object returned from the query.result() method

        Returns:
            list: All returned data from the BigQuery table (orderid and retailerCustomerid) in a list
        """
        logger = self.application_dependencies.get_default_logger()
        if result_of_query.total_rows == 0:
            logger.error("No items were returned from query")
            raise BigQueryNoItemsFoundError
        list_of_customer_orders = []
        for order_and_customer_id in result_of_query:
            list_of_customer_orders.append(
                # Each order_and_customer_id is a ROW object from google.cloud.table.RowIterator
                # which can be converted to a dictionary
                # Example: Row(('12345678908987', '123445678'), {'orderId': 0, 'retailerCustomerId': 1})
                dict(order_and_customer_id)
            )
        return list_of_customer_orders