Notes

def copy_data_to_temp_table(engine, schema_name, table_name, files):
"""Copy data from S3 files to temporary database table using COPY command."""
total_copied = 0
try:
logging.info("Starting to copy data to the database table")
for file_key in files:
logging.info(f"Copying data from {file_key}")
try:
obj = S3_CLIENT.get_object(Bucket=S3_BUCKET_NAME, Key=file_key)
with gzip.open(obj['Body'], 'rt') as gz:
copy_sql = f"COPY {schema_name}.{table_name} FROM stdin WITH CSV HEADER DELIMITER ','"
with engine.connect() as conn:
conn.connection.set_isolation_level(0)
with conn.connection.cursor() as cur:
cur.copy_expert(sql=copy_sql, file=gz)
conn.connection.set_isolation_level(1)
cur.execute(f"SELECT COUNT(*) FROM {schema_name}.{table_name}")
count = cur.fetchone()[0]
logging.info(f"Total records copied from {file_key}: {count}")
total_copied += count
except Exception as e:
logging.error(f"Failed to copy data from {file_key}: {str(e)}")
raise # Raise exception if copying fails
logging.info(f"Total records copied from all files: {total_copied}")
return True
except Exception as e:
logging.error(f"Error copying data to temporary table: {str(e)}")
return False