add static method to get schema migration details (#164)

Author: 3nids

pum/schema_migrations.py

@@ -99,6 +99,75 @@ def schemas_with_migrations(connection: psycopg.Connection) -> list[str]:
         cursor = SqlContent(query).execute(connection, parameters={})
         return [row[0] for row in (cursor._pum_results or [])]
 
+    @staticmethod
+    def schemas_with_migration_details(
+        connection: psycopg.Connection,
+    ) -> list[dict]:
+        """Return detailed migration info for each schema that has a pum_migrations table.
+
+        For each schema, returns the module name, current version, first installation
+        date and latest upgrade date (if different from the installation date).
+
+        Args:
+            connection: The database connection.
+
+        Returns:
+            list[dict]: A list of dicts with keys: schema, module, version,
+                        installed_date, upgrade_date (None if never upgraded).
+        """
+        schemas = SchemaMigrations.schemas_with_migrations(connection)
+        details = []
+        for schema in schemas:
+            table_id = psycopg.sql.SQL(".").join(
+                [psycopg.sql.Identifier(schema), psycopg.sql.Identifier(MIGRATION_TABLE_NAME)]
+            )
+            query = psycopg.sql.SQL(
+                """
+                SELECT
+                    module,
+                    version,
+                    installed_date,
+                    CASE WHEN upgrade_date > installed_date THEN upgrade_date END AS upgrade_date,
+                    beta_testing
+                FROM (
+                    SELECT
+                        module,
+                        (SELECT version FROM {table} ORDER BY version DESC, date_installed DESC LIMIT 1) AS version,
+                        MIN(date_installed) AS installed_date,
+                        MAX(date_installed) AS upgrade_date,
+                        (SELECT beta_testing FROM {table} ORDER BY version DESC, date_installed DESC LIMIT 1) AS beta_testing
+                    FROM {table}
+                    GROUP BY module
+                ) sub
+                """
+            )
+            try:
+                cursor = SqlContent(query).execute(connection, parameters={"table": table_id})
+                for row in cursor._pum_results or []:
+                    details.append(
+                        {
+                            "schema": schema,
+                            "module": row[0],
+                            "version": row[1],
+                            "installed_date": row[2],
+                            "upgrade_date": row[3],
+                            "beta_testing": row[4],
+                        }
+                    )
+            except Exception:
+                logger.warning(f"Could not read migration details from schema {schema}")
+                details.append(
+                    {
+                        "schema": schema,
+                        "module": None,
+                        "version": None,
+                        "installed_date": None,
+                        "upgrade_date": None,
+                        "beta_testing": None,
+                    }
+                )
+        return details
+
     def exists_in_other_schemas(self, connection: psycopg.Connection) -> list[str]:
         """Check if the schema_migrations information table exists in other schemas.
 

test/test_schema_migrations.py

@@ -172,3 +172,79 @@ def test_schemas_with_migrations(self) -> None:
             # Test exists_in_other_schemas for sm2
             other_schemas = sm2.exists_in_other_schemas(conn)
             self.assertEqual(other_schemas, ["public"])
+
+    def test_schemas_with_migration_details_empty(self) -> None:
+        """Test schemas_with_migration_details when no migrations exist."""
+        with psycopg.connect(f"service={self.pg_service}") as conn:
+            details = SchemaMigrations.schemas_with_migration_details(conn)
+            self.assertEqual(details, [])
+
+    def test_schemas_with_migration_details_single_schema(self) -> None:
+        """Test schemas_with_migration_details with a single installed module."""
+        test_dir = Path("test") / "data" / "single_changelog"
+        cfg = PumConfig(test_dir, pum={"module": "test_module"})
+        sm = SchemaMigrations(cfg)
+
+        with psycopg.connect(f"service={self.pg_service}") as conn:
+            sm.create(connection=conn)
+            sm.set_baseline(connection=conn, version="1.2.3")
+
+            details = SchemaMigrations.schemas_with_migration_details(conn)
+            self.assertEqual(len(details), 1)
+            detail = details[0]
+            self.assertEqual(detail["schema"], "public")
+            self.assertEqual(detail["module"], "test_module")
+            self.assertEqual(detail["version"], "1.2.3")
+            self.assertIsNotNone(detail["installed_date"])
+            self.assertIsNone(detail["upgrade_date"])  # No upgrade yet
+            self.assertFalse(detail["beta_testing"])
+
+    def test_schemas_with_migration_details_after_upgrade(self) -> None:
+        """Test schemas_with_migration_details shows upgrade_date after a second baseline."""
+        test_dir = Path("test") / "data" / "single_changelog"
+        cfg = PumConfig(test_dir, pum={"module": "test_module"})
+        sm = SchemaMigrations(cfg)
+
+        with psycopg.connect(f"service={self.pg_service}") as conn:
+            sm.create(connection=conn)
+            sm.set_baseline(connection=conn, version="1.2.3", commit=True)
+
+        # Use a separate connection/transaction so now() differs
+        with psycopg.connect(f"service={self.pg_service}") as conn:
+            sm.set_baseline(connection=conn, version="1.2.4", commit=True)
+
+            details = SchemaMigrations.schemas_with_migration_details(conn)
+            self.assertEqual(len(details), 1)
+            detail = details[0]
+            self.assertEqual(detail["version"], "1.2.4")
+            self.assertIsNotNone(detail["installed_date"])
+            self.assertIsNotNone(detail["upgrade_date"])  # Upgrade happened
+
+    def test_schemas_with_migration_details_multiple_schemas(self) -> None:
+        """Test schemas_with_migration_details with modules in different schemas."""
+        test_dir = Path("test") / "data" / "single_changelog"
+        cfg1 = PumConfig(test_dir, pum={"module": "module_a", "migration_table_schema": "public"})
+        cfg2 = PumConfig(
+            test_dir,
+            pum={"module": "module_b", "migration_table_schema": "pum_custom_migrations_schema"},
+        )
+        sm1 = SchemaMigrations(cfg1)
+        sm2 = SchemaMigrations(cfg2)
+
+        with psycopg.connect(f"service={self.pg_service}") as conn:
+            sm1.create(connection=conn, allow_multiple_modules=True)
+            sm1.set_baseline(connection=conn, version="1.0.0")
+
+            sm2.create(connection=conn, allow_multiple_modules=True)
+            sm2.set_baseline(connection=conn, version="2.0.0")
+
+            details = SchemaMigrations.schemas_with_migration_details(conn)
+            self.assertEqual(len(details), 2)
+
+            by_schema = {d["schema"]: d for d in details}
+            self.assertIn("public", by_schema)
+            self.assertIn("pum_custom_migrations_schema", by_schema)
+            self.assertEqual(by_schema["public"]["module"], "module_a")
+            self.assertEqual(by_schema["public"]["version"], "1.0.0")
+            self.assertEqual(by_schema["pum_custom_migrations_schema"]["module"], "module_b")
+            self.assertEqual(by_schema["pum_custom_migrations_schema"]["version"], "2.0.0")