Fix pre-commit config - calling different Python versions

Author: lukaszlacinski

.github/workflows/pre-commit.yaml

@@ -12,7 +12,7 @@ jobs:
     - uses: actions/checkout@v3
     - uses: actions/setup-python@v4
       with:
-        python-version: '3.10'
+        python-version: '3.13'
     - run: python -m pip install pre-commit
       shell: bash
     - run: python -m pip freeze --local

src/consumer.py

@@ -21,9 +21,7 @@ def process_messages(self, messages):
                 data = json.loads(msg.value())
                 messages_data.append(data)
             except json.JSONDecodeError as e:
-                logging.error(
-                    f"Data deserialization error at offset {msg.offset()}: {e}."
-                )
+                logging.error(f"Data deserialization error at offset {msg.offset()}: {e}.")
                 return False
         return messages_data
 
@@ -39,9 +37,7 @@ def start(self):
                 logging.info(f"Consumed {len(messages)} messages")
                 first_msg = messages[0]
                 offset = 0 if not first_msg.offset() else first_msg.offset()
-                self.seek_partition = TopicPartition(
-                    first_msg.topic(), first_msg.partition(), offset
-                )
+                self.seek_partition = TopicPartition(first_msg.topic(), first_msg.partition(), offset)
 
                 messages_data = self.process_messages(messages)
                 if not messages_data:

src/globus.py

@@ -87,9 +87,7 @@ def ingest(self, messages_data):
     def post(self, message_data):
         item = message_data.get("data").get("payload").get("item")
         try:
-            globus_response = self.search_client.get_subject(
-                self.esgf_index, item.get("id")
-            )
+            globus_response = self.search_client.get_subject(self.esgf_index, item.get("id"))
         except SearchAPIError as e:
             if e.http_status == 404:
                 item["assets"] = self.normalize_assets(item.get("assets"))
@@ -121,9 +119,7 @@ def json_patch(self, message_data):
             return None
         item = globus_response.data.get("entries")[0].get("content")
         item["assets"] = self.denormalize_assets(item.get("assets"))
-        patched_item = jsonpatch.apply_patch(
-            item, payload.get("patch").get("operations")
-        )
+        patched_item = jsonpatch.apply_patch(item, payload.get("patch").get("operations"))
         patched_item["assets"] = self.normalize_assets(patched_item.get("assets"))
         gmeta_entry = self.gmetaentry(patched_item)
         return gmeta_entry

src/main.py

@@ -3,21 +3,16 @@
 from consumer import KafkaConsumerService
 from globus import ConsumerSearchClient
 from producer import KafkaProducer
-from settings.consumer import (event_stream, globus_search,
-                               globus_search_client_credentials)
+from settings.consumer import event_stream, globus_search, globus_search_client_credentials
 from settings.producer import error_event_stream
 
-logging.basicConfig(
-    format="%(asctime)s - %(levelname)s - %(message)s", level=logging.INFO
-)
+logging.basicConfig(format="%(asctime)s - %(levelname)s - %(message)s", level=logging.INFO)
 
 
 if __name__ == "__main__":
     error_producer = KafkaProducer(config=error_event_stream.get("config"))
 
-    message_processor = ConsumerSearchClient(
-        globus_search_client_credentials, globus_search.get("index"), error_producer
-    )
+    message_processor = ConsumerSearchClient(globus_search_client_credentials, globus_search.get("index"), error_producer)
 
     consumer_service = KafkaConsumerService(
         event_stream.get("config"),

src/producer.py

@@ -32,13 +32,9 @@ def delivery_report(err, msg):
             if err is not None:
                 logger.error(f"Delivery failed for message {msg.key()}: {err}")
             else:
-                logger.info(
-                    f"Message {msg.key()} successfully delivered to {msg.topic()} [{msg.partition()}] at offset {msg.offset()}"
-                )
+                logger.info(f"Message {msg.key()} successfully delivered to {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")
             delivery_reports.append((err, msg))
 
-        self.producer.produce(
-            topic=topic, key=key, value=value, callback=delivery_report
-        )
+        self.producer.produce(topic=topic, key=key, value=value, callback=delivery_report)
         self.producer.flush()
         return delivery_reports

src/settings/producer.py

@@ -4,8 +4,6 @@
 
 from dotenv import load_dotenv
 
-from src.utils import load_access_control_policy
-
 # Load the .env file
 load_dotenv()