Repository: meltano/sdk
Pull request #3023 (2025-05-08 19:48:51)

feat(taps): Apply `post_process` automatically to all stream types
With this change, developers won't need to re-implement the call to `post_process` when they override the public `Stream.get_records(...)` method.

A few examples just from MeltanoLabs: https://github.com/search?q=org%3AMeltanoLabs+%22self.post_process%22&type=code.

i.e.

- https://github.com/MeltanoLabs/tap-dbt/blob/49cbf08f91d297d73f2cf19d73614a9a65a31f70/tap_dbt/streams.py#L107-L111
- https://github.com/MeltanoLabs/tap-postgres/blob/d7e11bcaae0d87717771a5a38744d0ec674dabf5/tap_postgres/client.py#L259-L267
- https://github.com/MeltanoLabs/tap-jira/blob/caa12cdc5ee890a440e61df880f76172c9a85e8a/tap_jira/streams.py#L2402-L2406
- https://github.com/MeltanoLabs/tap-messagebird/blob/37a64a4b178e7c1e511ac4481bd69b2e145c6df6/tap_messagebird/streams.py#L98-L102

## Summary by Sourcery

Automatically apply `post_process` method to all stream types during record synchronization

New Features:
- Automatically invoke `post_process` method for all stream types during record sync

Enhancements:
- Updated record processing workflow to consistently apply post-processing across different stream types

Tests:
- Added test case to verify post-processing functionality for streams

sourcery-ai[bot]: <!-- Generated by sourcery-ai[bot]: start review_guide -->

🧙 Sourcery is reviewing your pull request!

---

<details>
<summary>Tips and commands</summary>

#### Interacting with Sourcery

- **Trigger a new review:** Comment `@sourcery-ai review` on the pull request.
- **Continue discussions:** Reply directly to Sourcery's review comments.
- **Generate a GitHub issue from a review comment:** Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with `@sourcery-ai issue` to create an issue from it.
- **Generate a pull request title:** Write `@sourcery-ai` anywhere in the pull
  request title to generate a title at any time. You can also comment
  `@sourcery-ai title` on the pull request to (re-)generate the title at any time.
- **Generate a pull request summary:** Write `@sourcery-ai summary` anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment `@sourcery-ai summary` on the pull request to
  (re-)generate the summary at any time.
- **Generate reviewer's guide:** Comment `@sourcery-ai guide` on the pull
  request to (re-)generate the reviewer's guide at any time.
- **Resolve all Sourcery comments:** Comment `@sourcery-ai resolve` on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
- **Dismiss all Sourcery reviews:** Comment `@sourcery-ai dismiss` on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  `@sourcery-ai review` to trigger a new review!

#### Customizing Your Experience

