Update old MySQL packages

integrations/acquisition/covidcast/test_covidcast_meta_caching.py

@@ -5,7 +5,7 @@
 import unittest
 
 # third party
-import mysql.connector
+import MySQLdb
 import requests
 
 # first party
@@ -32,7 +32,7 @@ def setUp(self):
     """Perform per-test setup."""
 
     # connect to the `epidata` database
-    cnx = mysql.connector.connect(
+    cnx = MySQLdb.connect(
         user='user',
         password='pass',
         host='delphi_database_epidata',
@@ -58,7 +58,7 @@ def setUp(self):
     secrets.db.host = 'delphi_database_epidata'
     secrets.db.epi = ('user', 'pass')
 
-    epidata_cnx = mysql.connector.connect(
+    epidata_cnx = MySQLdb.connect(
         user='user',
         password='pass',
         host='delphi_database_epidata',

integrations/acquisition/covidcast/test_csv_uploading.py

@@ -7,7 +7,7 @@
 import argparse
 
 # third party
-import mysql.connector
+import MySQLdb
 import pandas as pd
 import numpy as np
 
@@ -28,7 +28,7 @@ def setUp(self):
     """Perform per-test setup."""
 
     # connect to the `epidata` database and clear the `covidcast` table
-    cnx = mysql.connector.connect(
+    cnx = MySQLdb.connect(
         user='user',
         password='pass',
         host='delphi_database_epidata',
@@ -55,7 +55,7 @@ def setUp(self):
     secrets.db.host = 'delphi_database_epidata'
     secrets.db.epi = ('user', 'pass')
 
-    epidata_cnx = mysql.connector.connect(
+    epidata_cnx = MySQLdb.connect(
         user='user',
         password='pass',
         host='delphi_database_epidata',

integrations/acquisition/covidcast/test_db.py

@@ -23,7 +23,8 @@ def _find_matches_for_row(self, row):
             cur.execute(q)
             res = cur.fetchone()
             if res:
-                results[table] = dict(zip(cur.column_names, res))
+                # Create a dictionary of column names (from cursor.description) & values
+                results[table] = dict(zip([desc[0] for desc in cur.description], res))
             else:
                 results[table] = None
         return results

integrations/server/test_covidcast.py

@@ -5,7 +5,7 @@
 import unittest
 
 # third party
-import mysql.connector
+import MySQLdb
 
 # first party
 from delphi_utils import Nans
@@ -369,7 +369,7 @@ def test_unique_key_constraint(self):
     row = self._insert_placeholder_set_one()
 
     # fail to insert different placeholder data under the same key
-    with self.assertRaises(mysql.connector.errors.IntegrityError):
+    with self.assertRaises(MySQLdb.errors.IntegrityError):
       self._insert_placeholder_set_one()
 
     # succeed to insert different placeholder data under a different time_type

integrations/server/test_covidcast_meta.py

@@ -4,7 +4,7 @@
 import unittest
 
 # third party
-import mysql.connector
+import MySQLdb
 import requests
 
 #first party
@@ -53,7 +53,7 @@ def localSetUp(self):
     """Perform per-test setup."""
 
     # connect to the `epidata` database and clear the `covidcast` table
-    cnx = mysql.connector.connect(
+    cnx = MySQLdb.connect(
         user='user',
         password='pass',
         host='delphi_database_epidata',

integrations/server/test_fluview.py

@@ -4,7 +4,7 @@
 import unittest
 
 # third party
-import mysql.connector
+import MySQLdb
 
 # first party
 from delphi.epidata.client.delphi_epidata import Epidata
@@ -25,7 +25,7 @@ def setUp(self):
     """Perform per-test setup."""
 
     # connect to the `epidata` database and clear the `fluview` table
-    cnx = mysql.connector.connect(
+    cnx = MySQLdb.connect(
         user='user',
         password='pass',
         host='delphi_database_epidata',

integrations/server/test_fluview_meta.py

@@ -4,7 +4,7 @@
 import unittest
 
 # third party
-import mysql.connector
+import MySQLdb
 
 # first party
 from delphi.epidata.client.delphi_epidata import Epidata
@@ -25,7 +25,7 @@ def setUp(self):
     """Perform per-test setup."""
 
     # connect to the `epidata` database and clear the `fluview` table
-    cnx = mysql.connector.connect(
+    cnx = MySQLdb.connect(
         user='user',
         password='pass',
         host='delphi_database_epidata',

requirements.api.txt

@@ -4,7 +4,7 @@ Flask==2.2.5
 Flask-Limiter==3.3.0
 jinja2==3.1.3
 more_itertools==8.4.0
-mysqlclient==2.1.1
+mysqlclient==2.2.1
 orjson==3.4.7
 pandas==1.2.3
 python-dotenv==0.15.0
@@ -13,7 +13,7 @@ redis==3.5.3
 requests==2.31.0
 scipy==1.10.0
 sentry-sdk[flask]
-SQLAlchemy==1.4.40
+SQLAlchemy==1.4.51
 structlog==22.1.0
 tenacity==7.0.0
 typing-extensions

requirements.dev.txt

@@ -10,10 +10,8 @@ invoke>=1.4.1
 lxml==4.9.1
 matplotlib==3.6.2
 mypy>=0.790
-mysql-connector==2.2.9
 numpy==1.22.4
 pycountry==22.3.5
-pymysql==1.0.2
 pytest==7.2.0
 pytest-check==1.3.0
 sas7bdat==2.2.3

src/acquisition/cdcp/cdc_dropbox_receiver.py

@@ -22,7 +22,7 @@
 
 # third party
 import dropbox
-import mysql.connector
+import MySQLdb
 
 # first party
 import delphi.operations.secrets as secrets
@@ -47,7 +47,7 @@ def trigger_further_processing():
 
     # connect
     u, p = secrets.db.auto
-    cnx = mysql.connector.connect(user=u, password=p, database="automation")
+    cnx = MySQLdb.connect(user=u, password=p, database="automation")
     cur = cnx.cursor()
 
     # add step "Process CDCP Data" to queue

src/acquisition/cdcp/cdc_extract.py

@@ -66,7 +66,7 @@
 import sys
 
 # third party
-import mysql.connector
+import MySQLdb
 
 # first party
 import delphi.operations.secrets as secrets
@@ -154,7 +154,7 @@ def extract(first_week=None, last_week=None, test_mode=False):
 
     # connect
     u, p = secrets.db.epi
-    cnx = mysql.connector.connect(user=u, password=p, database="epidata")
+    cnx = MySQLdb.connect(user=u, password=p, database="epidata")
     cur = cnx.cursor()
 
     # weeks to update

src/acquisition/cdcp/cdc_upload.py

@@ -80,7 +80,7 @@
 from zipfile import ZipFile
 
 # third party
-import mysql.connector
+import MySQLdb
 
 # first party
 import delphi.operations.secrets as secrets
@@ -165,7 +165,7 @@
 def upload(test_mode):
     # connect
     u, p = secrets.db.epi
-    cnx = mysql.connector.connect(user=u, password=p, database="epidata")
+    cnx = MySQLdb.connect(user=u, password=p, database="epidata")
     cur = cnx.cursor()
 
     # insert (or update) table `cdc`

src/acquisition/covid_hosp/common/database.py

@@ -6,7 +6,7 @@
 import math
 
 # third party
-import mysql.connector
+import MySQLdb
 import pandas as pd
 
 # first party
@@ -60,7 +60,7 @@ def logger(database_class):
 
   @classmethod
   @contextmanager
-  def connect(database_class, mysql_connector_impl=mysql.connector):
+  def connect(database_class, mysql_connector_impl=MySQLdb):
     """Connect to a database and provide the connection as a context manager.
 
     As long as the context manager exits normally, the connection's transaction

src/acquisition/covidcast/database.py

@@ -10,7 +10,7 @@
 
 # third party
 import json
-import mysql.connector
+import MySQLdb
 
 # first party
 import delphi.operations.secrets as secrets
@@ -38,7 +38,7 @@ class Database:
   # TODO: also consider that for composite key tuples, like short_comp_key and long_comp_key as used in delete_batch()
 
 
-  def connect(self, connector_impl=mysql.connector):
+  def connect(self, connector_impl=MySQLdb):
     """Establish a connection to the database."""
 
     u, p = secrets.db.epi
@@ -71,8 +71,7 @@ def disconnect(self, commit):
 
   def count_all_load_rows(self):
     self._cursor.execute(f'SELECT count(1) FROM `{self.load_table}`')
-    for (num,) in self._cursor:
-      return num
+    return self._cursor.fetchone()[0]
 
   def _reset_load_table_ai_counter(self):
     """Corrects the AUTO_INCREMENT counter in the load table.
@@ -101,7 +100,8 @@ def do_analyze(self):
       f'''ANALYZE TABLE
           signal_dim, geo_dim,
           {self.load_table}, {self.history_table}, {self.latest_table}''')
-    output = [self._cursor.column_names] + self._cursor.fetchall()
+    # Append lists of column names (from cursor.description) & values
+    output = [desc[0] for desc in self._cursor.description] + list(self._cursor.fetchall())
     get_structured_logger('do_analyze').info("ANALYZE results", results=str(output))
 
   def insert_or_update_bulk(self, cc_rows):
@@ -456,8 +456,8 @@ def compute_covidcast_meta(self, table_name=None, n_threads=None):
     srcsigs = Queue() # multi-consumer threadsafe!
     sql = f'SELECT `source`, `signal` FROM `{table_name}` GROUP BY `source`, `signal` ORDER BY `source` ASC, `signal` ASC;'
     self._cursor.execute(sql)
-    for source, signal in self._cursor:
-      srcsigs.put((source, signal))
+    for res in self._cursor.fetchall():
+      srcsigs.put((res[0], res[1])) # source, signal
 
     inner_sql = f'''
       SELECT
@@ -505,8 +505,9 @@ def worker():
           logger.info("starting pair", thread=name, pair=f"({source}, {signal})")
           w_cursor.execute(inner_sql, (source, signal))
           with meta_lock:
+            # Create a dictionary of column names (from cursor.description) & values
             meta.extend(list(
-              dict(zip(w_cursor.column_names, x)) for x in w_cursor
+              dict(zip([desc[0] for desc in w_cursor.description], x)) for x in w_cursor
             ))
           srcsigs.task_done()
       except Empty:

src/acquisition/ecdc/ecdc_db_update.py

@@ -37,7 +37,7 @@
 import tempfile
 
 # third party
-import mysql.connector
+import MySQLdb
 
 # first party
 import delphi.operations.secrets as secrets
@@ -48,7 +48,7 @@
 
 def ensure_tables_exist():
     (u, p) = secrets.db.epi
-    cnx = mysql.connector.connect(user=u, password=p, database="epidata")
+    cnx = MySQLdb.connect(user=u, password=p, database="epidata")
     try:
         cursor = cnx.cursor()
         cursor.execute(
@@ -98,7 +98,7 @@ def update_from_file(issue, date, dir, test_mode=False):
     # Read ECDC data from CSVs and insert into (or update) the database.
     # database connection
     u, p = secrets.db.epi
-    cnx = mysql.connector.connect(user=u, password=p, database="epidata")
+    cnx = MySQLdb.connect(user=u, password=p, database="epidata")
     rows1 = get_rows(cnx, "ecdc_ili")
     print(f"rows before: {int(rows1)}")
     insert = cnx.cursor()

src/acquisition/flusurv/flusurv_update.py

@@ -72,7 +72,7 @@
 import argparse
 
 # third party
-import mysql.connector
+import MySQLdb
 
 # first party
 from delphi.epidata.acquisition.flusurv import flusurv
@@ -105,7 +105,7 @@ def update(issue, location_name, test_mode=False):
 
     # connect to the database
     u, p = secrets.db.epi
-    cnx = mysql.connector.connect(host=secrets.db.host, user=u, password=p, database="epidata")
+    cnx = MySQLdb.connect(host=secrets.db.host, user=u, password=p, database="epidata")
     cur = cnx.cursor()
     rows1 = get_rows(cur)
     print(f"rows before: {int(rows1)}")

src/acquisition/fluview/fluview_notify.py

@@ -25,7 +25,7 @@
 import argparse
 
 # third party
-import mysql.connector
+import MySQLdb
 
 # first party
 import delphi.operations.secrets as secrets
@@ -46,7 +46,7 @@
 
     # connect
     u, p = secrets.db.epi
-    cnx = mysql.connector.connect(user=u, password=p, database="epidata", host=secrets.db.host)
+    cnx = MySQLdb.connect(user=u, password=p, database="epidata", host=secrets.db.host)
     cur = cnx.cursor()
 
     # get the last known issue from the automation table `variables`

src/acquisition/fluview/fluview_update.py

@@ -121,7 +121,7 @@
 import zipfile
 
 # third party
-import mysql.connector
+import MySQLdb
 
 # first party
 import delphi.operations.secrets as secrets
@@ -311,7 +311,7 @@ def update_from_file_clinical(issue, date, filename, test_mode=False):
 
     # database connection
     u, p = secrets.db.epi
-    cnx = mysql.connector.connect(user=u, password=p, database="epidata", host=secrets.db.host)
+    cnx = MySQLdb.connect(user=u, password=p, database="epidata", host=secrets.db.host)
     rows1 = get_rows(cnx, CL_TABLE)
     print(f"rows before: {int(rows1)}")
     insert = cnx.cursor()
@@ -376,7 +376,7 @@ def update_from_file_public(issue, date, filename, test_mode=False):
 
     # database connection
     u, p = secrets.db.epi
-    cnx = mysql.connector.connect(user=u, password=p, database="epidata", host=secrets.db.host)
+    cnx = MySQLdb.connect(user=u, password=p, database="epidata", host=secrets.db.host)
     rows1 = get_rows(cnx, PHL_TABLE)
     print(f"rows before: {int(rows1)}")
     insert = cnx.cursor()
@@ -445,7 +445,7 @@ def update_from_file(issue, date, filename, test_mode=False):
 
     # database connection
     u, p = secrets.db.epi
-    cnx = mysql.connector.connect(user=u, password=p, database="epidata", host=secrets.db.host)
+    cnx = MySQLdb.connect(user=u, password=p, database="epidata", host=secrets.db.host)
     rows1 = get_rows(cnx)
     print(f"rows before: {int(rows1)}")
     insert = cnx.cursor()