Notes

import boto3
import gzip
import pandas as pd
from sqlalchemy import create_engine, VARCHAR, DATE, FLOAT

# AWS S3 configuration
S3_bucket_name = 'aa00002695-quanthub-market-data-s3-dev-us-east-1'
s3_path = 'Datastore/Xpressfeed/compustatmarketcap/'
S3_client = boto3.client('s3', verify=False)
batch_size = 100000 # Adjust based on memory availability

# Database configuration
db_user = 'your_db_user'
db_password = 'your_db_password'
db_host = 'your_db_host'
db_port = 'your_db_port'
db_name = 'your_db_name'
table_name = 'temp_table'

# Database connection string
db_url = f'postgresql://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}'
engine = create_engine(db_url)

def list_files(bucket_name, prefix, suffix='.csv.gz'):
print("Fetching list")
response = S3_client.list_objects_v2(Bucket=bucket_name, Prefix=prefix)
if 'Contents' in response:
files = [obj['Key'] for obj in response['Contents'] if obj['Key'].endswith(suffix)]
return files
else:
return []

def create_temp_table(engine, table_name):
with engine.connect() as connection:
connection.execute(f"CREATE TEMP TABLE {table_name} (GVKEY VARCHAR(6), DATADATE DATE, CURCDD VARCHAR, MKVAL DOUBLE PRECISION)")

def load_data_to_temp_table(engine, table_name, file_key, batch_size):
obj = S3_client.get_object(Bucket=S3_bucket_name, Key=file_key)
with gzip.open(obj['Body'], 'rt') as gz:
reader = pd.read_csv(gz, chunksize=batch_size)
for chunk in reader:
chunk.to_sql(table_name, engine, if_exists='append', index=False, dtype={'GVKEY': VARCHAR(6), 'DATADATE': DATE, 'CURCDD': VARCHAR, 'MKVAL': FLOAT})

def main():
print("Starting the file listing process")
files = list_files(S3_bucket_name, s3_path)
print("Files:")
if files:
print(f"CSV.gz files in {S3_bucket_name}/{s3_path}:")
for file in files:
print(file)

create_temp_table(engine, table_name)

# Load data from all files into the temporary table
for file in files:
load_data_to_temp_table(engine, table_name, file, batch_size)
else:
print("No CSV files found")

if __name__ == "__main__":
main()