Better bulk import: separate beacon creation from db insert

ogn/gateway/manage.py

@@ -1,7 +1,8 @@
 import logging
 
 from ogn.client import AprsClient
-from ogn.gateway.process import process_beacon
+from ogn.gateway.process import process_beacon, message_to_beacon
+from ogn.commands.dbutils import session
 from datetime import datetime
 from manager import Manager
 
@@ -65,9 +66,15 @@ def import_logfile(ogn_logfile, reference_date, logfile='main.log', loglevel='IN
         log_handlers.append(logging.FileHandler(logfile))
     logging.basicConfig(format=logging_formatstr, level=loglevel, handlers=log_handlers)
 
+    beacons = list()
+
     print('Start importing ogn-logfile')
     for line in f:
-        process_beacon(line, reference_date=reference_date)
+        beacon = message_to_beacon(line, reference_date=reference_date)
+        if beacon is not None:
+            beacons.append(beacon)
+
+    session.bulk_save_objects(beacons)
 
     f.close()
     logging.shutdown()

ogn/gateway/process.py

@@ -14,11 +14,12 @@ def replace_lonlat_with_wkt(message):
     return message
 
 
-def process_beacon(raw_message, reference_date=None, reference_time=None):
+def message_to_beacon(raw_message, reference_date):
-    if raw_message[0] == '#':
+    beacon = None
-        return
+
+    if raw_message[0] != '#':
         try:
-        message = parse_aprs(raw_message, reference_date, reference_time)
+            message = parse_aprs(raw_message, reference_date)
             # symboltable / symbolcodes used by OGN:
             # I&: used as receiver
             # /X: helicopter_rotorcraft
@@ -34,22 +35,27 @@ def process_beacon(raw_message, reference_date=None, reference_time=None):
                 # we have a receiver_beacon (status message)
                 message.update(parse_ogn_receiver_beacon(message['comment']))
                 beacon = ReceiverBeacon(**message)
-
-            # connect beacon with receiver
-            receiver = session.query(Receiver.id) \
-                .filter(Receiver.name == beacon.name) \
-                .first()
-            if receiver is None:
-                receiver = Receiver()
-                receiver.name = beacon.name
-                session.add(receiver)
-            beacon.receiver_id = receiver.id
             elif message['symboltable'] == "I" and message['symbolcode'] == '&':
                 # ... we have a receiver_beacon
                 message.update(parse_ogn_receiver_beacon(message['comment']))
                 message = replace_lonlat_with_wkt(message)
                 beacon = ReceiverBeacon(**message)
+            else:
+                # ... we have a aircraft_beacon
+                message.update(parse_ogn_aircraft_beacon(message['comment']))
+                message = replace_lonlat_with_wkt(message)
+                beacon = AircraftBeacon(**message)
+        except ParseError as e:
+            logger.error('Received message: {}'.format(raw_message))
+            logger.error('Drop packet, {}'.format(e.message))
+        except TypeError as e:
+            logger.error('TypeError: {}'.format(raw_message))
 
+    return beacon
+
+
+def add_beacon_to_db(beacon):
+    if type(beacon == ReceiverBeacon):
         # connect beacon with receiver
         receiver = session.query(Receiver.id) \
             .filter(Receiver.name == beacon.name) \
@@ -59,12 +65,7 @@ def process_beacon(raw_message, reference_date=None, reference_time=None):
             receiver.name = beacon.name
             session.add(receiver)
         beacon.receiver_id = receiver.id
-        else:
+    elif type(beacon == AircraftBeacon):
-            # ... we have a aircraft_beacon
-            message.update(parse_ogn_aircraft_beacon(message['comment']))
-            message = replace_lonlat_with_wkt(message)
-            beacon = AircraftBeacon(**message)
-
         # connect beacon with device
         device = session.query(Device) \
             .filter(Device.address == beacon.address) \
@@ -97,9 +98,10 @@ def process_beacon(raw_message, reference_date=None, reference_time=None):
 
     session.add(beacon)
     session.commit()
+
+
+def process_beacon(raw_message, reference_date=None):
+    beacon = message_to_beacon(raw_message, reference_date)
+    if beacon is not None:
+        add_beacon_to_db(beacon)
         logger.debug('Received message: {}'.format(raw_message))
-    except ParseError as e:
-        logger.error('Received message: {}'.format(raw_message))
-        logger.error('Drop packet, {}'.format(e.message))
-    except TypeError as e:
-        logger.error('TypeError: {}'.format(raw_message))