Merge pull request #28 from pharmorg/bulk_insert

Decrease time to load DB to approximately 5 minutes instead of 60 minutes.

Author: finish06

openfda/app/db/models.py

@@ -16,7 +16,9 @@ class Drugs(Base):
     pharm_class = Column(String(100))
     dosage_form = Column(String(100))
     product_type = Column(Integer, ForeignKey('productType.id'))
-    routes = relationship("Routes", backref='dx_route')
+    route1 = Column(Integer, ForeignKey('route.id'))
+    route2 = Column(Integer, ForeignKey('route.id'))
+    route3 = Column(Integer, ForeignKey('route.id'))
     pharm_classes = relationship("PharmClasses", backref="dx_pharmClass")
 
     def __repr__(self):
@@ -29,7 +31,7 @@ class Routes(Base):
 
     id = Column(Integer, primary_key=True)
     route = Column(String(100))
-    drug_id = Column(Integer, ForeignKey('drug.id'))
+    
 
     def __repr__(self):
         return "<Routes(route='%s')>" % (self.route)

openfda/app/utils/load_data.py

@@ -1,6 +1,7 @@
 import requests
 import sys
 import json
+import time
 from pathlib import Path
 from db.models import Drugs, Routes, ProductTypes, PharmClasses
 from db.connect import connection
@@ -24,64 +25,110 @@ def buildProductTypes(session, data):
     for item in responses:
         exists = session.query(ProductTypes).filter(ProductTypes.product_type == item).scalar()
         if not exists:
-            logger.info(f"New response added: {item}\n")
+            logger.debug(f"New response added: {item}\n")
             row = ProductTypes(product_type = item)
             session.add(row)
 
-def addData(session, data):
+def buildRouteTypes(session, data):
+    routes = set()
     try:
+        for item in data['results']:
+            try:
+                for route in item['route']:
+                    routes.add(route)
+            except:
+                pass
+    except:
+        logger.error("Failed to load routes table")
+        return
+    logger.debug(f"New data route(s): {str(routes)}\n")
+    for item in routes:
+        logger.debug(item)
+        exists = session.query(Routes).filter(Routes.route == item).scalar()
+        if not exists:
+            logger.debug(f"New response added: {item}\n")
+            row = Routes(route = item)
+            session.add(row)
+
+
+def buildDrugs(session, drugs_data):
+    objects = []
+    product_ids = []
+    for data in drugs_data['results']:
         try:
-            product_id = data['product_id']
+            try:
+                product_id = data['product_id']
+            except:
+                logger.warning(f"A product ID does not exists for {data['generic_name']}")
+                return
+            try:
+                generic_name = data['generic_name'].lower()[:300]
+            except:
+                generic_name = ""
+            try:
+                brand_name = data['brand_name'].lower()[:300]
+            except:
+                brand_name = ""
+            try:
+                classList = []
+                for classItem in data['pharm_class']:
+                    classList.append(classItem)
+            except:
+                classList = []
+            try:
+                routesList = []
+                for route in data['route']:
+                    route_id = session.query(Routes.id).filter(Routes.route == route)
+                    routesList.append(route_id)
+            except:
+                routesList = []
+            try:
+                form = data['dosage_form'].lower()[:100]
+            except:
+                form = ""
+            try:
+                productTypeId = session.query(ProductTypes.id).filter(ProductTypes.product_type == data['product_type'])
+            except:
+                productTypeId = None
         except:
-            logger.warning(f"A product ID does not exists for {data['generic_name']}")
+            logger.error(f"JSON failure\n")
             return
