Merge pull request #91 from ankhers/ecto-2

[WIP] Support for Ecto 2.0

Author: michalmuskala

.travis.yml

@@ -6,10 +6,10 @@ cache:
       - ~/.mongodb
 
 elixir:
-  - 1.1
+  - 1.2.6
 
 otp_release:
-  - 18.1
+  - 18.2
 
 before_install:
   - npm install -g mongodb-version-manager

config/test.exs

@@ -1,4 +1,6 @@
 use Mix.Config
 
-config :logger, :handle_otp_reports, true
-config :logger, :handle_sasl_reports, true
+config :logger,
+  level: :info,
+  handle_otp_reports: true
+  handle_sasl_reports: true

examples/simple/lib/simple.ex

@@ -16,7 +16,7 @@ defmodule Simple.Repo do
 end
 
 defmodule Weather do
-  use Ecto.Model
+  use Ecto.Schema
 
   @primary_key {:id, :binary_id, autogenerate: true}
 
@@ -38,4 +38,18 @@ defmodule Simple do
          select: w
     Simple.Repo.all(query)
   end
+
+  def sample_insert do
+    %Weather{}
+    |> Ecto.Changeset.change(%{})
+    |> Simple.Repo.insert()
+  end
+
+  def sample_update do
+    {:ok, weather} = sample_insert()
+
+    weather
+    |> Ecto.Changeset.change(%{city: "NYC"})
+    |> Simple.Repo.update
+  end
 end

examples/simple/mix.exs

@@ -9,7 +9,7 @@ defmodule Simple.Mixfile do
 
   def application do
     [mod: {Simple.App, []},
-     applications: [:mongodb_ecto, :ecto]]
+     applications: [:mongodb_ecto, :ecto, :logger]]
   end
 
   defp deps do

lib/mongo_ecto.ex

@@ -418,13 +418,36 @@ defmodule Mongo.Ecto do
   end
 
   @doc false
-  # TODO: handle date and time
+  def ensure_all_started(repo, type) do
+    {_, opts} = repo.__pool__
+    with {:ok, pool} <- DBConnection.ensure_all_started(opts, type),
+         {:ok, mongo} <- Application.ensure_all_started(:mongodb, type),
+      do: {:ok, pool ++ mongo}
+  end
+
+  @doc false
+  def loaders(:time,      type), do: [&load_time/1, type]
+  def loaders(:date,      type), do: [&load_date/1, type]
   def loaders(:datetime,  type), do: [&load_datetime/1, type]
   def loaders(:binary_id, type), do: [&load_objectid/1, type]
   def loaders(:uuid,      type), do: [&load_binary/1,   type]
   def loaders(:binary,    type), do: [&load_binary/1,   type]
   def loaders(_base,      type), do: [type]
 
+  defp load_time(%BSON.DateTime{} = time) do
+    {{_,_,_}, time} = BSON.DateTime.to_datetime(time)
+    {:ok, time}
+  end
+  defp load_time(_),
+    do: :error
+
+  defp load_date(%BSON.DateTime{} = date) do
+    {date, {_, _, _, _}} = BSON.DateTime.to_datetime(date)
+    {:ok, date}
+  end
+  defp load_date(_),
+    do: :error
+
   defp load_datetime(%BSON.DateTime{} = datetime),
     do: {:ok, BSON.DateTime.to_datetime(datetime)}
   defp load_datetime(_),
@@ -446,13 +469,24 @@ defmodule Mongo.Ecto do
   defp load_objectid(_), do: :error
 
   @doc false
-  # TODO: handle date and time
+  def dumpers(:time,      type), do: [type, &dump_time/1]
+  def dumpers(:date,      type), do: [type, &dump_date/1]
   def dumpers(:datetime,  type), do: [type, &dump_datetime/1]
   def dumpers(:binary_id, type), do: [type, &dump_objectid/1]
   def dumpers(:uuid,      type), do: [type, &dump_binary(&1, :uuid)]
   def dumpers(:binary,    type), do: [type, &dump_binary(&1, :generic)]
   def dumpers(_base,      type), do: [type]
 
