Stream a collection's features as an NDJSON lakehouse feed at GET .../export

GET /collections/{collectionId}/export writes the collection's moving features as
NDJSON — one Feature per line, the temporal geometry as MF-JSON — from a
server-side cursor, so memory stays bounded for any collection size and a lake
consumer (DuckDB / MobilityDuck / Spark) can ingest the stream directly.

Author: estebanzimanyi

resource/collection/Export.py

@@ -0,0 +1,48 @@
+# GET /collections/{collectionId}/export
+# Lakehouse bulk feed: the collection's moving features streamed as NDJSON (one
+# Feature per line, the temporal geometry as MF-JSON) from a server-side cursor,
+# so memory is bounded regardless of collection size. A lake consumer (DuckDB /
+# MobilityDuck / Spark) ingests the stream directly.
+import json
+
+from utils import handle_error
+
+
+def export_collection(self, collection_id, connection, cursor):
+    cursor.execute("SELECT id FROM collections WHERE id = %s", (collection_id,))
+    if cursor.fetchone() is None:
+        handle_error(self, 404, f"Collection '{collection_id}' not found")
+        return
+
+    self.send_response(200)
+    self.send_header("Content-Type", "application/x-ndjson")
+    self.end_headers()
+
+    # Server-side cursor: rows are fetched in batches, so a large collection
+    # streams at bounded memory.
+    stream = connection.cursor(name="mfapi_export")
+    stream.itersize = 1000
+    try:
+        stream.execute(
+            """
+            SELECT mf.id,
+                   coalesce(mf.properties, '{}'::jsonb)::text,
+                   asMFJSON(tg.trajectory)
+            FROM moving_features mf
+            LEFT JOIN temporal_geometries tg
+                   ON mf.id = tg.feature_id AND mf.collection_id = tg.collection_id
+            WHERE mf.collection_id = %s
+            ORDER BY mf.id
+            """,
+            (collection_id,),
+        )
+        for fid, properties, tgeom in stream:
+            feature = {
+                "type": "Feature",
+                "id": fid,
+                "properties": json.loads(properties),
+                "temporalGeometry": json.loads(tgeom) if tgeom else None,
+            }
+            self.wfile.write((json.dumps(feature) + "\n").encode("utf-8"))
+    finally:
+        stream.close()

server.py

@@ -9,6 +9,7 @@
 from resource.collections.Create import post_collections
 from resource.collections.Retrieve import get_collections
 from resource.collection.Retrieve import get_collection_id
+from resource.collection.Export import export_collection
 from resource.collection.Delete import delete_collection
 from resource.collection.Replace import put_collection
 from resource.moving_features.Create import post_collection_items, insert_feature
@@ -143,7 +144,13 @@ def do_GET(self):
         elif self.path == '/collections':
             self.get_collections(connection, cursor)
             return
-        
+
+        # /collections/{collectionId}/export — lakehouse NDJSON feed
+        elif self.path.startswith('/collections/') and urlparse(self.path).path.endswith('/export'):
+            collection_id = urlparse(self.path).path.split('/')[2]
+            self.export_collection(collection_id, connection, cursor)
+            return
+
         # /collections/{collectionId}
         elif self.path.startswith('/collections/'):
             path_only = urlparse(self.path).path
@@ -378,6 +385,9 @@ def delete_single_moving_feature(self, collectionId, mFeature_id, connection, cu
     def put_single_moving_feature(self, collectionId, mFeature_id, connection, cursor):
         put_single_moving_feature(self, collectionId, mFeature_id, connection, cursor)
 
+    def export_collection(self, collection_id, connection, cursor):
+        export_collection(self, collection_id, connection, cursor)
+
 
 ## Resource Temporal Geometry Sequence
     #Get