Remove type parameter, use Django ENGINE

Author: ssciolla

dashboard/common/db_util.py

@@ -14,23 +14,27 @@
 
 from dashboard.models import Course, User
 
+
 logger = logging.getLogger(__name__)
 
+BACKENDS_PATH = 'django.db.backends.'
+
 
-class DBParams(TypedDict):
+class DjangoDBParams(TypedDict):
+    ENGINE: Literal['django.db.backends.mysql', 'django.db.backends.postgresql']
     NAME: str
     USER: str
     PASSWORD: str
     HOST: str
     PORT: int
 
 
-def create_sqlalchemy_engine(db_params: DBParams, type: Literal['mysql', 'postgres']) -> Engine:
-    new_db_params: DBParams = db_params.copy()
+def create_sqlalchemy_engine(db_params: DjangoDBParams) -> Engine:
+    new_db_params: DjangoDBParams = db_params.copy()
     new_db_params['PASSWORD'] = quote_plus(db_params['PASSWORD'])
 
     core_string = '{USER}:{PASSWORD}@{HOST}:{PORT}/{NAME}'.format(**new_db_params)
-    if type == 'mysql':
+    if new_db_params['ENGINE'] == (BACKENDS_PATH + 'mysql'):
         return create_engine(f'mysql+mysqldb://{core_string}?charset=utf8mb4')
     else:
         return create_engine('postgresql://' + core_string)

dashboard/cron.py

@@ -22,8 +22,8 @@
 
 logger = logging.getLogger(__name__)
 
-engine = db_util.create_sqlalchemy_engine(settings.DATABASES['default'], 'mysql')
-data_warehouse_engine = db_util.create_sqlalchemy_engine(settings.DATABASES['DATA_WAREHOUSE'], 'postgres')
+engine = db_util.create_sqlalchemy_engine(settings.DATABASES['default'])
+data_warehouse_engine = db_util.create_sqlalchemy_engine(settings.DATABASES['DATA_WAREHOUSE'])
 
 # Set up queries array from configuration file
 CRON_QUERY_FILE = settings.CRON_QUERY_FILE

dashboard/views.py

@@ -43,7 +43,7 @@
 
 BinningGrade = namedtuple('BinningGrade', ['value', 'index', 'binning_all'])
 
-app_engine = create_sqlalchemy_engine(settings.DATABASES['default'], 'mysql')
+app_engine = create_sqlalchemy_engine(settings.DATABASES['default'])
 
 
 def gpa_map(grade):

scripts/data_validation/validate_udw_vs_udp.py

@@ -80,8 +80,8 @@ def main():
     ENV_VALIDATION = get_env_file(
         os.path.join(os.path.dirname(os.path.abspath('__file__')), 'scripts/data_validation/env_validation.hjson'), 'hjson')
 
-    udw_engine = create_sqlalchemy_engine(ENV_UDW['DATA_WAREHOUSE'], 'postgres')
-    udp_engine = create_sqlalchemy_engine(ENV_UDP['DATA_WAREHOUSE'], 'postgres')
+    udw_engine = create_sqlalchemy_engine(ENV_UDW['DATA_WAREHOUSE'])
+    udp_engine = create_sqlalchemy_engine(ENV_UDP['DATA_WAREHOUSE'])
 
     DATA_WAREHOUSE_COURSE_IDS = ENV_VALIDATION["DATA_WAREHOUSE_COURSE_IDS"]