+  defp dump_time({_, _, _, _} = time),
+    do: {:ok, BSON.DateTime.from_datetime({{0, 0, 0}, time})}
+  defp dump_time(_),
+    do: :error
+
+  defp dump_date({_, _, _} = date),
+    do: {:ok, BSON.DateTime.from_datetime({date, {0, 0, 0, 0}})}
+  defp dump_date(_),
+    do: :error
+
   defp dump_datetime({{_, _, _}, {_, _, _, _}} = datetime),
     do: {:ok, BSON.DateTime.from_datetime(datetime)}
   defp dump_datetime(_),
@@ -520,50 +554,31 @@ defmodule Mongo.Ecto do
     end
   end
 
-  @doc false
-  def update(_repo, meta, _fields, _filter, {key, :id, _}, _returning, _opts) do
-    raise ArgumentError,
-      "MongoDB adapter does not support :id field type in models. " <>
-      "The #{inspect key} field in #{inspect meta.model} is tagged as such."
-  end
-
-  def update(_repo, meta, _fields, _filter, _autogen, [_|_] = returning, _opts) do
-    raise ArgumentError,
-      "MongoDB adapter does not support :read_after_writes in models. " <>
-      "The following fields in #{inspect meta.model} are tagged as such: #{inspect returning}"
-  end
-
-  def update(repo, meta, fields, filter, {pk, :binary_id, _value}, [], opts) do
-    normalized = NormalizedQuery.update(meta, fields, filter, pk)
+  def insert_all(repo, meta, fields, params, returning, opts) do
+    normalized = NormalizedQuery.insert(meta, params)
 
-    Connection.update(repo, normalized, opts)
+    case Connection.insert_all(repo, normalized, opts) do
+      {:ok, _} ->
+        {:ok, []}
+      other ->
+        other
+    end
   end
 
-  def update(repo, meta, fields, filter, nil, [], opts) do
-    normalized = NormalizedQuery.update(meta, fields, filter, nil)
+  @doc false
+  def update(repo, meta, fields, filters, returning, opts) do
+    normalized = NormalizedQuery.update(meta, fields, filters)
 
     Connection.update(repo, normalized, opts)
   end
 
   @doc false
-  def delete(_repo, meta, _filter, {key, :id, _}, _opts) do
-    raise ArgumentError,
-      "MongoDB adapter does not support :id field type in models. " <>
-      "The #{inspect key} field in #{inspect meta.model} is tagged as such."
-  end
-
-  def delete(repo, meta, filter, {pk, :binary_id, _value}, opts) do
-    normalized = NormalizedQuery.delete(meta, filter, pk)
+  def delete(repo, meta, filter, opts) do
+    normalized = NormalizedQuery.delete(meta, filter)
 
     Connection.delete(repo, normalized, opts)
   end
 
-  def delete(repo, meta, fields, filter, nil, [], opts) do
-    normalized = NormalizedQuery.update(meta, fields, filter, nil)
-
-    Connection.update(repo, normalized, opts)
-  end
-
   defp process_document(document, %{fields: fields, pk: pk}, preprocess) do
     document = Conversions.to_ecto_pk(document, pk)
 

lib/mongo_ecto/connection.ex

@@ -6,18 +6,25 @@ defmodule Mongo.Ecto.Connection do
   alias Mongo.Ecto.NormalizedQuery.CommandQuery
   alias Mongo.Ecto.NormalizedQuery.CountQuery
   alias Mongo.Ecto.NormalizedQuery.AggregateQuery
+  alias Mongo.Query
 
   ## Worker
 
   def storage_down(opts) do
     opts = Keyword.put(opts, :pool, DBConnection.Connection)
 
