perf(ingestor_livetiming): add opt-in concurrency/parallelism for historical data ingestion

requirements.txt

@@ -1,5 +1,6 @@
 aiohttp==3.12.13
 aiomqtt==2.4.0
+async-lru==2.1.0
 beautifulsoup4==4.13.4
 cachetools==5.5.2
 fastapi==0.115.13

src/openf1/services/ingestor_livetiming/core/objects.py

@@ -152,7 +152,12 @@ def _get_collections_cls_by_name() -> dict[str, type[Collection]]:
 def get_collections(meeting_key: int, session_key: int) -> list[Collection]:
     """Returns the instances of all available collections for a given session"""
     collections = [
-        cls(meeting_key=meeting_key, session_key=session_key)
+        cls(
+            meeting_key=meeting_key,
+            session_key=session_key,
+            name=cls.name,
+            source_topics=cls.source_topics,
+        )
         for cls in _get_collections_cls_by_name().values()
     ]
     collections = sorted(collections, key=lambda c: c.__class__.name)

src/openf1/services/ingestor_livetiming/core/processing/main.py

@@ -9,10 +9,13 @@
 
 
 def process_message(
-    meeting_key: int, session_key: int, message: Message
+    meeting_key: int,
+    session_key: int,
+    message: Message,
 ) -> dict[str, list[Document]]:
     """Processes a message from a given topic and returns processed documents grouped
-    by collection"""
+    by collection. If at least one collection name is given, only those collections are used.
+    """
     # Select collections which could use this message
     selected_collections = get_topics_to_collections_mapping(
         meeting_key=meeting_key, session_key=session_key
@@ -34,9 +37,7 @@ def process_message(
 
 
 def process_messages(
-    meeting_key: int,
-    session_key: int,
-    messages: list[Message],
+    meeting_key: int, session_key: int, messages: list[Message]
 ) -> dict[str, list[Document]]:
     """Processes messages and returns the generated documents by collection"""
     docs_buf = defaultdict(dict)

src/openf1/services/ingestor_livetiming/historical/README.md

@@ -81,3 +81,16 @@ To ingest processed documents of all the available collections for season `2024`
 ```bash
 python -m openf1.services.ingestor_livetiming.historical.main ingest-season 2024
 ```
+
+### Enable parallelization
+
+> [!NOTE]
+> Enabling parallelization will require significantly more system resources (CPU, RAM), especially when ingesting seasons. Use at your own discretion.
+
+To speed up data ingestion, the `--parallel` flag can be used to enable parallel message processing for `get-messages`, `get-processed-documents`, and `ingest-` commands. To ingest processed documents of all the available collections for meeting `1242` (year `2024`):
+
+```bash
+python -m openf1.services.ingestor_livetiming.historical.main ingest-meeting 2024 1242 --parallel
+```
+
+For `ingest-meeting` and `ingest-season`, the flags `--by-session` and by `--by-meeting` can be enabled for a hybrid style of ingestion, reducing resource consumption at the cost of increased ingestion time.