fix: limit objects to insert

ex_cubic_ingestion/lib/ex_cubic_ingestion/schema/cubic_load.ex

@@ -97,15 +97,15 @@ defmodule ExCubicIngestion.Schema.CubicLoad do
   From a list of S3 objects defined as maps, filter in only new objects since last load and
   insert them in database.
   """
-  @spec insert_new_from_objects_with_table([map()], CubicTable.t()) ::
+  @spec insert_new_from_objects_with_table([map()], CubicTable.t(), integer()) ::
           {:ok,
            {CubicOdsTableSnapshot.t() | nil,
             [
               {t(), CubicOdsLoadSnapshot.t() | nil, CubicTable.t(),
                CubicOdsTableSnapshot.t() | nil}
             ]}}
           | {:error, term()}
-  def insert_new_from_objects_with_table(objects, table) do
+  def insert_new_from_objects_with_table(objects, table, limit \\ 500) do
     # get ODS snapshot if available
     ods_table_snapshot = CubicOdsTableSnapshot.get_by(table_id: table.id)
 
@@ -122,12 +122,14 @@ defmodule ExCubicIngestion.Schema.CubicLoad do
     # create a list of objects that have a last modified that is later than
     # the last object we have in database
     new_objects =
-      Enum.filter(objects, fn object ->
+      objects
+      |> Enum.filter(fn object ->
         last_modified = parse_and_drop_msec(object.last_modified)
 
         is_nil(last_inserted_load) or
           DateTime.compare(last_modified, last_inserted_load.s3_modified) == :gt
       end)
+      |> Enum.take(limit)
 
     if Enum.empty?(new_objects) do
       {:ok, {ods_table_snapshot, []}}

ex_cubic_ingestion/test/ex_cubic_ingestion/schema/cubic_load_test.exs

@@ -154,6 +154,22 @@ defmodule ExCubicIngestion.Schema.CubicLoadTest do
                ]}} = CubicLoad.insert_new_from_objects_with_table(ods_load_objects, ods_table)
     end
 
+    test "limiting list of ODS objects", %{
+      ods_table: ods_table,
+      ods_load_objects: ods_load_objects
+    } do
+      {:ok, {_last_ods_table_snapshot, new_ods_load_recs}} =
+        CubicLoad.insert_new_from_objects_with_table(ods_load_objects, ods_table, 1)
+
+      assert [List.first(ods_load_objects).key] ==
+               new_ods_load_recs
+               |> Enum.reverse()
+               |> Enum.map(fn {%CubicLoad{status: "ready", s3_key: s3_key}, _ods_load_snapshot,
+                               _table, _ods_table_snapshot} ->
+                 s3_key
+               end)
+    end
+
     test "providing an empty list of objects", %{
       dmap_table: dmap_table
     } do