From c08f4ac47ce1bd004c83b6c8754960cd14b81db7 Mon Sep 17 00:00:00 2001 From: deroob Date: Tue, 14 Jul 2026 11:15:18 +0200 Subject: [PATCH] update in sql storage object functions. --- src/eo_processing/utils/storage.py | 72 ++++++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) diff --git a/src/eo_processing/utils/storage.py b/src/eo_processing/utils/storage.py index 740dd57..7730214 100644 --- a/src/eo_processing/utils/storage.py +++ b/src/eo_processing/utils/storage.py @@ -2,6 +2,7 @@ import os import boto3 +import pandas as pd from botocore.exceptions import ClientError from getpass import getpass import geopandas as gpd @@ -978,6 +979,77 @@ def QueryItems(self, table: str, lcolumns: List[str]) -> List[tuple]: return lresults + def AddColumns(self, table: str, column_names: lst(str)) -> None: + """ + Adds a new column to a specified table in the database. The method dynamically constructs + an SQL ALTER TABLE statement to add the column with the specified name, ensuring + compatibility with the existing table structure. If the column already exists, the query + does not raise an error. + + Args: + table (str): The name of the table to which the column should be added. + column_names (str): The names of the new columns. + + Returns: + None + """ + + sql_statement = f"""ALTER TABLE {table} + {',\n'.join([f'ADD COLUMN IF NOT EXISTS "{col.lower()}" numeric' for col in column_names])};""" + + self.GenericQueryWithOUTResult(sql_statement) + + def UploadTrainingPointData(self, table: str, df: pd.DataFrame, column_names: List[str]) -> None: + """ + Uploads data to the specified database table by adding missing columns, handling existing + data updates, and inserting new data points in bulk or individually. + + This method performs the following steps: + 1. Checks and ensures all necessary columns in the table. + 2. Identifies whether existing data points need to be updated or if new points need to be + inserted. + 3. Performs a bulk data insert for new data points if any. + 4. Updates rows individually for existing data points. + + Parameters: + table: str + The name of the database table where the data should be uploaded. + df: pd.DataFrame + A pandas DataFrame containing the data to be uploaded. + column_names: List[str] + A list of column names that need to be added to the table if they do not exist. + + Raises: + None + + Returns: + None + """ + + #First check if all columns are in the table + self.AddColumns(table, column_names) + + #First check which points are new and which are not + sql_statement = f"""SELECT mgrsid10, year + FROM {table}""" + + available_keys = self.GenericQueryWithResult(sql_statement) + + #split according to PK available are not + bulk_df = df[~df.set_index(['MGRSid10', 'year']).index.isin(available_keys)] + line_df = df[df.set_index(['MGRSid10', 'year']).index.isin(available_keys)] + + if len(bulk_df) != 0: + #upload bulk + data = ReadFaker(bulk_df) + self.BulkInsert(table, data, tuple([column.lower() for column in df.columns])) + + for index, row in line_df.iterrows(): + #upload line + self.StatusUpdateTiles(table, (row['MGRSid10'], row['year']), + tuple([column.lower() for column in column_names]), + row[column_names].to_list()) + def StatusUpdateTiles(self, table: str, PK: tuple(str, int), lcolumns: List[str], lmsg: List[str]) -> bool: """ Updates specific columns in a particular database table for a given tile ID with new values.