Merge pull request #1053 from devinit/feature/update-pandas

Pin Pandas version and move back to native to_sql

.github/workflows/CI.yml

@@ -53,10 +53,10 @@ jobs:
 
     steps:
     - uses: actions/checkout@v4
-    - name: Set up Python 3.7
+    - name: Set up Python 3.8
       uses: actions/setup-python@v5
       with:
-        python-version: '3.7'
+        python-version: '3.8'
         architecture: 'x64'
 
     - uses: actions/cache@v4

Dockerfile

@@ -1,5 +1,5 @@
 # start with a base image
-FROM python:3.7
+FROM python:3.8
 LABEL maintainer="akmiller01 <Alex Miller, [email protected]>"
 
 RUN mkdir /src

data_updates/Python/iati_transactions.py

@@ -3,7 +3,7 @@
 import progressbar
 import pandas as pd
 import sqlalchemy
-from sqlalchemy import and_, distinct, create_engine, MetaData, or_, select, Table, text, insert, Column
+from sqlalchemy import and_, distinct, create_engine, MetaData, or_, select, Table, text, Column
 from lxml import etree
 from lxml.etree import XMLParser
 from iati_transaction_spec import IatiFlat, A_DTYPES, A_NUMERIC_DTYPES, T_DTYPES, T_NUMERIC_DTYPES
@@ -13,7 +13,6 @@
 from requests.packages.urllib3.exceptions import InsecureRequestWarning
 import boto3
 from datetime import datetime
-from sql_utils import batch_generator, dataframe_records_gen
 
 
 current_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@@ -181,15 +180,8 @@ def main(args):
             flat_activity_data[numeric_column] = pd.to_numeric(flat_activity_data[numeric_column], errors='coerce')
         flat_activity_data = flat_activity_data.astype(dtype=A_DTYPES)
 
-        flat_activity_data_records = dataframe_records_gen(flat_activity_data)
-        flat_activity_data_batches_generator = batch_generator(flat_activity_data_records)
         with engine.begin() as conn:
-            for flat_activity_data_batch in flat_activity_data_batches_generator:
-                conn.execute(
-                    insert(tmp_activity_table).values(
-                        flat_activity_data_batch
-                    )
-                )
+            flat_activity_data.to_sql(name=TMP_ACTIVITY_DATA_TABLENAME, con=conn, schema=DATA_SCHEMA, index=False, if_exists="append")
 
         if not flat_transactions:
             continue
@@ -201,16 +193,8 @@ def main(args):
             flat_transaction_data[numeric_column] = pd.to_numeric(flat_transaction_data[numeric_column], errors='coerce')
         flat_transaction_data = flat_transaction_data.astype(dtype=T_DTYPES)
 
-        flat_transaction_data_records = dataframe_records_gen(flat_transaction_data)
-        flat_transaction_data_batches_generator = batch_generator(flat_transaction_data_records)
         with engine.begin() as conn:
-            for flat_transaction_data_batch in flat_transaction_data_batches_generator:
-                conn.execute(
-                    insert(tmp_transaction_table).values(
-                        flat_transaction_data_batch
-                    )
-                )
-
+            flat_transaction_data.to_sql(name=TMP_DATA_TABLENAME, con=conn, schema=DATA_SCHEMA, index=False, if_exists="append")
 
     # Delete repeats, insert tmp into permanent, erase tmp
     with engine.begin() as conn:

data_updates/Python/iati_transactions_reconfigure.py

@@ -9,7 +9,6 @@
 import boto3
 from botocore.exceptions import ClientError
 from datetime import datetime
-from sql_utils import batch_generator, dataframe_records_gen
 
 
 current_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@@ -99,15 +98,8 @@ def main():
         for numeric_column in A_NUMERIC_DTYPES:
             flat_activity_data[numeric_column] = pd.to_numeric(flat_activity_data[numeric_column], errors='coerce')
         flat_activity_data = flat_activity_data.astype(dtype=A_DTYPES)
-        flat_activity_data_records = dataframe_records_gen(flat_activity_data)
-        flat_activity_data_batches_generator = batch_generator(flat_activity_data_records)
         with engine.begin() as conn:
-            for flat_activity_data_batch in flat_activity_data_batches_generator:
-                conn.execute(
-                    insert(activity_table).values(
-                        flat_activity_data_batch
-                    )
-                )
+            flat_activity_data.to_sql(name=ACTIVITY_DATA_TABLENAME, con=conn, schema=DATA_SCHEMA, index=False, if_exists="append")
 
         if not flat_transactions:
             continue
@@ -118,15 +110,8 @@ def main():
         for numeric_column in T_NUMERIC_DTYPES:
             flat_transaction_data[numeric_column] = pd.to_numeric(flat_transaction_data[numeric_column], errors='coerce')
         flat_transaction_data = flat_transaction_data.astype(dtype=T_DTYPES)
