✨ Add catalogPath column to datasets table

apps/anomalist/cli.py

@@ -148,10 +148,9 @@ def load_datasets_new_ids(source_engine: Engine) -> list[int]:
     target_engine = production_or_master_engine()
 
     # Get new datasets
-    # TODO: replace by real catalogPath when we have it in MySQL
     q = """SELECT
         id,
-        CONCAT(namespace, "/", version, "/", shortName) as catalogPath
+        catalogPath
     FROM datasets
     """
     source_datasets = read_sql(q, source_engine)

etl/grapher_import.py

@@ -87,7 +87,10 @@ def upsert_dataset(
             short_name=short_name,
         )
         ds = gm.Dataset.from_dataset_metadata(
-            dataset.metadata, namespace=namespace, user_id=int(cast(str, config.GRAPHER_USER_ID))
+            dataset.metadata,
+            namespace=namespace,
+            user_id=int(cast(str, config.GRAPHER_USER_ID)),
+            table_names=dataset.table_names,
         ).upsert(session)
 
         session.commit()

etl/grapher_model.py

@@ -570,7 +570,7 @@ class Dataset(Base):
         Index("datasets_createdByUserId", "createdByUserId"),
         Index("datasets_dataEditedByUserId", "dataEditedByUserId"),
         Index("datasets_metadataEditedByUserId", "metadataEditedByUserId"),
-        Index("unique_short_name_version_namespace", "shortName", "version", "namespace", unique=True),
+        Index("datasets_catalogpath", "catalogPath", unique=True),
     )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, init=False)
@@ -591,13 +591,13 @@ class Dataset(Base):
     metadataEditedAt: Mapped[datetime] = mapped_column(DateTime, default=func.utc_timestamp())
     isArchived: Mapped[Optional[int]] = mapped_column(TINYINT(1), server_default=text("'0'"), default=0)
     sourceChecksum: Mapped[Optional[str]] = mapped_column(VARCHAR(64), default=None)
+    catalogPath: Mapped[Optional[str]] = mapped_column(VARCHAR(767), default=None)
+    tables: Mapped[Optional[list]] = mapped_column(JSON, default=None)
 
     def upsert(self, session: Session) -> "Dataset":
         cls = self.__class__
         q = select(cls).where(
-            cls.shortName == self.shortName,
-            cls.version == self.version,
-            cls.namespace == self.namespace,
+            cls.catalogPath == self.catalogPath,
         )
         ds = session.scalar(q)
         if not ds:
@@ -611,6 +611,8 @@ def upsert(self, session: Session) -> "Dataset":
             ds.isPrivate = self.isPrivate
             ds.updatePeriodDays = self.updatePeriodDays
             ds.nonRedistributable = self.nonRedistributable
+            ds.catalogPath = self.catalogPath
+            ds.tables = self.tables
             ds.updatedAt = datetime.utcnow()
             ds.metadataEditedAt = datetime.utcnow()
             ds.dataEditedAt = datetime.utcnow()
@@ -619,17 +621,13 @@ def upsert(self, session: Session) -> "Dataset":
         ds.sourceChecksum = None
 
         session.add(ds)
-
-        # select added object to get its id
-        q = select(cls).where(
-            cls.shortName == self.shortName,
-            cls.version == self.version,
-            cls.namespace == self.namespace,
-        )
-        return session.scalars(q).one()
+        session.flush()  # Ensure the object is written to the database and its ID is generated
+        return ds
 
     @classmethod
-    def from_dataset_metadata(cls, metadata: catalog.DatasetMeta, namespace: str, user_id: int) -> "Dataset":
+    def from_dataset_metadata(
+        cls, metadata: catalog.DatasetMeta, namespace: str, user_id: int, table_names: List[str]
+    ) -> "Dataset":
         assert metadata.title
         return cls(
             shortName=metadata.short_name,
@@ -643,6 +641,8 @@ def from_dataset_metadata(cls, metadata: catalog.DatasetMeta, namespace: str, us
             isPrivate=not metadata.is_public,
             updatePeriodDays=metadata.update_period_days,
             nonRedistributable=metadata.non_redistributable,
+            catalogPath=f"{namespace}/{metadata.version}/{metadata.short_name}",
+            tables=table_names,
         )
 
     @classmethod
@@ -751,9 +751,8 @@ def upsert(self, session: Session) -> "Source":
             ds.description = self.description
 
         session.add(ds)
-
-        # select added object to get its id
-        return session.scalars(self._upsert_select).one()
+        session.flush()  # Ensure the object is written to the database and its ID is generated
+        return ds
 
     @classmethod
     def from_catalog_source(cls, source: catalog.Source, dataset_id: int) -> "Source":
@@ -1163,13 +1162,8 @@ def upsert(self, session: Session) -> "Variable":
                 ds.sort = self.sort
 
         session.add(ds)
-
-        # select added object to get its id
-        q = select(cls).where(
-            cls.shortName == self.shortName,
-            cls.datasetId == self.datasetId,
-        )
-        return session.scalars(q).one()
+        session.flush()  # Ensure the object is written to the database and its ID is generated
+        return ds
 
     @classmethod
     def from_variable_metadata(