From 9b7a6bfd5e96f871713c083ad67a46c7dad75f88 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Edgar=20Ram=C3=ADrez=20Mondrag=C3=B3n?=
 <edgarrm358@gmail.com>
Date: Tue, 19 Nov 2024 19:21:16 -0600
Subject: [PATCH] fix: Respect standard Singer stream metadata for key
 properties, replication key and replication method

- https://hub.meltano.com/singer/spec/#metadata
- https://github.com/singer-io/getting-started/blob/master/docs/DISCOVERY_MODE.md#metadata
---
 singer_sdk/_singerlib/catalog.py |  1 +
 singer_sdk/streams/core.py       | 19 +++++++++++++++----
 2 files changed, 16 insertions(+), 4 deletions(-)

diff --git a/singer_sdk/_singerlib/catalog.py b/singer_sdk/_singerlib/catalog.py
index 986272e3c..c1281e091 100644
--- a/singer_sdk/_singerlib/catalog.py
+++ b/singer_sdk/_singerlib/catalog.py
@@ -87,6 +87,7 @@ class StreamMetadata(Metadata):
     """Stream metadata."""
 
     table_key_properties: t.Sequence[str] | None = None
+    replication_key: str | None = None
     forced_replication_method: str | None = None
     valid_replication_keys: list[str] | None = None
     schema_name: str | None = None
diff --git a/singer_sdk/streams/core.py b/singer_sdk/streams/core.py
index 009726b7f..423de6994 100644
--- a/singer_sdk/streams/core.py
+++ b/singer_sdk/streams/core.py
@@ -1277,10 +1277,21 @@ def apply_catalog(self, catalog: singer.Catalog) -> None:
 
         catalog_entry = catalog.get_stream(self.name)
         if catalog_entry:
-            self.primary_keys = catalog_entry.key_properties
-            self.replication_key = catalog_entry.replication_key
-            if catalog_entry.replication_method:
-                self.forced_replication_method = catalog_entry.replication_method
+            stream_metadata = catalog_entry.metadata.root
+
+            self.primary_keys = (
+                catalog_entry.key_properties or stream_metadata.table_key_properties
+            )
+            self.replication_key = (
+                catalog_entry.replication_key or stream_metadata.replication_key
+            )
+
+            replication_method = (
+                catalog_entry.replication_method
+                or stream_metadata.forced_replication_method
+            )
+            if replication_method:
+                self.forced_replication_method = replication_method
 
     def _get_state_partition_context(
         self,