-        flat_transaction_data_records = dataframe_records_gen(flat_transaction_data)
-        flat_transaction_data_batches_generator = batch_generator(flat_transaction_data_records)
         with engine.begin() as conn:
-            for flat_transaction_data_batch in flat_transaction_data_batches_generator:
-                conn.execute(
-                    insert(transaction_table).values(
-                        flat_transaction_data_batch
-                    )
-                )
+            flat_transaction_data.to_sql(name=DATA_TABLENAME, con=conn, schema=DATA_SCHEMA, index=False, if_exists="append")
 
 
 if __name__ == '__main__':

data_updates/Python/manual_data.py

@@ -3,7 +3,6 @@
 import glob
 import os
 import sys
-from sql_utils import df_to_sql
 
 
 def main(filename=None):
@@ -29,7 +28,8 @@ def process_file(filename):
             csv_dat = pd.read_csv(filename, keep_default_na=False, na_values=[''])
         except UnicodeDecodeError:
             csv_dat = pd.read_csv(filename, keep_default_na=False, na_values=[''], encoding='latin1')
-        df_to_sql(csv_dat, engine, table_name, "repo", "replace")
+        with engine.begin() as conn:
+            csv_dat.to_sql(name=table_name, con=conn, schema="repo", index=False, if_exists="replace")
 
 
 if __name__ == '__main__':

data_updates/Python/manual_data_fts.py

@@ -2,7 +2,6 @@
 from sqlalchemy import create_engine
 import glob
 import os
-from sql_utils import df_to_sql
 
 
 def main():
@@ -18,7 +17,8 @@ def main():
         table_name = os.path.splitext(basename)[0]
         if table_name not in ["meta", "meta_columns"]:
             csv_dat = pd.read_csv(abs_csv_path, keep_default_na=False, na_values=[''], encoding='latin1')
-            df_to_sql(csv_dat, engine, table_name, "repo", "replace")
+            with engine.begin() as conn:
+                csv_dat.to_sql(name=table_name, con=conn, schema="repo", index=False, if_exists="replace")
 
 
 if __name__ == '__main__':

data_updates/Python/povcal_agg.py

@@ -6,7 +6,6 @@
 import numpy as np
 from requests.adapters import HTTPAdapter
 from requests.packages.urllib3.util.retry import Retry
-from sql_utils import df_to_sql
 
 
 def requests_retry_session(
@@ -69,7 +68,8 @@ def fetch_and_write_full_data(schema_name, table_name, engine):
     append_or_replace = "replace"
     for povline in progressbar.progressbar(np.linspace(0.01, 10, 1000)):
         pov_data = fetch_data(poverty_line=povline)
-        df_to_sql(pov_data, engine, "PovCalNetAgg", "repo", append_or_replace)
+        with engine.connect() as conn:
+            pov_data.to_sql(name="PovCalNetAgg", con=conn, schema="repo", index=False, if_exists=append_or_replace)
         append_or_replace = "append"
 
 

data_updates/Python/povcal_p20.py

@@ -1,6 +1,5 @@
 import pandas as pd
 from sqlalchemy import create_engine
-from sql_utils import df_to_sql
 
 
 def main():
@@ -21,7 +20,8 @@ def main():
             pass
 
     p20_data = pd.concat(p20_data_list, ignore_index=True)
-    df_to_sql(p20_data, engine, "PovCalNetP20", "repo", "replace")
+    with engine.connect() as conn:
+        p20_data.to_sql(name="PovCalNetP20", con=conn, schema="repo", index=False, if_exists="replace")
 
 
 if __name__ == '__main__':

data_updates/Python/povcal_smy.py

@@ -6,7 +6,6 @@
 import numpy as np
 from requests.adapters import HTTPAdapter
 from requests.packages.urllib3.util.retry import Retry
-from sql_utils import df_to_sql
 
 
 def requests_retry_session(
@@ -70,7 +69,8 @@ def fetch_and_write_full_data(schema_name, table_name, engine):
     append_or_replace = "replace"
     for povline in progressbar.progressbar(np.linspace(0.01, 10, 1000)):
         pov_data = fetch_data(poverty_line=povline)
-        df_to_sql(pov_data, engine, "PovCalNetSmy", "repo", append_or_replace)
+        with engine.connect() as conn:
+            pov_data.to_sql(name="PovCalNetSmy", con=conn, schema="repo", index=False, if_exists=append_or_replace)
         append_or_replace = "append"
 
 

requirements.txt

@@ -13,7 +13,7 @@ importlib-metadata==4.13.0
 lxml>=4.6.3,<6.0
 markdown>=3.2.2,<4.0
 numpy>=1.18.4,<2.0
-pandas>=1.0.3,<3.0
+pandas>=2.0.3,<3.0
 progressbar2>=3.39.3,<4.0
 psycopg2>=2.8.5,<3.0
 psycopg2-binary>=2.8.5,<3.0