-        try:
-            generic_name = data['generic_name'].lower()[:300]
-        except:
-            generic_name = ""
-        try:
-            brand_name = data['brand_name'].lower()[:300]
-        except:
-            brand_name = ""
-        try:
-            classList = []
-            for classItem in data['pharm_class']:
-                classList.append(classItem)
-        except:
-            classList = []
-        try:
-            routesList = []
-            for route in data['route']:
-                routesList.append(route.lower())
-        except:
-            routesList = []
-        try:
-            form = data['dosage_form'].lower()[:100]
-        except:
-            form = ""
-        try:
-            productTypeId = session.query(ProductTypes.id).filter(ProductTypes.product_type == data['product_type'])
-        except:
-            productTypeId = None
-    except:
-        logger.error(f"JSON failure\n")
-        return
-    exists = session.query(Drugs).filter(Drugs.product_id == product_id).scalar()
-    if not exists:
-        logger.info(f"New data added: {generic_name}|{brand_name}|{form}|{product_id}\n")
-        row = Drugs(product_id=product_id, generic_name=generic_name, brand_name=brand_name, dosage_form=form, product_type = productTypeId)            
-        session.add(row)
-        for route in routesList:
-            logger.info(f"New data added: {generic_name}|{route}\n")
-            route_row = Routes(route=route, dx_route=row)
-            session.add(route_row)
-        for pharmClass in classList:
-            logger.info(f"New data added: {generic_name}|{pharmClass}\n")
-            class_row = PharmClasses(pharm_class=pharmClass, dx_pharmClass=row)
-            session.add(class_row)
-    else:
-        logger.warning(f"Data already exists: {product_id}|{generic_name}|{brand_name}\n")
+        # exists = session.query(Drugs).filter(Drugs.product_id == product_id).scalar()
+        if product_id not in product_ids:
+            product_ids.append(product_id)
+            logger.debug(f"New data added: {generic_name}|{brand_name}|{form}|{product_id}\n")
+            if len(routesList) == 3:
+                route1 = routesList[0]
+                route2 = routesList[1]
+                route3 = routesList[2]
+            elif len(routesList) == 2:
+                route1 = routesList[0]
+                route2 = routesList[1]
+                route3 = None
+            elif len(routesList) == 1:
+                route1 = routesList[0]
+                route2 = None
+                route3 = None
+            else:
+                route1 = None
+                route2 = None
+                route3 = None
+            row = Drugs(product_id=product_id, generic_name=generic_name, brand_name=brand_name, dosage_form=form, route1=route1, route2=route2, route3=route3, product_type=productTypeId)            
+            objects.append(row)
+            # for route in routesList:
+            #     logger.info(f"New data added: {generic_name}|{route}\n")
+            #     route_row = Routes(route=route, dx_route=row)
+            #     objects.append(route_row)
+            for pharmClass in classList:
+                logger.debug(f"New data added: {generic_name}|{pharmClass}\n")
+                class_row = PharmClasses(pharm_class=pharmClass, dx_pharmClass=row)
+                objects.append(class_row)
+        else:
+            logger.warning(f"Data already exists: {product_id}|{generic_name}|{brand_name}\n")
+    start = time.process_time()
+    session.add_all(objects)
     session.commit()
+    logger.info(f"Commit drugs db: {str(time.process_time() - start)}")
     return
 
 
@@ -101,10 +148,16 @@ def main(session):
     # Need a flag / config / etc. to drop tables on demand for rebuild.
 
     # Need improved process to minimze circling through JSON file twice.
+    start = time.process_time()
     buildProductTypes(session, data)
-    for line in data['results']:
-        logger.debug(f"{line}")
-        addData(session, line)
+    logger.info(f"Add product types: {str(time.process_time() - start)}")
+    start = time.process_time()
+    buildRouteTypes(session, data)
+    logger.info(f"Add routes: {str(time.process_time() - start)}")
+    start= time.process_time()
+    buildDrugs(session, data)
+    logger.info(f"Add drugs information: {str(time.process_time() - start)}")
+    f.close()
 
 
 if __name__ == "__main__":

openfda/app/utils/log.py

@@ -14,7 +14,7 @@ def startLogging(logger_name):
     Path(LOG_DIR).mkdir(parents=True, exist_ok=True)
     logger = logging.getLogger(logger_name)
     formatter = logging.Formatter('[%(asctime)s] p%(process)s {%(pathname)s:%(lineno)d} %(levelname)s - %(message)s','%Y/%m/%d %H:%M:%S')
-    logger.setLevel(logging.DEBUG)
+    logger.setLevel(logging.INFO)
     # enable file logging
     log_filename = datetime.datetime.now().strftime("%y%m%d_") + logger_name + '.log'
     filehandler = logging.FileHandler(LOG_DIR / log_filename)