Def start(changesets) do changesets

|> split
end

defp split(changeset) do
changeset
|> Enum.split_with(fn %Ecto.Changeset{} = item_changeset -> item_changeset.valid? end)
|> changeset_valid?()
end

defp changeset_valid?(valid?) do
case valid? do
{valid_item_changeset, []} ->
valid_item_changeset
|> Stream.map(fn %Ecto.Changeset{} = item_changeset -> Ecto.Changeset.apply_changes(item_changeset) end)
|> Stream.map(fn item -> item_into_map(item) end)
# add timestamps
|> Enum.map(fn struct ->
with_inserted = Map.put(struct, :inserted_at, Timestamp.now())
Map.put(with_inserted, :updated_at, Timestamp.now())
end)

{_valid_item_changesets, invalid_item_changesets} -> "Changeset to struct reply: not valid changeset"
end
end

defp item_into_map(item) do
item
|> Map.from_struct()
|> Stream.reject(fn {_key, nil} ->
true

{_key, %_struct{}} ->
# rejects __meta__: #Ecto.Schema.Metadata<:built, "items">
# and association: #Ecto.Association.NotLoaded<association :association is not loaded>
true

_other ->
false
end)
|> Enum.into(%{}) # not really necessary since `insert_all` also accepts a list of lists.
end

спасибо!

Только Timestamp.now() замени, оно у меня с другого модуля тянет зависимость.

Я бы заменил все Stream тут на обычный Enum Например, Stream.reject |> Enum.into(%{}) превращается в один обычный Enum.reduce Проблема в том, что Stream сам по себе очень медленный и Stream.map |> Stream.map |> Stream.map медленнее чем Enum.map |> Enum.map |> Enum.map (и точно медленнее чем один Enum.map) Но это так, на заметку, это не сильная оптимизация будет

Не знал, я думал он наоборот за один проход все делает если Stream.map пайпать, а Enum.map за 3 соответственно.

Ну, у тебя в одном проходе вызовется 3 функции Либо в трёх по одной Вытащить элемент из списка и применить к нему функцию будет быстрее, чем достать cписок обёрнутых функций из структуры, выполнить их, проверить на окончание стрима, проверить на наличие аккумуляторов, положить новый enum в стрим и т.д.

Понял, спасибо

не факт что так будет всегда, надеемся на жит