Add lazy loading for access to database

Author: meier-rene

similarity-service-cosine/README.md

@@ -26,7 +26,7 @@ To run the server, please execute the following from the root directory:
 
 ```
 pip3 install -r requirements.txt
-PYTHONPATH=gen python3 -m similarity-score-impl
+PYTHONPATH=gen python3 -m cosine_impl
 ```
 
 and open your browser to here:

similarity-service-cosine/cosine_impl/__main__.py

@@ -3,14 +3,8 @@
 import connexion
 
 from openapi_server import encoder
-from cosine_impl import db
-
 
 def main():
-    print("Loading spectra... ", end='', flush=True)
-    db.load_spectra()
-    print("Finished")
-
     app = connexion.App(__name__, specification_dir='..')
     app.app.json_encoder = encoder.JSONEncoder
     app.add_api('openapi.yaml',

similarity-service-cosine/cosine_impl/cosine_impl_controller.py

@@ -3,11 +3,20 @@
 from matchms import Spectrum, calculate_scores
 from matchms.similarity import CosineGreedy
 
-from cosine_impl.db import spectra
+from cosine_impl.db import ReferenceSpectra
 from openapi_server.models import SimilarityScore
 from openapi_server.models.similarity_calculation import SimilarityCalculation  # noqa: E501
 from openapi_server.models.similarity_score_list import SimilarityScoreList  # noqa: E501
 
+import os
+import psycopg
+
+DB_PORT = os.environ.get('DB_PORT', 5432)
+DB_USER = os.environ.get('DB_USER', "massbank3")
+DB_PASSWORD = os.environ.get('DB_PASSWORD', "massbank3password")
+DB_HOST = os.environ.get('DB_HOST', "localhost")
+DB_NAME = os.environ.get('DB_NAME', "massbank3")
+
 
 def similarity_post(similarity_calculation):  # noqa: E501
     """Create a new similarity calculation.
@@ -22,6 +31,9 @@ def similarity_post(similarity_calculation):  # noqa: E501
     if connexion.request.is_json:
         request = SimilarityCalculation.from_dict(similarity_calculation)
 
+        reference_spectra = ReferenceSpectra(psycopg.connect(f"postgresql://{DB_NAME}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"))
+        reference_spectra.load_spectra()
+
         mz = []
         intensities = []
 
@@ -35,11 +47,11 @@ def similarity_post(similarity_calculation):  # noqa: E501
             def filter_fn(spectrum):
                 return spectrum.metadata['spectrum_id'] in request.reference_spectra_list
 
-            references = list(filter(filter_fn, spectra))
+            references = list(filter(filter_fn, ReferenceSpectra.spectra))
 
             scores = calculate_scores(references, [query], CosineGreedy())
         else:
-            scores = calculate_scores(spectra, [query], CosineGreedy())
+            scores = calculate_scores(ReferenceSpectra.spectra, [query], CosineGreedy())
         matches = scores.scores_by_query(query, 'CosineGreedy_score', sort=True)
         match_list = SimilarityScoreList([])
 

similarity-service-cosine/cosine_impl/db.py

@@ -1,58 +1,53 @@
-import os
-import psycopg
 from matchms import set_matchms_logger_level, Spectrum
 import numpy as np
+from datetime import datetime
+import logging
 
-DB_PORT = os.environ.get('DB_PORT', 5432)
-DB_USER = os.environ.get('DB_USER', "massbank3")
-DB_PASSWORD = os.environ.get('DB_PASSWORD', "massbank3password")
-DB_HOST = os.environ.get('DB_HOST', "localhost")
-DB_NAME = os.environ.get('DB_NAME', "massbank3")
-spectra = []
 
 class ReferenceSpectra:
     """This class loads all reference spectra from the database"""
+    timestamp = datetime.fromisoformat('2010-01-01')
+    spectra = []
+
     def __init__(self, connection):
         """initialize the class with a database connection"""
         self.connection = connection
-        self.spectra = []
 
-    def get_spectra(self):
+    @property
+    def load_spectra(self):
+        """load all spectra from the database if the metadata indicates newer data and stores them"""
         with self.connection.cursor() as cur:
             cur.execute("select * from metadata;")
             timestamp = cur.fetchone()[2]
-        print(timestamp)
-
-if __name__ == '__main__':
-    myspec=ReferenceSpectra(psycopg.connect(f"postgresql://{DB_NAME}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"))
-    myspec.get_spectra()
+        logging.info("Database timestamp: %s", timestamp)
+        logging.info("In-memory timestamp: %s", ReferenceSpectra.timestamp)
+        timestamp_diff = timestamp - ReferenceSpectra.timestamp
+        if timestamp_diff.total_seconds() > 0:
+            logging.info("Database timestamp is %s newer. Reloading...", timestamp_diff)
 
-# Load all (non-deprecated) spectra from the database for faster lookup
-def load_spectra():
-    global spectra
-
-    spectra = []
+            # Prevent matchms from complaining about spectra not having a precursor_mz
+            set_matchms_logger_level("ERROR")
 
-    # Prevent matchms from complaining about spectra not having a precursor_mz
-    set_matchms_logger_level("ERROR")
+            with self.connection.cursor() as cur:
+                cur.execute(
+                    "SELECT massbank.accession as accession, peak.mz as mz, peak.intensity as intensity, "
+                    "peak.relative_intensity as rel FROM massbank JOIN spectrum ON massbank.id = spectrum.massbank_id "
+                    "JOIN peak ON spectrum.id = peak.spectrum_id ORDER BY massbank.id, peak.mz;"
+                )
 
-    with psycopg.connect(f"postgresql://{DB_NAME}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}") as conn:
-        with conn.cursor() as cur:
-            cur.execute(
-                "SELECT massbank.accession as accession, peak.mz as mz, peak.intensity as intensity, peak.relative_intensity as rel FROM massbank JOIN spectrum ON massbank.id = spectrum.massbank_id JOIN peak ON spectrum.id = peak.spectrum_id ORDER BY massbank.id, peak.mz;"
-            )
+                entries = {}
+                ReferenceSpectra.spectra = []
 
-            entries = {}
-            for res in cur:
-                accession = res[0]
-                if accession in entries:
-                    entries[accession]["mz"].append(res[1])
-                    entries[accession]["intensity"].append(res[2])
-                    entries[accession]["rel"].append(res[3])
-                else:
-                    entries[accession] = {"mz": [res[1]], "intensity": [res[2]], "rel": [res[3]]}
+                for res in cur:
+                    accession = res[0]
+                    if accession in entries:
+                        entries[accession]["mz"].append(res[1])
+                        entries[accession]["intensity"].append(res[2])
+                        entries[accession]["rel"].append(res[3])
+                    else:
+                        entries[accession] = {"mz": [res[1]], "intensity": [res[2]], "rel": [res[3]]}
 
-            for accession in entries:
+                for accession in entries:
                     mz = []
                     intensities = []
                     for key in entries[accession]:
@@ -61,10 +56,34 @@ def load_spectra():
                             mz = prop
                         elif key == "rel":
                             intensities = prop
-                         
-                    if len(mz) > 0 and len(intensities) > 0 and len(mz) == len(intensities):    
-                        #metadata key "accession" gets silently converted to spectrum_id, so we can use spectrum_id right away
-                        spectra.append(Spectrum(mz=np.array(mz).astype(float), intensities=np.array(intensities).astype(float),
-                                        metadata={'spectrum_id': accession}))
-    
-    print(f"\nLoaded {len(spectra)} spectra from the database")
+
+                    if 0 < len(mz) == len(intensities):
+                        # metadata key "accession" gets silently converted to spectrum_id, so we can use spectrum_id
+                        # right away
+                        ReferenceSpectra.spectra.append(
+                            Spectrum(mz=np.array(mz).astype(float), intensities=np.array(intensities).astype(float),
+                                     metadata={'spectrum_id': accession}))
+                    else:
+                        logging.warning("Empty spectrum or error in spectrum")
+
+                logging.info("Loaded %s spectra from the database.", len(ReferenceSpectra.spectra))
+                ReferenceSpectra.timestamp = timestamp
+
+
+# import os
+# import psycopg
+#
+# if __name__ == '__main__':
+#     DB_PORT = os.environ.get('DB_PORT', 5432)
+#     DB_USER = os.environ.get('DB_USER', "massbank3")
+#     DB_PASSWORD = os.environ.get('DB_PASSWORD', "massbank3password")
+#     DB_HOST = os.environ.get('DB_HOST', "localhost")
+#     DB_NAME = os.environ.get('DB_NAME', "massbank3")
+#
+#     logging.basicConfig(format='%(levelname)s:%(message)s', level=logging.DEBUG)
+#     myspec = ReferenceSpectra(psycopg.connect(f"postgresql://{DB_NAME}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"))
+#     print(len(ReferenceSpectra.spectra))
+#     myspec.load_spectra
+#     print(len(ReferenceSpectra.spectra))
+#     myspec.load_spectra
+#     print(len(ReferenceSpectra.spectra))