Emit record count for BATCH messages only

`RECORD` messages are already counted by the SDK goes through, and processing of them here goes through `insert_batch_files_via_internal_stage` by default

Author: ReubenFrankel

Diff for: target_snowflake/sinks.py

@@ -171,7 +171,7 @@ def insert_batch_files_via_internal_stage(
         self,
         full_table_name: str,
         files: t.Sequence[str],
-    ) -> None:
+    ) -> int:
         """Process a batch file with the given batch context.
 
         Args:
@@ -206,9 +206,6 @@ def insert_batch_files_via_internal_stage(
                     file_format=file_format,
                 )
 
-            with self.record_counter_metric as counter:
-                counter.increment(record_count)
-
         finally:
             self.logger.debug("Cleaning up after batch processing")
             self.connector.drop_file_format(file_format=file_format)
@@ -220,6 +217,8 @@ def insert_batch_files_via_internal_stage(
                     if os.path.exists(file_path):  # noqa: PTH110
                         os.remove(file_path)  # noqa: PTH107
 
+        return record_count
+
     def process_batch_files(
         self,
         encoding: BaseBatchFileEncoding,
@@ -235,7 +234,7 @@ def process_batch_files(
             NotImplementedError: If the batch file encoding is not supported.
         """
         if encoding.format == BatchFileFormat.JSONL:
-            self.insert_batch_files_via_internal_stage(
+            record_count = self.insert_batch_files_via_internal_stage(
                 full_table_name=self.full_table_name,
                 files=files,
             )
@@ -245,6 +244,9 @@ def process_batch_files(
                 msg,
             )
 
+        with self.record_counter_metric as counter:
+            counter.increment(record_count)
+
     # TODO: remove after https://github.com/meltano/sdk/issues/1819 is fixed
     def _singer_validate_message(self, record: dict) -> None:
         """Ensure record conforms to Singer Spec.