1"""
  2Contains the utility functions for the load lambda_handler.
  3"""
  4
  5import awswrangler as wr
  6import boto3
  7
  8from src.utils.db_connection import close_conn, create_conn
  9
 10
 11def access_files_from_processed_bucket(table_name, bucket_name):
 12    """
 13    This function connects to the s3 bucket with transformed data and returns all the information in the given table as a dataframe.
 14
 15    # Arguments:
 16        table_name: a string representing the name of the table in the database that we want to extract.
 17        bucket_name: a string representing the name of the s3 bucket with transformed data to connect to.
 18
 19    # Returns:
 20        A dataframe containing the most recent data for the given table in the bucket.
 21
 22    # Raises:
 23        RuntimeError: An error occurred during data retrieval.
 24    """
 25    try:
 26        client = boto3.client("s3")
 27        listing_response = client.list_objects_v2(Bucket=bucket_name, Prefix=table_name)
 28        object_key = listing_response["Contents"][-1]["Key"]
 29        path = f"s3://{bucket_name}/{object_key}"
 30        response = wr.s3.read_parquet([path])
 31        return response
 32
 33    except Exception as e:
 34        raise RuntimeError(f"Retrieval of data from processed bucket failed: {e}")
 35
 36
 37def load_dim_dates_into_warehouse(df):
 38    """
 39    Loads data from a dataframe into the warehouse's dim_dates table.
 40
 41    # Arguments:
 42        df: a dataframe representing the contents of the dim_dates table.
 43
 44    # Returns:
 45        None.
 46    """
 47    conn = create_conn()
 48    if conn:
 49        try:
 50            query = """
 51            INSERT INTO dim_date
 52            VALUES
 53            """
 54
 55            for _, row in df.iterrows():
 56                query += f"('{row['date_id']}', {row['year']}, {row['month']}, {row['day']}, {row['day_of_week']}, '{row['day_name']}', '{row['month_name']}', {row['quarter']}), "
 57            query = query[:-2]
 58            query += "ON CONFLICT (date_id) DO NOTHING;"
 59
 60            conn.run(query)
 61
 62        except Exception as e:
 63            raise RuntimeError(f"Database query failed: {e}")
 64
 65        finally:
 66            close_conn(conn)
 67
 68
 69def load_dim_staff_into_warehouse(df):
 70    """
 71    Loads data from a dataframe into the warehouse's dim_staff table.
 72
 73    # Arguments:
 74        df: a dataframe representing the contents of the dim_staff table.
 75
 76    # Returns:
 77        None.
 78    """
 79    conn = create_conn()
 80    if conn:
 81        try:
 82            query = """
 83            INSERT INTO dim_staff
 84            VALUES
 85            """
 86
 87            for _, row in df.iterrows():
 88                query += f"({row['staff_id']}, $${row['first_name']}$$, $${row['last_name']}$$, '{row['department_name']}', '{row['location']}', $${row['email_address']}$$), "
 89            query = query[:-2]
 90            query += " ON CONFLICT (staff_id) DO NOTHING;"
 91
 92            conn.run(query)
 93
 94        except Exception as e:
 95            raise RuntimeError(f"Database query failed: {e}")
 96
 97        finally:
 98            close_conn(conn)
 99
