Format Python Files

Author: José Mendes

Airflow/dags/minio-dag.py

@@ -10,7 +10,9 @@
 import pkg_resources
 import logging
 
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logging.basicConfig(level=logging.INFO,
+                    format='%(asctime)s - %(levelname)s - %(message)s')
+
 
 def check_for_new_files():
     """
@@ -32,6 +34,7 @@ def check_for_new_files():
     logging.info(f"New files detected: {[record[1] for record in records]}")
     return records
 
+
 def process_files(**kwargs):
     """
     Processes the files detected from the database and marks them as processed.
@@ -47,7 +50,7 @@ def process_files(**kwargs):
     for record in records:
         key, value = record
         logging.info(f"Processing file: {key}, uploaded at {value}")
-        
+
         bucketName, fileName = key.split('/', 1)
         logging.info(f"{bucketName},{fileName}")
 
@@ -62,17 +65,22 @@ def process_files(**kwargs):
         pg_hook.run(update_sql)
 
 # Function to install a library if not already installed
+
+
 def install(package, version="7.0.3"):
     try:
         # Check if the package is installed
         pkg_resources.get_distribution(package)
         logging.info(f"{package} is already installed.")
     except pkg_resources.DistributionNotFound:
         logging.info(f"{package} not found. Installing version {version}...")
-        subprocess.check_call([sys.executable, "-m", "pip", "install", f"{package}=={version}"])
+        subprocess.check_call(
+            [sys.executable, "-m", "pip", "install", f"{package}=={version}"])
+
 
 def fileProcessor(bucket_name, object_name):
-    install('minio', version="7.0.3")  # Install compatible Minio version if needed
+    # Install compatible Minio version if needed
+    install('minio', version="7.0.3")
     from minio import Minio
     from minio.error import S3Error
 
@@ -97,7 +105,8 @@ def fileProcessor(bucket_name, object_name):
     try:
         # Download the object from MinIO to a local file
         minio_client.fget_object(bucket_name, object_name, local_file_path)
-        logging.info(f"File '{object_name}' successfully downloaded from MinIO.")
+        logging.info(
+            f"File '{object_name}' successfully downloaded from MinIO.")
 
         # Process the file here
         sendProcessedInfoToDb(local_file_path)
@@ -111,13 +120,15 @@ def fileProcessor(bucket_name, object_name):
     except Exception as e:
         logging.info(f"Unexpected error: {e}")
 
+
 def sendProcessedInfoToDb(localFilePath):
     dataCoffeePrice = pd.read_csv(localFilePath)
 
-    coffeePrice = dataCoffeePrice ["value"]
-    atTime = dataCoffeePrice ["date"]
+    coffeePrice = dataCoffeePrice["value"]
+    atTime = dataCoffeePrice["date"]
 
-    logging.info(f"The coffee price is {coffeePrice.iloc[0]} at date {atTime.iloc[0]}.")
+    logging.info(
+        f"The coffee price is {coffeePrice.iloc[0]} at date {atTime.iloc[0]}.")
 
     pg_hook = PostgresHook(postgres_conn_id='postgres')
 
@@ -130,7 +141,8 @@ def sendProcessedInfoToDb(localFilePath):
     pg_hook.run(insertSql)
 
     logging.info("Newest price instance added to the coffee_price table")
-    
+
+
 # Define default arguments for the DAG
 default_args = {
     'owner': 'your_name',
@@ -165,5 +177,3 @@ def sendProcessedInfoToDb(localFilePath):
 
 # Define task dependencies
 check_files_task >> process_files_task
-
-

src/download_from_minio.py

@@ -1,6 +1,7 @@
 from minio import Minio
 from minio.error import S3Error
 
+
 def download_from_minio(bucket_name, object_name):
     minio_client = Minio(
         "minio:9000",  # MinIO server address
@@ -21,4 +22,4 @@ def download_from_minio(bucket_name, object_name):
         minio_client.fget_object(bucket_name, object_name, object_name)
         print(f"File {object_name} successfully downloaded from MinIO.")
     except S3Error as e:
-        print(f"Error occurred: {e}")
+        print(f"Error occurred: {e}")

src/main.py

@@ -15,6 +15,7 @@ def connect_to_postgres():
     print("Connected to Postgres")
     return conn
 
+
 def create_tables(conn):
     cursor = conn.cursor()
 
@@ -43,7 +44,7 @@ def create_tables(conn):
             consumption FLOAT
         );
         """)
-        
+
     cursor.execute("""
         CREATE TABLE IF NOT EXISTS coffee_production (
             id SERIAL PRIMARY KEY,
@@ -52,16 +53,17 @@ def create_tables(conn):
             production FLOAT
         );
         """)
-    
+
     cursor.execute("""
         CREATE TABLE IF NOT EXISTS coffee_price(
             date DATE,
             price REAL
         );
         """)
-    
+
     conn.commit()
 
+
 def populate_tables(conn, filenames):
 
     cursor = conn.cursor()
@@ -70,7 +72,8 @@ def populate_tables(conn, filenames):
     dataFrameProduction = pd.read_csv(filenames[1])
     country_number = 0
     for country in dataFrameComsumption["Country"]:
-        cursor.execute("INSERT INTO countries (country, coffeeType) VALUES (%s, %s) ON CONFLICT (country) DO NOTHING  RETURNING country_id", (country, dataFrameComsumption["Coffee type"][country_number]))
+        cursor.execute("INSERT INTO countries (country, coffeeType) VALUES (%s, %s) ON CONFLICT (country) DO NOTHING  RETURNING country_id",
+                       (country, dataFrameComsumption["Coffee type"][country_number]))
 
         result = cursor.fetchone()  # Get the inserted country_id
         if result is None:
@@ -81,26 +84,29 @@ def populate_tables(conn, filenames):
         country_id = result[0]
 
         years = dataFrameComsumption.columns[2:-1]
-        
+
         for year in years:
-            cursor.execute ("INSERT INTO coffee_domestic_consumption (country_id, year, consumption) VALUES (%s, %s, %s)", (country_id , year, int(dataFrameComsumption[year][country_number])))
+            cursor.execute("INSERT INTO coffee_domestic_consumption (country_id, year, consumption) VALUES (%s, %s, %s)",
+                           (country_id, year, int(dataFrameComsumption[year][country_number])))
 
-            cursor.execute ("INSERT INTO coffee_production (country_id, year, production) VALUES (%s, %s, %s)", (country_id , year, int(dataFrameProduction[year][country_number])))
+            cursor.execute("INSERT INTO coffee_production (country_id, year, production) VALUES (%s, %s, %s)",
+                           (country_id, year, int(dataFrameProduction[year][country_number])))
 
         country_number += 1
-            
+
     conn.commit()
 
+
 if __name__ == "__main__":
 
-    fileNames =["Coffee_domestic_consumption.csv", "Coffee_production.csv"]
-    
+    fileNames = ["Coffee_domestic_consumption.csv", "Coffee_production.csv"]
+
     for fileName in fileNames:
         if not os.path.exists(fileName):
-            download_from_minio("coffee-dataset-example", fileName)  # Update with your MinIO bucket name
-    
+            # Update with your MinIO bucket name
+            download_from_minio("coffee-dataset-example", fileName)
 
     connection = connect_to_postgres()
     create_tables(connection)
-    
+
     populate_tables(connection, fileNames)