fix: Replace pendulum with ab_datetime_now() and fix linting issues

airbyte/progress.py

@@ -422,14 +422,19 @@ def _log_sync_start(self) -> None:
         )
 
     def _log_sync_cancel(self) -> None:
-        print(f"Canceled `{self.job_description}` sync at `{pendulum.now().format('HH:mm:ss')}`.")
+        print(
+            f"Canceled `{self.job_description}` sync at `{ab_datetime_now().strftime('%H:%M:%S')}`."
+        )
         self._send_telemetry(
             state=EventState.CANCELED,
             event_type=EventType.SYNC,
         )
 
     def _log_stream_read_start(self, stream_name: str) -> None:
-        print(f"Read started on stream `{stream_name}` at `{pendulum.now().format('HH:mm:ss')}`...")
+        print(
+            f"Read started on stream `{stream_name}` at "
+            f"`{ab_datetime_now().strftime('%H:%M:%S')}`..."
+        )
         self.stream_read_start_times[stream_name] = time.time()
 
     def log_stream_start(self, stream_name: str) -> None:
@@ -559,7 +564,7 @@ def log_success(
 
         print(
             f"Completed `{self.job_description}` sync at "
-            f"`{pendulum.now().format('HH:mm:ss')}`{streams_str}."
+            f"`{ab_datetime_now().strftime('%H:%M:%S')}`{streams_str}."
         )
         self._log_read_metrics()
         self._send_telemetry(

airbyte/sources/base.py

@@ -208,9 +208,9 @@ def set_primary_keys(
         - Stream names are not validated by PyAirbyte. If the stream name
           does not exist in the catalog, the override may be ignored.
         """
-        self._primary_key_overrides.update({
-            k.lower(): v if isinstance(v, list) else [v] for k, v in kwargs.items()
-        })
+        self._primary_key_overrides.update(
+            {k.lower(): v if isinstance(v, list) else [v] for k, v in kwargs.items()}
+        )
 
     def _log_warning_preselected_stream(self, streams: str | list[str]) -> None:
         """Logs a warning message indicating stream selection which are not selected yet."""
@@ -701,12 +701,14 @@ def print_samples(
                     )
 
                 for record in dataset:
-                    table.add_row(*[
-                        escape(str(val))
-                        for key, val in record.items()
-                        # Exclude internal Airbyte columns.
-                        if key not in internal_cols
-                    ])
+                    table.add_row(
+                        *[
+                            escape(str(val))
+                            for key, val in record.items()
+                            # Exclude internal Airbyte columns.
+                            if key not in internal_cols
+                        ]
+                    )
 
             console.print(table)
 
@@ -746,11 +748,13 @@ def _read_with_catalog(
         * Send out telemetry on the performed sync (with information about which source was used and
           the type of the cache)
         """
-        with as_temp_files([
-            self._hydrated_config,
-            catalog.model_dump_json(),
-            state.to_state_input_file_text() if state else "[]",
-        ]) as [
+        with as_temp_files(
+            [
+                self._hydrated_config,
+                catalog.model_dump_json(),
+                state.to_state_input_file_text() if state else "[]",
+            ]
+        ) as [
             config_file,
             catalog_file,
             state_file,
@@ -769,7 +773,7 @@ def _read_with_catalog(
             )
             for message in progress_tracker.tally_records_read(message_generator):
                 if stop_event and stop_event.is_set():
-                    progress_tracker._log_sync_cancel()
+                    progress_tracker._log_sync_cancel()  # noqa: SLF001
                     return
 
                 yield message