100
101def load_dim_location_into_warehouse(df):
102    """
103    Loads data from a dataframe into the warehouse's dim_location table.
104
105    # Arguments:
106        df: a dataframe representing the contents of the dim_location table.
107
108    # Returns:
109        None.
110    """
111    conn = create_conn()
112    if conn:
113        try:
114            query = """
115            INSERT INTO dim_location
116            VALUES
117            """
118
119            for _, row in df.iterrows():
120                query += f"({row['location_id']}, $${row['address_line_1']}$$, $${row['address_line_2']}$$, $${row['district']}$$, $${row['city']}$$, $${row['postal_code']}$$, $${row['country']}$$, $${row['phone']}$$), "
121            query = query[:-2]
122            query += " ON CONFLICT (location_id) DO NOTHING;"
123
124            conn.run(query)
125
126        except Exception as e:
127            raise RuntimeError(f"Database query failed: {e}")
128
129        finally:
130            close_conn(conn)
131
132
133def load_dim_currency_into_warehouse(df):
134    """
135    Loads data from a dataframe into the warehouse's dim_currency table.
136
137    # Arguments:
138        df: a dataframe representing the contents of the dim_currency table.
139
140    # Returns:
141        None.
142    """
143    conn = create_conn()
144    if conn:
145        try:
146            query = """
147            INSERT INTO dim_currency
148            VALUES
149            """
150
151            for _, row in df.iterrows():
152                query += f"({row['currency_id']}, $${row['currency_code']}$$, $${row['currency_name']}$$), "
153            query = query[:-2]
154            query += " ON CONFLICT (currency_id) DO NOTHING;"
155
156            conn.run(query)
157
158        except Exception as e:
159            raise RuntimeError(f"Database query failed: {e}")
160
161        finally:
162            close_conn(conn)
163
164
165def load_dim_design_into_warehouse(df):
166    """
167    Loads data from a dataframe into the warehouse's dim_design table.
168
169    # Arguments:
170        df: a dataframe representing the contents of the dim_design table.
171
172    # Returns:
173        None.
174    """
175    conn = create_conn()
176    if conn:
177        try:
178            query = """
179            INSERT INTO dim_design
180            VALUES
181            """
182
183            for _, row in df.iterrows():
184                query += f"({row['design_id']}, $${row['design_name']}$$, $${row['file_location']}$$, $${row['file_name']}$$), "
185            query = query[:-2]
186            query += " ON CONFLICT (design_id) DO NOTHING;"
187
188            conn.run(query)
189
190        except Exception as e:
191            raise RuntimeError(f"Database query failed: {e}")
192
193        finally:
194            close_conn(conn)
195
196
197def load_dim_counterparty_into_warehouse(df):
198    """
199    Loads data from a dataframe into the warehouse's dim_counterparty table.
200
201    # Arguments:
202        df: a dataframe representing the contents of the dim_counterparty table.
203
204    # Returns:
205        None.
206    """
207    conn = create_conn()
208    if conn:
209        try:
210            query = """
211            INSERT INTO dim_counterparty
212            VALUES
213            """
214
215            for _, row in df.iterrows():
216                query += f"({row['counterparty_id']}, $${row['counterparty_legal_name']}$$, $${row['counterparty_legal_address_line_1']}$$, $${row['counterparty_legal_address_line_2']}$$, $${row['counterparty_legal_district']}$$, $${row['counterparty_legal_city']}$$, $${row['counterparty_legal_postal_code']}$$, $${row['counterparty_legal_country']}$$, $${row['counterparty_legal_phone']}$$), "
217            query = query[:-2]
218            query += " ON CONFLICT (counterparty_id) DO NOTHING;"
219
220            conn.run(query)
221
222        except Exception as e:
223            raise RuntimeError(f"Database query failed: {e}")
224
225        finally:
226            close_conn(conn)
227
228
229def load_fact_sales_order_into_warehouse(df):
230    """
231    Loads data from a dataframe into the warehouse's fact_sales_order table. Updates to a sales order are stored as new rows, without overwriting previous data.
232
233    # Arguments:
234        df: a dataframe representing the contents of the fact_sales_order table.
235
236    # Returns:
237        None.
238    """
239    conn = create_conn()
240    if conn:
241        try:
242            temp_create_query = """
243            CREATE TEMP TABLE temp_sales_order (
244                sales_record_id SERIAL PRIMARY KEY NOT NULL,
245                sales_order_id INT NOT NULL,
246                created_date DATE NOT NULL,
247                created_time TIME NOT NULL DEFAULT CURRENT_TIME,
248                last_updated_date DATE NOT NULL,
249                last_updated_time TIME NOT NULL DEFAULT CURRENT_TIME,
250                sales_staff_id INT NOT NULL,
251                counterparty_id INT NOT NULL,
252                units_sold INT NOT NULL,
253                unit_price NUMERIC(10, 2) NOT NULL,
254                currency_id INT NOT NULL,
255                design_id INT NOT NULL,
256                agreed_payment_date DATE NOT NULL,
257                agreed_delivery_date DATE NOT NULL,
258                agreed_delivery_location_id INT NOT NULL
259                );
260            """
261
262            temp_insert_query = """
263            INSERT INTO temp_sales_order
264            (sales_order_id, created_date, created_time, last_updated_date, last_updated_time, sales_staff_id, counterparty_id, units_sold, unit_price, currency_id, design_id, agreed_payment_date, agreed_delivery_date, agreed_delivery_location_id)
265            VALUES
266            """
267
268            for _, row in df.iterrows():
269                temp_insert_query += f"({row['sales_order_id']}, '{row['created_date']}', $${row['created_time']}$$, '{row['last_updated_date']}', $${row['last_updated_time']}$$, {row['sales_staff_id']}, {row['counterparty_id']}, {row['units_sold']}, {row['unit_price']}, {row['currency_id']}, {row['design_id']}, '{row['agreed_payment_date']}', '{row['agreed_delivery_date']}', {row['agreed_delivery_location_id']}), "
270
271            temp_insert_query = temp_insert_query[:-2]
272            temp_insert_query += ";"
273
274            query = """
275            INSERT INTO fact_sales_order
276            (sales_order_id, created_date, created_time, last_updated_date, last_updated_time, sales_staff_id, counterparty_id, units_sold, unit_price, currency_id, design_id, agreed_payment_date, agreed_delivery_date, agreed_delivery_location_id)
277
278            SELECT
279            sales_order_id, created_date, created_time, last_updated_date, last_updated_time, sales_staff_id, counterparty_id, units_sold, unit_price, currency_id, design_id, agreed_payment_date, agreed_delivery_date, agreed_delivery_location_id
280            FROM temp_sales_order
281
282            WHERE
283            (sales_order_id, created_date, created_time, last_updated_date, last_updated_time, sales_staff_id, counterparty_id, units_sold, unit_price, currency_id, design_id, agreed_payment_date, agreed_delivery_date, agreed_delivery_location_id)
284
285            NOT IN
286            (SELECT
287            sales_order_id, created_date, created_time, last_updated_date, last_updated_time, sales_staff_id, counterparty_id, units_sold, unit_price, currency_id, design_id, agreed_payment_date, agreed_delivery_date, agreed_delivery_location_id
288            FROM fact_sales_order)
289            ;
290            """
291            conn.run(temp_create_query)
292            conn.run(temp_insert_query)
293            conn.run(query)
294
295        except Exception as e:
296            raise RuntimeError(f"Database query failed: {e}")
297
298        finally:
299            close_conn(conn)