File Streaming

Dockerfile

@@ -14,6 +14,8 @@ WORKDIR /code
 
 COPY libs /code/libs
 
+RUN apt-get update && apt-get install -y jq
+
 RUN --mount=type=cache,target=/root/.cache/uv \
     --mount=type=bind,source=uv.lock,target=uv.lock \
     --mount=type=bind,source=pyproject.toml,target=pyproject.toml \

apps/etl/transform/sources/emdat.py

@@ -1,10 +1,11 @@
-import json
 import logging
 
+from pystac_monty.sources.common import DataType, File
 from pystac_monty.sources.emdat import EMDATDataSource, EMDATTransformer
 
 from apps.etl.models import ExtractionData
 from apps.etl.transform.sources.handler import BaseTransformerHandler
+from apps.etl.utils import write_into_temp_file
 from main.celery import app
 
 logger = logging.getLogger(__name__)
@@ -16,13 +17,13 @@ class EMDATTransformHandler(BaseTransformerHandler[EMDATTransformer, EMDATDataSo
 
     @classmethod
     def get_schema_data(cls, extraction_obj: ExtractionData):
-        with extraction_obj.resp_data.open() as file_data:
-            data = json.loads(file_data.read())
+        with extraction_obj.resp_data.open("rb") as f:
+            data = f.read()
+        data_file = write_into_temp_file(data)
 
-        return cls.transformer_schema(
-            source_url=extraction_obj.url,
-            data=data,
-        )
+        data_source = {"source_url": extraction_obj.url, "source_data": File(path=data_file.name, data_type=DataType.FILE)}
+
+        return cls.transformer_schema(data_source)
 
     @staticmethod
     @app.task

apps/etl/transform/sources/gdacs.py

@@ -1,43 +1,69 @@
-import json
 import logging
 
+from pystac_monty.sources.common import DataType, File, GdacsDataSourceType, GdacsEpisodes, GenericDataSource
 from pystac_monty.sources.gdacs import (
-    GDACSDataSource,
     GDACSDataSourceType,
+    GDACSDataSourceV3,
     GDACSTransformer,
 )
 
 from apps.etl.transform.sources.handler import BaseTransformerHandler
+from apps.etl.utils import write_into_temp_file
 from main.celery import app
 
 logger = logging.getLogger(__name__)
 
 # FIXME: start_end_handler base zzz
 
 
-class GDACSTransformHandler(BaseTransformerHandler[GDACSTransformer, GDACSDataSource]):
+class GDACSTransformHandler(BaseTransformerHandler[GDACSTransformer, GDACSDataSourceV3]):
     transformer_class = GDACSTransformer
-    transformer_schema = GDACSDataSource
+    transformer_schema = GDACSDataSourceV3
 
     @classmethod
     def get_schema_data(cls, extraction_object):
-        data = extraction_object.resp_data.read()
+        with extraction_object.resp_data.open("rb") as f:
+            file_content = f.read()
+        data_file = write_into_temp_file(file_content)
+
         episodes = []
         event_objects = extraction_object.child_extractions.all()
         for episode_obj in event_objects:
-            episodes_data_dict = {}
-            event_episode_data = episode_obj.resp_data.read()
-            episodes_data_dict[GDACSDataSourceType.EVENT] = (episode_obj.url, json.loads(event_episode_data))
-            geometry_objects = episode_obj.child_extractions.all()
+            with episode_obj.resp_data.open("rb") as f:
+                file_content = f.read()
+            episode_data_temp_file = write_into_temp_file(file_content)
+
+            event_episode_data = GdacsEpisodes(
+                type=GDACSDataSourceType.EVENT,
+                data=GenericDataSource(
+                    source_url=episode_obj.url, data_source=File(path=episode_data_temp_file.name, data_type=DataType.FILE)
+                ),
+            )
+            geometry_object = episode_obj.child_extractions.all().first()
+
+            with geometry_object.resp_data.open("rb") as f:
+                file_content = f.read()
+            geometry_detail_temp_file = write_into_temp_file(file_content)
+            geometry_episode_data = GdacsEpisodes(
+                type=GDACSDataSourceType.GEOMETRY,
+                data=GenericDataSource(
+                    source_url=geometry_object.url,
+                    data_source=File(path=geometry_detail_temp_file.name, data_type=DataType.FILE),
+                ),
+            )
 
-            for geometry_detail in geometry_objects:
-                geometry_episode_data = geometry_detail.resp_data.read()
-                episodes_data_dict[GDACSDataSourceType.GEOMETRY] = (geometry_detail.url, json.loads(geometry_episode_data))
+            episode_data_tuple = (event_episode_data, geometry_episode_data)
+            episodes.append(episode_data_tuple)
 
-            episodes.append(episodes_data_dict)
-        return cls.transformer_schema(source_url=extraction_object.url, data=json.loads(data), episodes=episodes)
+        return cls.transformer_schema(
+            data=GdacsDataSourceType(
+                source_url=extraction_object.url,
+                event_data=File(path=data_file.name, data_type=DataType.FILE),
+                episodes=episodes,
+            )
+        )
 
     @staticmethod
-    @app.task
+    @app.task(rate_limit="50/m")
     def task(extraction_id):
         GDACSTransformHandler().handle_transformation(extraction_id)

apps/etl/transform/sources/glide.py

@@ -1,9 +1,11 @@
 import json
 
+from pystac_monty.sources.common import DataType, File, GenericDataSource
 from pystac_monty.sources.glide import GlideDataSource, GlideTransformer
 
 from apps.etl.models import ExtractionData
 from apps.etl.transform.sources.handler import BaseTransformerHandler
+from apps.etl.utils import write_into_temp_file
 from main.celery import app
 
 
@@ -13,10 +15,16 @@ class GlideTransformHandler(BaseTransformerHandler[GlideTransformer, GlideDataSo
 
     @classmethod
     def get_schema_data(cls, extraction_obj):
-        with extraction_obj.resp_data.open() as file_data:
-            data = file_data.read()
-
-        return cls.transformer_schema(source_url=extraction_obj.url, data=data)
+        with extraction_obj.resp_data.open("rb") as f:
+            data = f.read()
+        data_file = write_into_temp_file(data)
+
+        return cls.transformer_schema(
+            data=GenericDataSource(
+                source_url=extraction_obj.url,
+                data_source=File(path=data_file.name, data_type=DataType.FILE),
+            )
+        )
 
     @staticmethod
     @app.task

apps/etl/transform/sources/idu.py

@@ -1,19 +1,25 @@
-from pystac_monty.sources.idu import IDUDataSource, IDUTransformer
+from pystac_monty.sources.common import DataType, File
+from pystac_monty.sources.idu import IDUDataSourceV2, IDUTransformer
 
 from apps.etl.transform.sources.handler import BaseTransformerHandler
+from apps.etl.utils import write_into_temp_file
 from main.celery import CeleryQueue, app
 
 
-class IDUTransformHandler(BaseTransformerHandler[IDUTransformer, IDUDataSource]):
+class IDUTransformHandler(BaseTransformerHandler[IDUTransformer, IDUDataSourceV2]):
     transformer_class = IDUTransformer
-    transformer_schema = IDUDataSource
+    transformer_schema = IDUDataSourceV2
 
     @classmethod
     def get_schema_data(cls, extraction_obj):
-        with extraction_obj.resp_data.open() as file_data:
-            data = file_data.read()
+        with extraction_obj.resp_data.open("rb") as file_data:
+            file_content = file_data.read()
 
-        return cls.transformer_schema(source_url=extraction_obj.url, data=data)
+        data_file = write_into_temp_file(content=file_content)
+
+        data_source = {"source_url": extraction_obj.url, "source_data": File(path=data_file.name, data_type=DataType.FILE)}
+
+        return cls.transformer_schema(data_source)
 
     @staticmethod
     @app.task(queue=CeleryQueue.DEFAULT)

pyproject.toml

@@ -34,6 +34,7 @@ dependencies = [
     "colorlog",
     "requests_cache",
     "termcolor",
+    "ijson>=3.4.0",
 ]
 
 [tool.uv.sources]