+    {:ok, pid} = case Application.ensure_started(:mongodb)  do
+      :ok -> {:ok, nil}
+      {:error, reason} -> Mongo.App.start(nil, nil)
+      pid -> pid
+    end
     {:ok, conn} = Mongo.start_link(opts)
 
     try do
       Mongo.command!(conn, dropDatabase: 1)
       :ok
     after
+      if pid, do: Supervisor.stop(pid)
       GenServer.stop(conn)
     end
   end
@@ -27,7 +34,8 @@ defmodule Mongo.Ecto.Connection do
   def read(repo, query, opts \\ [])
 
   def read(repo, %ReadQuery{} = query, opts) do
-    opts  = [projection: query.projection, sort: query.order] ++ query.opts ++ opts
+    projection = Map.put_new(query.projection, :_id, false)
+    opts  = [projection: projection, sort: query.order] ++ query.opts ++ opts
     coll  = query.coll
     query = query.query
 
@@ -80,8 +88,12 @@ defmodule Mongo.Ecto.Connection do
     opts     = query.opts ++ opts
     query    = query.query
 
-    %{modified_count: n} = query(repo, :update, [coll, query, command], opts)
-    n
+    case query(repo, :update_many, [coll, query, command], opts) do
+      {:ok, %Mongo.UpdateResult{modified_count: m}} ->
+        m
+      {:error, error} ->
+        check_constraint_errors(error)
+    end
   end
 
   def update(repo, %WriteQuery{} = query, opts) do
@@ -90,7 +102,7 @@ defmodule Mongo.Ecto.Connection do
     opts     = query.opts ++ opts
     query    = query.query
 
-    case query(repo, :update, [coll, query, command], opts) do
+    case query(repo, :update_one, [coll, query, command], opts) do
       {:ok, %{modified_count: 1}} ->
         {:ok, []}
       {:ok, _} ->
@@ -111,6 +123,19 @@ defmodule Mongo.Ecto.Connection do
     end
   end
 
+  def insert_all(repo, %WriteQuery{} = query, opts) do
+    coll     = query.coll
+    command  = query.command
+    opts     = query.opts ++ opts
+
+    case query(repo, :insert_many, [coll, command], opts) do
+      {:ok, %{inserted_ids: ids}} ->
+        {Enum.count(ids), nil}
+      {:error, error} ->
+        check_constraint_errors(error)
+    end
+  end
+
   def command(repo, %CommandQuery{} = query, opts) do
     command  = query.command
     opts     = query.opts ++ opts
@@ -126,17 +151,21 @@ defmodule Mongo.Ecto.Connection do
 
   defp with_log(repo, opts) do
     case Keyword.pop(opts, :log, true) do
-      {true, opts}  -> [log: &log(repo, &1)] ++ opts
+      {true, opts}  -> [log: &log(repo, &1, opts)] ++ opts
       {false, opts} -> opts
     end
   end
 
-  defp log(repo, entry) do
+  defp log(repo, entry, opts) do
     %{connection_time: query_time, decode_time: decode_time,
-      pool_time: queue_time, result: result, query: query, params: params} = entry
+      pool_time: queue_time, result: result,
+      query: query, params: params} = entry
+    source = Keyword.get(opts, :source)
+
     repo.__log__(%Ecto.LogEntry{query_time: query_time, decode_time: decode_time,
                                 queue_time: queue_time, result: log_result(result),
-                                params: [], query: &format_query(&1, query, params)})
+                                params: [], query: format_query(query, params),
+                                source: source})
   end
 
   defp log_result({:ok, _query, res}), do: {:ok, res}
@@ -160,48 +189,66 @@ defmodule Mongo.Ecto.Connection do
     end
   end
 
-  alias Mongo.Query
-
-  # TODO: fix logging
-  defp format_query(_entry, %Query{action: :command}, [command]) do
+  defp format_query(%Query{action: :command}, [command]) do
     ["COMMAND " | inspect(command)]
   end
