Merge remote-tracking branch 'origin/main' into refactor/schemaless_c…

…onsumer_use_path
alercebroker · Nov 17, 2023 · cf07873 · cf07873
2 parents 413d6a3 + c2f4a06
commit cf07873
Show file tree

Hide file tree

Showing 33 changed files with 73 additions and 76 deletions.
diff --git a/alert_archiving_step/pyproject.toml b/alert_archiving_step/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "alert-archiving-step"
-version = "6.1.1a78"
+version = "6.1.1a79"
 description = "Group alerts and upload to S3 by date"
 authors = ["ALeRCE"]
 readme = "README.md"

diff --git a/charts/alert_archiving_step/Chart.yaml b/charts/alert_archiving_step/Chart.yaml
@@ -1,6 +1,6 @@
 apiVersion: v2
-appVersion: 6.1.1a78
+appVersion: 6.1.1a79
 description: A Helm chart for Kubernetes
 name: alert-archive-step
 type: application
-version: 1.2.46
+version: 1.2.47
diff --git a/charts/correction_step/Chart.yaml b/charts/correction_step/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v2
-appVersion: 6.2.1a82
+appVersion: 6.2.1a83
 description: Correction step chart
 name: correction-step
 type: application

diff --git a/charts/early_classification_step/Chart.yaml b/charts/early_classification_step/Chart.yaml
@@ -1,6 +1,6 @@
 apiVersion: v2
-appVersion: 6.2.1a82
+appVersion: 6.2.1a83
 description: A Helm chart for Kubernetes
 name: early-classifier
 type: application
-version: 1.2.157
+version: 1.2.158
diff --git a/charts/feature_step/Chart.yaml b/charts/feature_step/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v2
-appVersion: 6.2.1a82
+appVersion: 6.2.1a83
 description: Step for feature calculation
 name: feature-step
 type: application

diff --git a/charts/lc_classification_step/Chart.yaml b/charts/lc_classification_step/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v2
-appVersion: 6.2.1a82
+appVersion: 6.2.1a83
 description: Lightcurve classifier step
 name: lc-classifier-step
 type: application

diff --git a/charts/lightcurve-step/Chart.yaml b/charts/lightcurve-step/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v2
-appVersion: 6.2.1a82
+appVersion: 6.2.1a83
 description: Full lightcurve provider step
 name: lightcurve-step
 type: application

diff --git a/charts/magstats_step/Chart.yaml b/charts/magstats_step/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v2
-appVersion: 6.2.1a82
+appVersion: 6.2.1a83
 description: A Helm chart for Kubernetes
 name: magstats-step
 type: application

diff --git a/charts/metadata_step/Chart.yaml b/charts/metadata_step/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v2
-appVersion: 6.2.1a49
+appVersion: 6.2.1a50
 description: A Helm chart for Kubernetes
 name: metadata-step
 type: application

diff --git a/charts/prv_candidates_step/Chart.yaml b/charts/prv_candidates_step/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v2
-appVersion: 6.2.1a82
+appVersion: 6.2.1a83
 description: Previous candidates processor step
 name: prv-candidates
 type: application

diff --git a/charts/s3_step/Chart.yaml b/charts/s3_step/Chart.yaml
@@ -1,6 +1,6 @@
 apiVersion: v2
-appVersion: 6.2.1a82
+appVersion: 6.2.1a83
 description: A Helm chart for Kubernetes
 name: s3-step
 type: application
-version: 1.3.159
+version: 1.3.160
diff --git a/charts/scribe/Chart.yaml b/charts/scribe/Chart.yaml
@@ -21,4 +21,4 @@ version: 2.0.0
 # incremented each time you make changes to the application. Versions are not expected to
 # follow Semantic Versioning. They should reflect the version the application is using.
 # It is recommended to use it with quotes.
-appVersion: 6.2.1a82
+appVersion: 6.2.1a83
diff --git a/charts/sorting_hat_step/Chart.yaml b/charts/sorting_hat_step/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v2
-appVersion: 6.2.1a82
+appVersion: 6.2.1a83
 description: Sorting Hat deployment chart
 name: sorting-hat
 type: application

diff --git a/charts/watchlist_step/Chart.yaml b/charts/watchlist_step/Chart.yaml
@@ -1,7 +1,7 @@
 apiVersion: v2
-appVersion: 6.2.1a82
+appVersion: 6.2.1a83
 description: A Helm chart for Kubernetes
 name: watchlist-step
 type: application
-version: 1.2.153
+version: 1.2.154
 
diff --git a/charts/xmatch_step/Chart.yaml b/charts/xmatch_step/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v2
-appVersion: 6.2.1a82
+appVersion: 6.2.1a83
 description: A Helm chart for Kubernetes
 name: xmatch-step
 type: application

diff --git a/correction_step/pyproject.toml b/correction_step/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "correction"
-version = "6.2.1a82"
+version = "6.2.1a83"
 description = "Correction library for ALeRCE pipeline."
 authors = ["ALeRCE Broker <[email protected]>"]
 readme = "README.md"