Access your [dashboard](https://app.sourcery.ai) to:
- Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
- Change the review language.
- Add, remove or edit custom review instructions.
- Adjust other review settings.

#### Getting Help

- [Contact our support team](mailto:support@sourcery.ai) for questions or feedback.
- Visit our [documentation](https://docs.sourcery.ai) for detailed guides and information.
- Keep in touch with the Sourcery team by following us on [X/Twitter](https://x.com/SourceryAI), [LinkedIn](https://www.linkedin.com/company/sourcery-ai/) or [GitHub](https://github.com/sourcery-ai).

</details>

<!-- Generated by sourcery-ai[bot]: end review_guide -->
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/meltano/sdk/pull/3023?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) Report
Attention: Patch coverage is `80.00000%` with `4 lines` in your changes missing coverage. Please review.
> Project coverage is 91.66%. Comparing base [(`f0c0b6b`)](https://app.codecov.io/gh/meltano/sdk/commit/f0c0b6bbbb68a855c6206b0b0f0664151153f6a2?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) to head [(`768afba`)](https://app.codecov.io/gh/meltano/sdk/commit/768afba89d48ee242630dcca0c6774302b79516b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano).

| [Files with missing lines](https://app.codecov.io/gh/meltano/sdk/pull/3023?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) | Patch % | Lines |
|---|---|---|
| [singer\_sdk/streams/core.py](https://app.codecov.io/gh/meltano/sdk/pull/3023?src=pr&el=tree&filepath=singer_sdk%2Fstreams%2Fcore.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano#diff-c2luZ2VyX3Nkay9zdHJlYW1zL2NvcmUucHk=) | 71.42% | [2 Missing and 2 partials :warning: ](https://app.codecov.io/gh/meltano/sdk/pull/3023?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano) |

:x: Your patch check has failed because the patch coverage (80.00%) is below the target coverage (100.00%). You can increase the patch coverage or adjust the [target](https://docs.codecov.com/docs/commit-status#target) coverage.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main    #3023      +/-   ##
==========================================
- Coverage   91.66%   91.66%   -0.01%     
==========================================
  Files          62       62              
  Lines        5314     5313       -1     
  Branches      684      683       -1     
==========================================
- Hits         4871     4870       -1     
  Misses        311      311              
  Partials      132      132              
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/meltano/sdk/pull/3023?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=meltano).

<details><summary> :rocket: New features to boost your workflow: </summary>

- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
codspeed-hq[bot]: <!-- __CODSPEED_PERFORMANCE_REPORT_COMMENT__ -->
<!-- __CODSPEED_INSTRUMENTATION_PERFORMANCE_REPORT_COMMENT__ -->

## [CodSpeed Performance Report](https://codspeed.io/meltano/sdk/branches/post-process-all-streams)

### Merging #3023 will **not alter performance**

<sub>Comparing <code>post-process-all-streams</code> (768afba) with <code>main</code> (f0c0b6b)</sub>



### Summary

`✅ 8` untouched benchmarks  




edgarrmondragon: @sourcery-ai review
edgarrmondragon: @sourcery-ai review
edgarrmondragon: @sourcery-ai review
edgarrmondragon: @sourcery-ai review

Patch:
diff --git a/pyproject.toml b/pyproject.toml
index f4c4384a..a6a8181b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -118,6 +118,7 @@ testing = [
     "moto>=5.0.14",
     "pytest>=7.2.1",
     "pytest-benchmark>=4.0.0",
+    "pytest-github-actions-annotate-failures>=0.3.0",
     "pytest-snapshot>=0.9.0",
     "pytest-subtests>=0.13.1",
     "pytz>=2022.2.1",
@@ -384,7 +385,7 @@ unfixable = [
     "INP",      # Allow implicit namespace packages in tests
     "PLR2004",
     "S101",
-    "SLF001",
+    "SLF001",   # Allow private method access in tests
     "PLC2701",  # Allow usage of private members in tests
     "PLR6301",  # Don't suggest making test methods static, etc.
 ]
diff --git a/samples/sample_target_csv/csv_target_sink.py b/samples/sample_target_csv/csv_target_sink.py
index 4f02bf0b..a6fcf8bd 100644
--- a/samples/sample_target_csv/csv_target_sink.py
+++ b/samples/sample_target_csv/csv_target_sink.py
@@ -26,13 +26,13 @@ class SampleCSVTargetSink(BatchSink):
     def process_batch(self, context: dict) -> None:
         """Write `records_to_drain` out to file."""
         records_to_drain = context["records"]
-        self.logger.info("Draining records...")
+        self.logger.debug("Draining records...")
         records_written = 0
         newfile = False
         openmode = "a"
         outpath = self.target_filepath.absolute()
         if not outpath.is_file():
-            self.logger.info("Writing to new file: %s", outpath)
+            self.logger.debug("Writing to new file: %s", outpath)
             newfile = True
             openmode = "w"
         with outpath.open(openmode, newline="\n", encoding="utf-8") as csvfile:
diff --git a/singer_sdk/_singerlib/__init__.py b/singer_sdk/_singerlib/__init__.py
index a8f32048..2fa0ada5 100644
--- a/singer_sdk/_singerlib/__init__.py
+++ b/singer_sdk/_singerlib/__init__.py
@@ -10,7 +10,9 @@ from singer_sdk.helpers._compat import SingerSDKDeprecationWarning
 if t.TYPE_CHECKING:
     from singer_sdk.singerlib import (
         Catalog,  # noqa: F401
+        CatalogEntry,  # noqa: F401
         Schema,  # noqa: F401
+        SelectionMask,  # noqa: F401
         resolve_schema_references,  # noqa: F401
     )
 
diff --git a/singer_sdk/helpers/_catalog.py b/singer_sdk/helpers/_catalog.py
index c0aa55de..8334c389 100644
--- a/singer_sdk/helpers/_catalog.py
+++ b/singer_sdk/helpers/_catalog.py
@@ -63,6 +63,8 @@ def _pop_deselected_schema(
         selected = mask[property_breadcrumb]
         if not selected:
             schema_at_breadcrumb["properties"].pop(property_name, None)
+            if property_name in schema_at_breadcrumb.get("required", []):
+                schema_at_breadcrumb["required"].remove(property_name)
             continue
 
         if is_object_type(property_def):
@@ -122,14 +124,6 @@ def set_catalog_stream_selected(
     breadcrumb is the path to a property within the stream.
     """
     breadcrumb = breadcrumb or ()
-    if not isinstance(breadcrumb, tuple):  # pragma: no cover
-        msg = (  # type: ignore[unreachable]
-            f"Expected tuple value for breadcrumb '{breadcrumb}'. Got "
-            f"{type(breadcrumb).__name__}"
-        )
-        # TODO: this should be a ValueError, but it's a breaking change.
-        raise ValueError(msg)  # noqa: TRY004
-
     catalog_entry = catalog.get_stream(stream_name)
     if not catalog_entry:
         msg = f"Catalog entry missing for '{stream_name}'. Skipping."
diff --git a/singer_sdk/sinks/core.py b/singer_sdk/sinks/core.py
index f2cbb976..373d5bbc 100644
--- a/singer_sdk/sinks/core.py
+++ b/singer_sdk/sinks/core.py
@@ -164,7 +164,7 @@ class Sink(metaclass=abc.ABCMeta):  # noqa: PLR0904
         self._config = dict(target.config)
         self._pending_batch: dict | None = None
         self.stream_name = stream_name
-        self.logger.info(
+        self.logger.debug(
             "Initializing target sink for stream '%s'...",
             stream_name,
         )
@@ -710,7 +710,7 @@ class Sink(metaclass=abc.ABCMeta):  # noqa: PLR0904
         Setup is executed once per Sink instance, after instantiation. If a Schema
         change is detected, a new Sink is instantiated and this method is called again.
         """
-        self.logger.info("Setting up %s", self.stream_name)
+        self.logger.debug("Setting up %s", self.stream_name)
 
     def clean_up(self) -> None:
         """Perform any clean up actions required at end of a stream.
@@ -719,7 +719,7 @@ class Sink(metaclass=abc.ABCMeta):  # noqa: PLR0904
         that may be in use from other instances of the same sink. Stream name alone
         should not be relied on, it's recommended to use a uuid as well.
         """
-        self.logger.info("Cleaning up %s", self.stream_name)
+        self.logger.debug("Cleaning up %s", self.stream_name)
         self.record_counter_metric.exit()
 
     def process_batch_files(
diff --git a/singer_sdk/streams/core.py b/singer_sdk/streams/core.py
index 94e1ae42..20cfa88a 100644
--- a/singer_sdk/streams/core.py
+++ b/singer_sdk/streams/core.py
@@ -207,7 +207,7 @@ class Stream(metaclass=abc.ABCMeta):  # noqa: PLR0904
 
         if self._tap.mapper is not None:
             self._stream_maps = self._tap.mapper.stream_maps[self.name]
-            self.logger.info(
+            self.logger.debug(
                 "Tap has custom mapper. Using %d provided map(s).",
                 len(self.stream_maps),
             )
@@ -1145,13 +1145,16 @@ class Stream(metaclass=abc.ABCMeta):  # noqa: PLR0904
         Yields:
             Each record from the source.
         """
+        # Type definitions
+        context_element: types.Context | None
+        record: types.Record | None
+        context_list: list[types.Context] | list[dict] | None
+
         # Initialize metrics
         record_counter = metrics.record_counter(self.name)
         timer = metrics.sync_timer(self.name)
 
         record_index = 0
-        context_element: types.Context | None
-        context_list: list[types.Context] | list[dict] | None
         context_list = [context] if context is not None else self.partitions
         selected = self.selected
 
@@ -1178,6 +1181,11 @@ class Stream(metaclass=abc.ABCMeta):  # noqa: PLR0904
                         record, child_context = record_result
                     else:
                         record = record_result
+
+                    record = self.post_process(record, current_context)
+                    if record is None:
+                        continue
+
                     try:
                         self._process_record(
                             record,
@@ -1255,7 +1263,7 @@ class Stream(metaclass=abc.ABCMeta):  # noqa: PLR0904
         msg = f"Beginning {self.replication_method.lower()} sync of '{self.name}'"
         if context:
             msg += f" with context: {context}"
-        self.logger.info("%s...", msg)
+        self.logger.info(msg)
         self.context = MappingProxyType(context) if context else None
 
         # Use a replication signpost, if available
@@ -1415,7 +1423,7 @@ class Stream(metaclass=abc.ABCMeta):  # noqa: PLR0904
     def get_records(
         self,
         context: types.Context | None,
-    ) -> t.Iterable[dict | tuple[dict, dict | None]]:
+    ) -> t.Iterable[types.Record | tuple[dict, dict | None]]:
         """Abstract record generator function. Must be overridden by the child class.
 
         Each record emitted should be a dictionary of property names to their values.
@@ -1487,7 +1495,7 @@ class Stream(metaclass=abc.ABCMeta):  # noqa: PLR0904
         self,
         row: types.Record,
         context: types.Context | None = None,  # noqa: ARG002
-    ) -> dict | None:
+    ) -> types.Record | None:
         """As needed, append or transform raw data to match expected structure.
 
         Optional. This method gives developers an opportunity to "clean up" the results
diff --git a/singer_sdk/streams/rest.py b/singer_sdk/streams/rest.py
index 2bedc36e..43fdb4e9 100644
--- a/singer_sdk/streams/rest.py
+++ b/singer_sdk/streams/rest.py
@@ -623,12 +623,7 @@ class _HTTPStream(Stream, t.Generic[_TToken], metaclass=abc.ABCMeta):  # noqa: P
         Yields:
             One item per (possibly processed) record in the API.
         """
-        for record in self.request_records(context):
-            transformed_record = self.post_process(record, context)
-            if transformed_record is None:
-                # Record filtered out during post_process()
-                continue
-            yield transformed_record
+        yield from self.request_records(context)
 
     # Abstract methods:
 
diff --git a/singer_sdk/streams/sql.py b/singer_sdk/streams/sql.py
index 36a1600f..79341b99 100644
--- a/singer_sdk/streams/sql.py
+++ b/singer_sdk/streams/sql.py
@@ -15,7 +15,7 @@ from singer_sdk.streams.core import REPLICATION_INCREMENTAL, Stream
 
 if t.TYPE_CHECKING:
     from singer_sdk.connectors.sql import FullyQualifiedName
-    from singer_sdk.helpers.types import Context
+    from singer_sdk.helpers.types import Context, Record
     from singer_sdk.tap_base import Tap
 
 
@@ -168,7 +168,7 @@ class SQLStream(Stream, metaclass=abc.ABCMeta):
         return super().effective_schema
 
     # Get records from stream
-    def get_records(self, context: Context | None) -> t.Iterable[dict[str, t.Any]]:
+    def get_records(self, context: Context | None) -> t.Iterable[Record]:
         """Return a generator of record-type dictionary objects.
 
         If the stream has a replication_key value defined, records will be sorted by the
@@ -218,12 +218,9 @@ class SQLStream(Stream, metaclass=abc.ABCMeta):
             query = query.limit(self.ABORT_AT_RECORD_COUNT + 1)
 
         with self.connector._connect() as conn:  # noqa: SLF001
-            for record in conn.execute(query).mappings():
-                transformed_record = self.post_process(dict(record))
-                if transformed_record is None:
-                    # Record filtered out during post_process()
-                    continue
-                yield transformed_record
+            for row in conn.execute(query).mappings():
+                # https://github.com/sqlalchemy/sqlalchemy/discussions/10053#discussioncomment-6344965
+                yield dict(row)
 
     @property
     def is_sorted(self) -> bool:
diff --git a/singer_sdk/target_base.py b/singer_sdk/target_base.py
index 49497967..a6c1d47a 100644
--- a/singer_sdk/target_base.py
+++ b/singer_sdk/target_base.py
@@ -252,7 +252,7 @@ class Target(BaseSingerReader, metaclass=abc.ABCMeta):
         Returns:
             A new sink for the stream.
         """
-        self.logger.info("Initializing '%s' target sink...", self.name)
+        self.logger.debug("Initializing '%s' target sink...", self.name)
         sink_class = self.get_sink_class(stream_name=stream_name)
         sink = sink_class(
             target=self,
@@ -531,7 +531,7 @@ class Target(BaseSingerReader, metaclass=abc.ABCMeta):
             state: TODO
         """
         state_json = json.dumps(state)
-        self.logger.info("Emitting completed target state %s", state_json)
+        self.logger.debug("Emitting completed target state %s", state_json)
         sys.stdout.write(f"{state_json}\n")
         sys.stdout.flush()
 
@@ -718,7 +718,7 @@ class SQLTarget(Target):
         Returns:
             A new sink for the stream.
         """
-        self.logger.info("Initializing '%s' target sink...", self.name)
+        self.logger.debug("Initializing '%s' target sink...", self.name)
         sink_class = self.get_sink_class(stream_name=stream_name)
         sink = sink_class(
             target=self,
diff --git a/uv.lock b/uv.lock
index 2f14bef8..4bb24b70 100644
--- a/uv.lock
+++ b/uv.lock
@@ -800,7 +800,7 @@ name = "exceptiongroup"
 version = "1.3.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "typing-extensions", marker = "python_full_version < '3.13'" },
+    { name = "typing-extensions", marker = "python_full_version < '3.11'" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/0b/9f/a65090624ecf468cdca03533906e7c69ed7588582240cfe7cc9e770b50eb/exceptiongroup-1.3.0.tar.gz", hash = "sha256:b241f5885f560bc56a59ee63ca4c6a8bfa46ae4ad651af316d4e81817bb9fd88", size = 29749, upload-time = "2025-05-10T17:42:51.123Z" }
 wheels = [
@@ -1869,6 +1869,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/f1/9b/952c70bd1fae9baa58077272e7f191f377c86d812263c21b361195e125e6/pytest_codspeed-3.2.0-py3-none-any.whl", hash = "sha256:54b5c2e986d6a28e7b0af11d610ea57bd5531cec8326abe486f1b55b09d91c39", size = 15007, upload-time = "2025-01-31T14:28:24.458Z" },
 ]
 
+[[package]]
+name = "pytest-github-actions-annotate-failures"
+version = "0.3.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pytest" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/39/d4/c54ee6a871eee4a7468e3a8c0dead28e634c0bc2110c694309dcb7563a66/pytest_github_actions_annotate_failures-0.3.0.tar.gz", hash = "sha256:d4c3177c98046c3900a7f8ddebb22ea54b9f6822201b5d3ab8fcdea51e010db7", size = 11248, upload-time = "2025-01-17T22:39:32.722Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/6d/73/7b0b15cb8605ee967b34aa1d949737ab664f94e6b0f1534e8339d9e64ab2/pytest_github_actions_annotate_failures-0.3.0-py3-none-any.whl", hash = "sha256:41ea558ba10c332c0bfc053daeee0c85187507b2034e990f21e4f7e5fef044cf", size = 6030, upload-time = "2025-01-17T22:39:31.701Z" },
+]
+
 [[package]]
 name = "pytest-snapshot"
 version = "0.9.0"
@@ -2393,6 +2405,7 @@ benchmark = [
     { name = "pytest" },
     { name = "pytest-benchmark" },
     { name = "pytest-codspeed" },
+    { name = "pytest-github-actions-annotate-failures" },
     { name = "pytest-snapshot" },
     { name = "pytest-subtests" },
     { name = "pytz" },
@@ -2416,6 +2429,7 @@ dev = [
     { name = "pyarrow-stubs" },
     { name = "pytest" },
     { name = "pytest-benchmark" },
+    { name = "pytest-github-actions-annotate-failures" },
     { name = "pytest-snapshot" },
     { name = "pytest-subtests" },
     { name = "pytz" },
@@ -2459,6 +2473,7 @@ testing = [
     { name = "moto" },
     { name = "pytest" },
     { name = "pytest-benchmark" },
+    { name = "pytest-github-actions-annotate-failures" },
     { name = "pytest-snapshot" },
     { name = "pytest-subtests" },
     { name = "pytz" },
@@ -2522,6 +2537,7 @@ benchmark = [
     { name = "pytest", specifier = ">=7.2.1" },
     { name = "pytest-benchmark", specifier = ">=4.0.0" },
     { name = "pytest-codspeed", specifier = ">=2.2.0" },
+    { name = "pytest-github-actions-annotate-failures", specifier = ">=0.3.0" },
     { name = "pytest-snapshot", specifier = ">=0.9.0" },
     { name = "pytest-subtests", specifier = ">=0.13.1" },
     { name = "pytz", specifier = ">=2022.2.1" },
@@ -2544,6 +2560,7 @@ dev = [
     { name = "pyarrow-stubs", specifier = ">=19.1" },
     { name = "pytest", specifier = ">=7.2.1" },
     { name = "pytest-benchmark", specifier = ">=4.0.0" },
+    { name = "pytest-github-actions-annotate-failures", specifier = ">=0.3.0" },
     { name = "pytest-snapshot", specifier = ">=0.9.0" },
     { name = "pytest-subtests", specifier = ">=0.13.1" },
     { name = "pytz", specifier = ">=2022.2.1" },
@@ -2581,6 +2598,7 @@ testing = [
     { name = "moto", specifier = ">=5.0.14" },
     { name = "pytest", specifier = ">=7.2.1" },
     { name = "pytest-benchmark", specifier = ">=4.0.0" },
+    { name = "pytest-github-actions-annotate-failures", specifier = ">=0.3.0" },
     { name = "pytest-snapshot", specifier = ">=0.9.0" },
     { name = "pytest-subtests", specifier = ">=0.13.1" },
     { name = "pytz", specifier = ">=2022.2.1" },


Test patch:
diff --git a/tests/core/conftest.py b/tests/core/conftest.py
index 6fa9a5ff..8ed44a5e 100644
--- a/tests/core/conftest.py
+++ b/tests/core/conftest.py
@@ -42,9 +42,9 @@ class SimpleTestStream(Stream):
         context: dict | None,
     ) -> t.Iterable[dict[str, t.Any]]:
         """Generate records."""
-        yield {"id": 1, "value": "Egypt"}
-        yield {"id": 2, "value": "Germany"}
-        yield {"id": 3, "value": "India"}
+        yield {"id": 1, "value": "Egypt", "updatedAt": "2021-01-01T00:00:00Z"}
+        yield {"id": 2, "value": "Germany", "updatedAt": "2021-01-01T00:00:01Z"}
+        yield {"id": 3, "value": "India", "updatedAt": "2021-01-01T00:00:02Z"}
 
     @contextmanager
     def with_replication_method(self, method: str | None) -> t.Iterator[None]:
diff --git a/tests/core/snapshots/test_parent_child/test_child_deselected_parent/stderr.log b/tests/core/snapshots/test_parent_child/test_child_deselected_parent/stderr.log
index b00867dd..9c3a2493 100644
--- a/tests/core/snapshots/test_parent_child/test_child_deselected_parent/stderr.log
+++ b/tests/core/snapshots/test_parent_child/test_child_deselected_parent/stderr.log
@@ -1,6 +1,4 @@
-INFO my-tap.parent Beginning full_table sync of 'parent'...
-INFO my-tap.parent Tap has custom mapper. Using 1 provided map(s).
-INFO my-tap.child Beginning full_table sync of 'child' with context: {'pid': 1}...
-INFO my-tap.child Tap has custom mapper. Using 1 provided map(s).
-INFO my-tap.child Beginning full_table sync of 'child' with context: {'pid': 2}...
-INFO my-tap.child Beginning full_table sync of 'child' with context: {'pid': 3}...
+INFO my-tap.parent Beginning full_table sync of 'parent'
+INFO my-tap.child Beginning full_table sync of 'child' with context: {'pid': 1}
+INFO my-tap.child Beginning full_table sync of 'child' with context: {'pid': 2}
+INFO my-tap.child Beginning full_table sync of 'child' with context: {'pid': 3}
diff --git a/tests/core/snapshots/test_parent_child/test_parent_context_fields_in_child/stderr.log b/tests/core/snapshots/test_parent_child/test_parent_context_fields_in_child/stderr.log
index b00867dd..9c3a2493 100644
--- a/tests/core/snapshots/test_parent_child/test_parent_context_fields_in_child/stderr.log
+++ b/tests/core/snapshots/test_parent_child/test_parent_context_fields_in_child/stderr.log
@@ -1,6 +1,4 @@
-INFO my-tap.parent Beginning full_table sync of 'parent'...
-INFO my-tap.parent Tap has custom mapper. Using 1 provided map(s).
-INFO my-tap.child Beginning full_table sync of 'child' with context: {'pid': 1}...
-INFO my-tap.child Tap has custom mapper. Using 1 provided map(s).
-INFO my-tap.child Beginning full_table sync of 'child' with context: {'pid': 2}...
-INFO my-tap.child Beginning full_table sync of 'child' with context: {'pid': 3}...
+INFO my-tap.parent Beginning full_table sync of 'parent'
+INFO my-tap.child Beginning full_table sync of 'child' with context: {'pid': 1}
+INFO my-tap.child Beginning full_table sync of 'child' with context: {'pid': 2}
+INFO my-tap.child Beginning full_table sync of 'child' with context: {'pid': 3}
diff --git a/tests/core/test_catalog_selection.py b/tests/core/test_catalog_selection.py
index 02797eef..c893d838 100644
--- a/tests/core/test_catalog_selection.py
+++ b/tests/core/test_catalog_selection.py
@@ -9,8 +9,10 @@ import pytest
 
 import singer_sdk._singerlib as singer
 from singer_sdk.helpers._catalog import (
+    deselect_all_streams,
     get_selected_schema,
     pop_deselected_record_properties,
+    set_catalog_stream_selected,
 )
 from singer_sdk.typing import ObjectType, PropertiesList, Property, StringType
 
@@ -57,10 +59,11 @@ def schema():
         Property(
             "col_a",
             ObjectType(
-                Property("col_a_1", StringType),
+                Property("col_a_1", StringType, required=True),
                 Property("col_a_2", StringType),
                 Property("col_a_3", StringType),
             ),
+            required=True,
         ),
         Property(
             "col_b",
@@ -68,14 +71,17 @@ def schema():
                 Property("col_b_1", StringType),
                 Property("col_b_2", StringType),
             ),
+            required=True,
         ),
         Property(
             "col_c",
             StringType,
+            required=True,
         ),
         Property(
             "col_d",
             StringType,
+            required=True,
         ),
         Property(
             "col_e",
@@ -199,19 +205,20 @@ def test_schema_selection(
     """Test that schema selection rules are correctly applied to SCHEMA messages."""
     selected_schema = get_selected_schema(stream_name, schema, mask)
     assert (
-        selected_schema["properties"]
+        selected_schema
         == PropertiesList(
             Property(
                 "col_a",
                 ObjectType(
-                    Property("col_a_1", StringType),
+                    Property("col_a_1", StringType, required=True),
                     Property("col_a_3", StringType),
                 ),
+                required=True,
             ),
-            Property("col_d", StringType),
+            Property("col_d", StringType, required=True),
             Property("col_e", StringType),
             Property("col_f", StringType),
-        ).to_dict()["properties"]
+        ).to_dict()
     )
 
 
@@ -248,3 +255,93 @@ def test_record_property_pop(
     assert record_pop == record_selected, (
         f"Expected record={record_selected}, got {record_pop}"
     )
+
+
+def test_deselect_all_streams():
+    """Test that deselect_all_streams sets all streams to not selected."""
+    catalog = singer.Catalog.from_dict(
+        {
+            "streams": [
+                {
+                    "tap_stream_id": "test_stream_a",
+                    "stream": "test_stream_a",
+                    "schema": {
+                        "type": "object",
+                        "properties": {
+                            "col_a": {
+                                "type": "string",
+                            },
+                        },
+                    },
+                    "metadata": [
+                        {
+                            "breadcrumb": (),
+                            "metadata": {"selected": True},
+                        },
+                    ],
+                },
+                {
+                    "tap_stream_id": "test_stream_b",
+                    "stream": "test_stream_b",
+                    "schema": {
+                        "type": "object",
+                        "properties": {
+                            "col_a": {
+                                "type": "string",
+                            },
+                        },
+                    },
+                    "metadata": [
+                        {
+                            "breadcrumb": (),
+                            "metadata": {"selected": True},
+                        },
+                    ],
+                },
+            ],
+        }
+    )
+
+    # Stream is selected
+    assert catalog["test_stream_a"].metadata.root.selected is True
+    assert catalog["test_stream_b"].metadata.root.selected is True
+
+    deselect_all_streams(catalog)
+    # After deselection, should be False
+    assert catalog["test_stream_a"].metadata.root.selected is False
+    assert catalog["test_stream_b"].metadata.root.selected is False
+
+
+def test_set_catalog_stream_selected(catalog_entry_obj: singer.CatalogEntry):
+    """Test set_catalog_stream_selected for stream and property selection."""
+    catalog = singer.Catalog({catalog_entry_obj.tap_stream_id: catalog_entry_obj})
+    stream_name = catalog_entry_obj.tap_stream_id
+
+    # Set stream to not selected
+    set_catalog_stream_selected(catalog, stream_name, selected=False)
+    assert catalog_entry_obj.metadata.root.selected is False
+    # Set stream to selected
+    set_catalog_stream_selected(catalog, stream_name, selected=True)
+    assert catalog_entry_obj.metadata.root.selected is True
+    # Set a property to not selected
+    breadcrumb = ("properties", "col_a")
+    set_catalog_stream_selected(
+        catalog, stream_name, selected=False, breadcrumb=breadcrumb
+    )
+    assert catalog_entry_obj.metadata[breadcrumb].selected is False
+    # Set a property to selected
+    set_catalog_stream_selected(
+        catalog, stream_name, selected=True, breadcrumb=breadcrumb
+    )
+    assert catalog_entry_obj.metadata[breadcrumb].selected is True
+
+    with pytest.raises(
+        ValueError,
+        match="Catalog entry missing for 'non-existent-stream'",
+    ):
+        set_catalog_stream_selected(
+            catalog,
+            "non-existent-stream",
+            selected=True,
+            breadcrumb=breadcrumb,
+        )
diff --git a/tests/core/test_streams.py b/tests/core/test_streams.py
index 5d445ed0..f5825c30 100644
--- a/tests/core/test_streams.py
+++ b/tests/core/test_streams.py
@@ -32,6 +32,7 @@ if t.TYPE_CHECKING:
     import requests_mock
 
     from singer_sdk import Stream, Tap
+    from singer_sdk.helpers.types import Context, Record
     from tests.core.conftest import SimpleTestTap
 
 CONFIG_START_DATE = "2021-01-01"
@@ -741,3 +742,40 @@ def test_stream_class_selection(tap_class, input_catalog, selection):
     assert all(
         tap.streams[stream].selected is selection[stream] for stream in selection
     )
+
+
+def test_post_process_drops_record(tap: Tap):
+    """Test post-processing is applied to records."""
+
+    class DropsRecord(SimpleTestStream):
+        def post_process(
+            self,
+            record: Record,
+            context: Context | None,  # noqa: ARG002
+        ) -> Record | None:
+            # Drop even IDs
+            return None if record["id"] % 2 == 0 else record
+
+    stream = DropsRecord(tap)
+    records = list(stream._sync_records(None, write_messages=False))
+    assert records == [
+        {"id": 1, "value": "Egypt", "updatedAt": "2021-01-01T00:00:00Z"},
+        {"id": 3, "value": "India", "updatedAt": "2021-01-01T00:00:02Z"},
+    ]
+
+
+def test_post_process_transforms_record(tap: Tap):
+    """Test post-processing is applied to records."""
+
+    class TransformsRecord(SimpleTestStream):
+        def post_process(
+            self,
+            record: Record,
+            context: Context | None,  # noqa: ARG002
+        ) -> Record | None:
+            record["extra"] = "transformed"
+            return record
+
+    stream = TransformsRecord(tap)
+    records = stream._sync_records(None, write_messages=False)
+    assert all(record["extra"] == "transformed" for record in records)