chore: Enable SIM Ruff checks (#1509)

pyproject.toml

@@ -238,6 +238,7 @@ select = [
     "PT",   # flake8-pytest-style
     "RSE",  # flake8-raise
     "RET",  # flake8-return
+    "SIM",  # flake8-simplify
 ]
 src = ["samples", "singer_sdk", "tests"]
 target-version = "py37"

samples/sample_tap_hostile/hostile_streams.py

@@ -34,7 +34,7 @@ def get_records(self, context: dict | None) -> Iterable[dict | tuple[dict, dict]
         return (
             {
                 key: self.get_random_lowercase_string()
-                for key in self.schema["properties"].keys()
+                for key in self.schema["properties"]
             }
             for _ in range(10)
         )

singer_sdk/_singerlib/catalog.py

@@ -191,10 +191,12 @@ def get_standard_metadata(
             if schema_name:
                 root.schema_name = schema_name
 
-            for field_name in schema.get("properties", {}).keys():
-                if key_properties and field_name in key_properties:
-                    entry = Metadata(inclusion=Metadata.InclusionType.AUTOMATIC)
-                elif valid_replication_keys and field_name in valid_replication_keys:
+            for field_name in schema.get("properties", {}):
+                if (
+                    key_properties
+                    and field_name in key_properties
+                    or (valid_replication_keys and field_name in valid_replication_keys)
+                ):
                     entry = Metadata(inclusion=Metadata.InclusionType.AUTOMATIC)
                 else:
                     entry = Metadata(inclusion=Metadata.InclusionType.AVAILABLE)

singer_sdk/configuration/_dict_config.py

@@ -43,7 +43,7 @@ def parse_environment_config(
     logger.debug("Loading configuration from %s", dotenv_path)
     DotEnv(dotenv_path).set_as_environment_variables()
 
-    for config_key in config_schema["properties"].keys():
+    for config_key in config_schema["properties"]:
         env_var_name = prefix + config_key.upper().replace("-", "_")
         if env_var_name in os.environ:
             env_var_value = os.environ[env_var_name]

singer_sdk/connectors/sql.py

@@ -854,11 +854,7 @@ def merge_sql_types(
                 if issubclass(
                     generic_type,
                     (sqlalchemy.types.String, sqlalchemy.types.Unicode),
-                ):
-                    # If length None or 0 then is varchar max ?
-                    if (opt_len is None) or (opt_len == 0):
-                        return opt
-                elif isinstance(
+                ) or issubclass(
                     generic_type,
                     (sqlalchemy.types.String, sqlalchemy.types.Unicode),
                 ):

singer_sdk/helpers/_flattening.py

@@ -236,7 +236,7 @@ def _flatten_schema(
 
     for k, v in schema_node["properties"].items():
         new_key = flatten_key(k, parent_keys, separator)
-        if "type" in v.keys():
+        if "type" in v:
             if "object" in v["type"] and "properties" in v and level < max_level:
                 items.extend(
                     _flatten_schema(

singer_sdk/helpers/_state.py

@@ -235,17 +235,19 @@ def finalize_state_progress_markers(stream_or_partition_state: dict) -> dict | N
     """Promote or wipe progress markers once sync is complete."""
     signpost_value = stream_or_partition_state.pop(SIGNPOST_MARKER, None)
     stream_or_partition_state.pop(STARTING_MARKER, None)
-    if PROGRESS_MARKERS in stream_or_partition_state:
-        if "replication_key" in stream_or_partition_state[PROGRESS_MARKERS]:
-            # Replication keys valid (only) after sync is complete
-            progress_markers = stream_or_partition_state[PROGRESS_MARKERS]
-            stream_or_partition_state["replication_key"] = progress_markers.pop(
-                "replication_key",
-            )
-            new_rk_value = progress_markers.pop("replication_key_value")
-            if signpost_value and _greater_than_signpost(signpost_value, new_rk_value):
-                new_rk_value = signpost_value
-            stream_or_partition_state["replication_key_value"] = new_rk_value
+    if (
+        PROGRESS_MARKERS in stream_or_partition_state
+        and "replication_key" in stream_or_partition_state[PROGRESS_MARKERS]
+    ):
+        # Replication keys valid (only) after sync is complete
+        progress_markers = stream_or_partition_state[PROGRESS_MARKERS]
+        stream_or_partition_state["replication_key"] = progress_markers.pop(
+            "replication_key",
+        )
+        new_rk_value = progress_markers.pop("replication_key_value")
+        if signpost_value and _greater_than_signpost(signpost_value, new_rk_value):
+            new_rk_value = signpost_value
+        stream_or_partition_state["replication_key_value"] = new_rk_value
     # Wipe and return any markers that have not been promoted
     return reset_state_progress_markers(stream_or_partition_state)
 

singer_sdk/helpers/_typing.py

@@ -124,10 +124,7 @@ def is_datetime_type(type_dict: dict) -> bool:
             "Did you forget to define a property in the stream schema?",
         )
     if "anyOf" in type_dict:
-        for type_dict in type_dict["anyOf"]:
-            if is_datetime_type(type_dict):
-                return True
-        return False
+        return any(is_datetime_type(type_dict) for type_dict in type_dict["anyOf"])
     if "type" in type_dict:
         return type_dict.get("format") == "date-time"
     raise ValueError(

singer_sdk/plugin_base.py

@@ -87,7 +87,7 @@ def __init__(
         """
         if not config:
             config_dict = {}
-        elif isinstance(config, str) or isinstance(config, PurePath):
+        elif isinstance(config, (str, PurePath)):
             config_dict = read_json_file(config)
         elif isinstance(config, list):
             config_dict = {}
@@ -339,7 +339,7 @@ def print_about(cls: type[PluginBase], format: str | None = None) -> None:
         elif format == "markdown":
             max_setting_len = cast(
                 int,
-                max(len(k) for k in info["settings"]["properties"].keys()),
+                max(len(k) for k in info["settings"]["properties"]),
             )
 
             # Set table base for markdown

singer_sdk/sinks/core.py

@@ -327,7 +327,7 @@ def _parse_timestamps_in_record(
             schema: TODO
             treatment: TODO
         """
-        for key in record.keys():
+        for key in record:
             datelike_type = get_datelike_property_type(schema["properties"][key])
             if datelike_type:
                 try:
@@ -479,12 +479,14 @@ def process_batch_files(
                 storage = StorageTarget.from_url(head)
 
             if encoding.format == BatchFileFormat.JSONL:
-                with storage.fs(create=False) as batch_fs:
-                    with batch_fs.open(tail, mode="rb") as file:
-                        if encoding.compression == "gzip":
-                            file = gzip_open(file)
-                        context = {"records": [json.loads(line) for line in file]}
-                        self.process_batch(context)
+                with storage.fs(create=False) as batch_fs, batch_fs.open(
+                    tail,
+                    mode="rb",
+                ) as file:
+                    if encoding.compression == "gzip":
+                        file = gzip_open(file)
+                    context = {"records": [json.loads(line) for line in file]}
+                    self.process_batch(context)
             else:
                 raise NotImplementedError(
                     f"Unsupported batch encoding format: {encoding.format}",

singer_sdk/sinks/sql.py

@@ -199,7 +199,7 @@ def conform_schema(self, schema: dict) -> dict:
         """
         conformed_schema = copy(schema)
         conformed_property_names = {
-            key: self.conform_name(key) for key in conformed_schema["properties"].keys()
+            key: self.conform_name(key) for key in conformed_schema["properties"]
         }
         self._check_conformed_names_not_duplicated(conformed_property_names)
         conformed_schema["properties"] = {

singer_sdk/streams/core.py

@@ -1266,12 +1266,14 @@ def get_batches(
         ):
             filename = f"{prefix}{sync_id}-{i}.json.gz"
             with batch_config.storage.fs() as fs:
-                with fs.open(filename, "wb") as f:
-                    # TODO: Determine compression from config.
-                    with gzip.GzipFile(fileobj=f, mode="wb") as gz:
-                        gz.writelines(
-                            (json.dumps(record) + "\n").encode() for record in chunk
-                        )
+                # TODO: Determine compression from config.
+                with fs.open(filename, "wb") as f, gzip.GzipFile(
+                    fileobj=f,
+                    mode="wb",
+                ) as gz:
+                    gz.writelines(
+                        (json.dumps(record) + "\n").encode() for record in chunk
+                    )
                 file_url = fs.geturl(filename)
 
             yield batch_config.encoding, [file_url]

singer_sdk/streams/rest.py

@@ -103,11 +103,7 @@ def _url_encode(val: str | datetime | bool | int | list[str]) -> str:
         Returns:
             TODO
         """
-        if isinstance(val, str):
-            result = val.replace("/", "%2F")
-        else:
-            result = str(val)
-        return result
+        return val.replace("/", "%2F") if isinstance(val, str) else str(val)
 
     def get_url(self, context: dict | None) -> str:
         """Get stream entity URL.
@@ -199,10 +195,7 @@ def response_error_message(self, response: requests.Response) -> str:
             str: The error message
         """
         full_path = urlparse(response.url).path or self.path
-        if 400 <= response.status_code < 500:
-            error_type = "Client"
-        else:
-            error_type = "Server"
+        error_type = "Client" if 400 <= response.status_code < 500 else "Server"
 
         return (
             f"{response.status_code} {error_type} Error: "
@@ -430,8 +423,7 @@ def update_sync_costs(
         """
         call_costs = self.calculate_sync_cost(request, response, context)
         self._sync_costs = {
-            k: self._sync_costs.get(k, 0) + call_costs.get(k, 0)
-            for k in call_costs.keys()
+            k: self._sync_costs.get(k, 0) + call_costs.get(k, 0) for k in call_costs
         }
         return self._sync_costs
 

singer_sdk/tap_base.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import abc
+import contextlib
 import json
 from enum import Enum
 from pathlib import Path, PurePath
@@ -197,10 +198,8 @@ def run_connection_test(self) -> bool:
                     "Skipping direct invocation.",
                 )
                 continue
-            try:
+            with contextlib.suppress(MaxRecordsLimitException):
                 stream.sync()
-            except MaxRecordsLimitException:
-                pass
         return True
 
     @final

singer_sdk/testing/runners.py

@@ -231,7 +231,7 @@ def input(self) -> IO[str]:
             if self.input_io:
                 self._input = self.input_io
             elif self.input_filepath:
-                self._input = open(self.input_filepath)
+                self._input = open(self.input_filepath)  # noqa: SIM115
         return cast(IO[str], self._input)
 
     @input.setter

tests/core/configuration/test_dict_config.py

@@ -89,9 +89,8 @@ def test_get_env_var_config_not_parsable():
             "PLUGIN_TEST_PROP1": "hello",
             "PLUGIN_TEST_PROP3": '["repeated"]',
         },
-    ):
-        with pytest.raises(ValueError, match="A bracketed list was detected"):
-            parse_environment_config(CONFIG_JSONSCHEMA, "PLUGIN_TEST_")
+    ), pytest.raises(ValueError, match="A bracketed list was detected"):
+        parse_environment_config(CONFIG_JSONSCHEMA, "PLUGIN_TEST_")
 
 
 def test_merge_config_sources(config_file1, config_file2):

tests/core/test_connector_sql.py

@@ -157,15 +157,19 @@ def test_deprecated_functions_warn(self, connector):
             connector.connection
 
     def test_connect_calls_engine(self, connector):
-        with mock.patch.object(SQLConnector, "_engine") as mock_engine:
-            with connector._connect() as _:
-                mock_engine.connect.assert_called_once()
+        with mock.patch.object(
+            SQLConnector,
+            "_engine",
+        ) as mock_engine, connector._connect() as _:
+            mock_engine.connect.assert_called_once()
 
     def test_connect_calls_connect(self, connector):
         attached_engine = connector._engine
-        with mock.patch.object(attached_engine, "connect") as mock_conn:
-            with connector._connect() as _:
-                mock_conn.assert_called_once()
+        with mock.patch.object(
+            attached_engine,
+            "connect",
+        ) as mock_conn, connector._connect() as _:
+            mock_conn.assert_called_once()
 
     def test_connect_raises_on_operational_failure(self, connector):
         with pytest.raises(