diff --git a/early_classification_step/pyproject.toml b/early_classification_step/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "early_classification_step"
-version = "6.2.1a82"
+version = "6.2.1a83"
 description = "Early Classification Step"
 authors = []
 readme = "README.md"
@@ -139,6 +139,7 @@ line-length = 88
 
 
 
+
 
 
 [tool.poetry.group.test.dependencies]
@@ -263,6 +264,7 @@ pytest-docker = "^1.0.1"
 
 
 
+
 
 
 [tool.poetry.group.model.dependencies]

diff --git a/feature_step/pyproject.toml b/feature_step/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "feature-step"
-version = "6.2.1a82"
+version = "6.2.1a83"
 description = ""
 authors = []
 readme = "README.md"

diff --git a/lc_classification_step/pyproject.toml b/lc_classification_step/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "lc_classification_step"
-version = "6.2.1a82"
+version = "6.2.1a83"
 description = "LC Classification Step"
 authors = []
 readme = "README.md"

diff --git a/lightcurve-step/pyproject.toml b/lightcurve-step/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "lightcurve-step"
-version = "6.2.1a82"
+version = "6.2.1a83"
 description = "Lightcurve Step"
 authors = []
 readme = "README.md"

diff --git a/magstats_step/pyproject.toml b/magstats_step/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "magstats-step"
-version = "6.2.1a82"
+version = "6.2.1a83"
 description = ""
 authors = ["ASHuenchuleo <[email protected]>"]
 readme = "README.md"

diff --git a/metadata_step/metadata_step/step.py b/metadata_step/metadata_step/step.py
@@ -1,4 +1,4 @@
-from typing import Dict, List
+from typing import Dict, List, Tuple
 
 from apf.core.step import GenericStep
 from metadata_step.utils.parse import format_detection
@@ -24,9 +24,15 @@ def _format_detection(self, d: Dict, catalogs: Dict):
         extra_fields.pop("prv_candidates", {})
         return format_detection({**d, **extra_fields}, catalogs)
 
-    def _write_metadata_into_db(self, result: List[Dict]):
+    def _write_metadata_into_db(self, result: List[Dict], ps1: Dict[str, List]):
+        ps1_updates = []
+        flattened = sum(ps1.values(), [])
+        for el in flattened:
+            if el["updated"]:
+                ps1_updates.append(el)
+
         with self.db.session() as session:
-            insert_metadata(session, result)
+            insert_metadata(session, result, ps1_updates)
             session.commit()
 
     # Output format: [{oid: OID, ss: SS_DATA, ...}]
@@ -40,8 +46,9 @@ def execute(self, messages: List[Dict]):
             catalogs["gaia"] = get_gaia_catalog(session, oids)
 
         result = [self._format_detection(message, catalogs) for message in messages]
-        return result
+        return result, catalogs["ps1"]
 
-    def post_execute(self, result: List[Dict]):
-        self._write_metadata_into_db(result)
+    def post_execute(self, result: Tuple[List[Dict], List[Dict]]):
+        data, ps1 = result
+        self._write_metadata_into_db(data, ps1)
         return []
diff --git a/metadata_step/metadata_step/utils/database.py b/metadata_step/metadata_step/utils/database.py
@@ -3,7 +3,7 @@
 from contextlib import contextmanager
 from typing import Callable, ContextManager, Dict, List
 
-from sqlalchemy import create_engine, select
+from sqlalchemy import create_engine, select, update
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.orm import sessionmaker, Session
 from db_plugins.db.sql.models import Reference, Gaia_ztf, Ss_ztf, Dataquality, Ps1_ztf
@@ -50,6 +50,12 @@ def _parse_sql_dict(d: Dict):
     return {k: _none_to_nan(v) for k, v in d.items() if not k.startswith("_")}
 
 
+def _parse_ps1(d: Dict):
+    parsed = _parse_sql_dict(d)
+    parsed["updated"] = False
+    return parsed
+
+
 def _create_hashmap(catalog: List[Dict]):
     hashmap: Dict[List] = {}
     for item in catalog:
@@ -63,7 +69,7 @@ def _create_hashmap(catalog: List[Dict]):
 def get_ps1_catalog(session: Session, oids: List):
     stmt = select(Ps1_ztf).where(Ps1_ztf.oid.in_(oids))
     catalog = session.execute(stmt).all()
-    return _create_hashmap([_parse_sql_dict(c[0].__dict__) for c in catalog])
+    return _create_hashmap([_parse_ps1(c[0].__dict__) for c in catalog])
 
 
 def get_gaia_catalog(session: Session, oids: List):
@@ -79,12 +85,12 @@ def _accumulate(data: List):
         acc["dataquality"].append(d["dataquality"])
         acc["reference"].append(d["reference"])
         acc["gaia"].append(d["gaia"])
-        acc["ps1"].extend(d["ps1"])
+        acc["ps1"].append(d["ps1"])
 
     return acc
 
 
-def insert_metadata(session: Session, data: List):
+def insert_metadata(session: Session, data: List, ps1_updates: List):
     accumulated_metadata = _accumulate(data)
     # Reference
     reference_stmt = insert(Reference).values(accumulated_metadata["reference"])
@@ -115,12 +121,11 @@ def insert_metadata(session: Session, data: List):
     # PS1
     ps1_data = list({el["candid"]: el for el in accumulated_metadata["ps1"]}.values())
     ps1_stmt = insert(Ps1_ztf).values(ps1_data)
-    ps1_stmt = ps1_stmt.on_conflict_do_update(
-        constraint="ps1_ztf_pkey",
-        set_=dict(
-            unique1=ps1_stmt.excluded.unique1,
-            unique2=ps1_stmt.excluded.unique2,
-            unique3=ps1_stmt.excluded.unique3,
-        ),
-    )
     session.connection().execute(ps1_stmt)
+
+    for el in ps1_updates:
+        session.connection().execute(
+            update(Ps1_ztf)
+            .where(Ps1_ztf.candid == el["candid"])
+            .values(unique1=el["unique1"], unique2=el["unique2"], unique3=el["unique3"])
+        )
diff --git a/metadata_step/metadata_step/utils/parse.py b/metadata_step/metadata_step/utils/parse.py
@@ -81,16 +81,10 @@ def format_ps1(alert: Dict, catalog={}):
                 continue
 
             for ps1_aux in ps1_filtered:
-                candid = ps1_aux["candid"]
-                saved = candids.get(candid)
-                if not saved:
-                    saved = ps1_aux.copy()
-                saved[f"unique{i}"] = False
-                candids[candid] = saved
-        # new_ps1.extend(list(candids.values()))
-
-    new_ps1.append(alert)
-    return new_ps1
+                ps1_aux[f"unique{i}"] = False
+                ps1_aux["updated"] = True
+
+    return alert
 
 
 # formats each alert to send it to scribe psql

diff --git a/metadata_step/pyproject.toml b/metadata_step/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "metadata-step"
-version = "6.2.1a49"
+version = "6.2.1a50"
 description = ""
 authors = ["Pedro Gallardo <[email protected]>"]
 readme = "README.md"

diff --git a/metadata_step/tests/data/mocks.py b/metadata_step/tests/data/mocks.py
@@ -95,17 +95,6 @@
 ]
 
 ps1_mocks = [
-    {
-        "oid": "ZTF00llmn",
-        "candid": 987654321,
-        "objectidps1": 55.55,
-        "objectidps2": 44.44,
-        "objectidps3": 33.33,
-        "nmtchps": 1,
-        "unique1": True,
-        "unique2": True,
-        "unique3": True,
-    },
     {
         "oid": "ZTF00llmn",
         "candid": 1234567890,

diff --git a/metadata_step/tests/integration/test_database.py b/metadata_step/tests/integration/test_database.py
@@ -68,7 +68,7 @@ def test_metadata_insertion(psql_service):
                 "nframesref": 1,
             },
             "dataquality": {
-                "candid": 1234567890,
+                "candid": 987654321,
                 "oid": "ZTF00llmn",
                 "fid": 1,
                 "xpos": 0,
@@ -80,22 +80,22 @@ def test_metadata_insertion(psql_service):
                 "neargaia": 55.55,
                 "unique1": False,
             },
-            "ps1": [{
+            "ps1": {
                 "oid": "ZTF00llmn",
-                "candid": 1234567890,
+                "candid": 987654321,
                 "objectidps1": 15.11,
                 "objectidps2": 25.22,
                 "objectidps3": 35.33,
                 "nmtchps": 1,
                 "unique1": True,
                 "unique2": True,
                 "unique3": True,
-            }],
+            },
         }
     ]
 
     with conn.session() as session:
-        insert_metadata(session, data)
+        insert_metadata(session, data, [])
         session.commit()
         # perform queries to make sure they were inserted/updated
         ss_result = session.execute(select(Ss_ztf).where(Ss_ztf.oid == "ZTF00llmn"))

diff --git a/prv_candidates_step/pyproject.toml b/prv_candidates_step/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "prv-candidates-step"
-version = "6.2.1a82"
+version = "6.2.1a83"
 description = "Previous Detections Step"
 authors = []
 readme = "README.md"

diff --git a/s3_step/pyproject.toml b/s3_step/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "s3-step"
-version = "6.2.1a82"
+version = "6.2.1a83"
 description = "S3 step"
 authors = []
 readme = "README.md"

diff --git a/scribe/pyproject.toml b/scribe/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "scribe"
-version = "6.2.1a82"
+version = "6.2.1a83"
 description = "Mongo scribe"
 authors = []
 readme = "README.md"

diff --git a/sorting_hat_step/pyproject.toml b/sorting_hat_step/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "sorting-hat-step"
-version = "6.2.1a82"
+version = "6.2.1a83"
 description = "Sorting hat step"
 authors = []
 readme = "README.md"