-  defp format_query(_entry, :insert_one, [coll, doc, _opts]) do
-    ["INSERT", format_part("coll", coll), format_part("document", doc)]
-  end
-  defp format_query(_entry, :insert_many, [coll, docs, _opts]) do
-    ["INSERT", format_part("coll", coll), format_part("documents", docs)]
+  defp format_query(%Query{action: :find, extra: coll}, [query, projection]) do
+    ["FIND",
+     format_part("coll", coll),
+     format_part("query", query),
+     format_part("projection", projection)]
   end
-  defp format_query(_entry, :delete_one, [coll, filter, _opts]) do
-    ["DELETE", format_part("coll", coll), format_part("filter", filter),
-     format_part("many", false)]
+  defp format_query(%Query{action: :insert_one, extra: coll}, [doc]) do
+    ["INSERT",
+     format_part("coll", coll),
+     format_part("document", doc)]
   end
-  defp format_query(_entry, :delete_many, [coll, filter, _opts]) do
-    ["DELETE", format_part("coll", coll), format_part("filter", filter),
+  defp format_query(%Query{action: :insert_many, extra: coll}, docs) do
+    ["INSERT",
+     format_part("coll", coll),
+     format_part("documents", docs),
      format_part("many", true)]
   end
-  defp format_query(_entry, :replace_one, [coll, filter, doc, _opts]) do
-    ["REPLACE", format_part("coll", coll), format_part("filter", filter),
-     format_part("document", doc)]
+  defp format_query(%Query{action: :update_one, extra: coll}, [filter, update]) do
+    ["UPDATE",
+     format_part("coll", coll),
+     format_part("filter", filter),
+     format_part("update", update)]
   end
-  defp format_query(_entry, :update_one, [coll, filter, update, _opts]) do
-    ["UPDATE", format_part("coll", coll), format_part("filter", filter),
-     format_part("update", update), format_part("many", false)]
+  defp format_query(%Query{action: :update_many, extra: coll}, [filter, update]) do
+    ["UPDATE",
+     format_part("coll", coll),
+     format_part("filter", filter),
+     format_part("update", update),
+     format_part("many", true)]
   end
-  defp format_query(_entry, :update_many, [coll, filter, update, _opts]) do
-    ["UPDATE", format_part("coll", coll), format_part("filter", filter),
-     format_part("update", update), format_part("many", true)]
+  defp format_query(%Query{action: :delete_one, extra: coll}, [filter]) do
+    ["DELETE",
+     format_part("coll", coll),
+     format_part("filter", filter)]
   end
-  defp format_query(_entry, :find, [coll, query, projection, _opts]) do
-    ["FIND", format_part("coll", coll), format_part("query", query),
-     format_part("projection", projection)]
+  defp format_query(%Query{action: :delete_many, extra: coll}, [filter]) do
+    ["DELETE",
+     format_part("coll", coll),
+     format_part("filter", filter),
+     format_part("many", true)]
   end
-  defp format_query(_entry, :find_rest, [coll, cursor, _opts]) do
+  defp format_query(%Query{action: :replace_one, extra: coll}, [filter, doc]) do
+    ["REPLACE", format_part("coll", coll), format_part("filter", filter),
+     format_part("document", doc)]
+  end
+  defp format_query(%Query{action: :get_more, extra: coll}, [cursor]) do
     ["GET_MORE", format_part("coll", coll), format_part("cursor_id", cursor)]
   end
-  defp format_query(_entry, :kill_cursors, [cursors, _opts]) do
+  defp format_query(%Query{action: :get_more, extra: coll}, []) do
+    ["GET_MORE", format_part("coll", coll), format_part("cursor_id", "")]
+  end
+  defp format_query(%Query{action: :kill_cursors, extra: coll}, [cursors]) do
     ["KILL_CURSORS", format_part("cursor_ids", cursors)]
   end
+  defp format_query(%Query{action: :kill_cursors, extra: coll}, []) do
+    ["KILL_CURSORS", format_part("cursor_ids", "")]
+  end
 
   defp format_part(name, value) do
     [" ", name, "=" | inspect(value)]