feat: Settings write-back

docs/conf.py

@@ -152,6 +152,7 @@
 # -- Options for intersphinx -----------------------------------------------------------
 # https://www.sphinx-doc.org/en/master/usage/extensions/intersphinx.html#configuration
 intersphinx_mapping = {
+    "blinker": ("https://blinker.readthedocs.io/en/stable/", None),
     "requests": ("https://requests.readthedocs.io/en/latest/", None),
     "python": ("https://docs.python.org/3/", None),
 }

docs/guides/index.md

@@ -8,4 +8,5 @@ The following pages contain useful information for developers building on top of
 porting
 pagination-classes
 custom-clis
+signals
 ```

docs/guides/signals.md

@@ -0,0 +1,49 @@
+# Signals
+
+This guide will show you how to use the built-in [Blinker](inv:blinker:std:doc#index) signals in the Singer SDK.
+
+## Settings write-back
+
+The SDK provides a signal that allows you to write back settings to the configuration file. This is useful if you want to update the configuration file with new settings that were set during the run, like a `refresh_token`.
+
+```python
+import requests
+from singer_sdk.authenticators import OAuthAuthenticator
+from singer_sdk.plugin_base import PluginBase
+
+
+class RefreshTokenAuthenticator(OAuthAuthenticator):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.refresh_token = self.config["refresh_token"]
+
+    @property
+    def oauth_request_body(self):
+        return {
+            "client_id": self.config["client_id"],
+            "client_secret": self.config["client_secret"],
+            "grant_type": "refresh_token",
+            "refresh_token": self.refresh_token,
+            "user_type": "Location",
+        }
+
+    def update_access_token(self):
+        token_response = requests.post(
+            self.auth_endpoint,
+            headers=self._oauth_headers,
+            data=auth_request_payload,
+            timeout=60,
+        )
+        token_response.raise_for_status()
+        token_json = token_response.json()
+
+        self.access_token = token_json["access_token"]
+        self.refresh_token = token_json["refresh_token"]
+        PluginBase.config_updated.send(self, refresh_token=self.refresh_token)
+```
+
+In the example above, the `RefreshTokenAuthenticator` class is a subclass of `OAuthAuthenticator` that calls `PluginBase.config_updated.send` to send a signal to update the `refresh_token` in tap's configuration.
+
+```{note}
+Only when a single file is passed via the `--config` command line option, the SDK will write back the settings to the same file.
+```

pyproject.toml

@@ -41,6 +41,7 @@ license = "Apache-2.0"
 python = ">=3.8"
 backoff = { version = ">=2.0.0", python = "<4" }
 backports-datetime-fromisoformat = { version = ">=2.0.1", python = "<3.11" }
+blinker = ">=1.7.0"
 click = "~=8.0"
 cryptography = ">=3.4.6"
 fs = ">=2.4.16"

singer_sdk/authenticators.py

@@ -7,7 +7,6 @@
 import typing as t
 import warnings
 from datetime import timedelta
-from types import MappingProxyType
 from urllib.parse import parse_qs, urlencode, urlsplit, urlunsplit
 
 import jwt
@@ -19,6 +18,7 @@
 
 if t.TYPE_CHECKING:
     import logging
+    from types import MappingProxyType
 
     from pendulum import DateTime
 
@@ -93,19 +93,19 @@ def __init__(self, stream: RESTStream) -> None:
             stream: A stream for a RESTful endpoint.
         """
         self.tap_name: str = stream.tap_name
-        self._config: dict[str, t.Any] = dict(stream.config)
+        self._config = stream.config
         self._auth_headers: dict[str, t.Any] = {}
         self._auth_params: dict[str, t.Any] = {}
         self.logger: logging.Logger = stream.logger
 
     @property
-    def config(self) -> t.Mapping[str, t.Any]:
+    def config(self) -> MappingProxyType:
         """Get stream or tap config.
 
         Returns:
             A frozen (read-only) config dictionary map.
         """
-        return MappingProxyType(self._config)
+        return self._config
 
     @property
     def auth_headers(self) -> dict: