Bolty (Bolty v0.3.0)

Copy Markdown View Source

Bolt driver for Elixir — talk to Neo4j (and other Bolt servers) over the binary Bolt protocol.

Bolty is built on DBConnection, so a connection is a supervised, pooled process: start one with start_link/1 (or via child_spec/1 in a supervision tree) and query it with query/4 / query_many/4.

Example

{:ok, conn} =
  Bolty.start_link(
    hostname: "localhost",
    auth: [username: "neo4j", password: "password"]
  )

{:ok, %Bolty.Response{} = res} =
  Bolty.query(conn, "MATCH (n:Person) RETURN n.name AS name LIMIT $limit", %{limit: 5})

# A Response is Enumerable over its result rows
names = Enum.map(res, & &1["name"])

See start_link/1 for the full connection options (TLS, pooling, auth). Every failure — connection, TLS, version negotiation, and Neo4j server errors — is surfaced as a Bolty.Error, so callers can match a single error shape.

Guides

Summary

Types

basic_auth()

The basic authentication scheme relies on traditional username and password

conn()
option()
start_option()

Functions

child_spec(options)

Returns a supervisor child specification for a DBConnection pool.

connection_info(conn)

Returns metadata about the negotiated connection.

query(conn, statement, params \\ %{}, opts \\ [])

Executes a single query and returns the result.

query!(conn, statement, params \\ %{}, opts \\ [])

Executes a single query and returns the result.

query_many(conn, statement, params \\ %{}, opts \\ [])

Executes a batch of ;-separated statements and returns a list of results.

query_many!(conn, statement, params \\ %{}, opts \\ [])

Same as query_many/4, but returns the list of Bolty.Response structs directly and raises the Bolty.Error from the first failing statement.

rollback(conn, reason)

See DBConnection.rollback/2.

start_link(options)

Starts the connection process and connects to a Bolt/Neo4j server.

stream(conn, statement, params \\ %{}, opts \\ [])

Lazily streams a query's result in batches, with server-side backpressure.

transaction(conn, fun, opts \\ [], extra_parameters \\ %{})

Runs fun inside a Bolt transaction, committing on return and rolling back on rollback/2 or a raised error.

Types

basic_auth()

@type basic_auth() :: {:username, String.t()} | {:password, String.t() | nil}

The basic authentication scheme relies on traditional username and password

  • :username - Username (required)

  • :password - Password (default: nil)

conn()

@type conn() :: DBConnection.conn()

option()

@type option() :: %{
  bookmarks: list(),
  mode: String.t(),
  db: String.t() | nil,
  tx_metadata: map() | nil
}

start_option()

@type start_option() ::
  {:uri, String.t()}
  | {:hostname, String.t()}
  | {:port, :inet.port_number()}
  | {:scheme, String.t()}
  | {:routing, boolean()}
  | {:versions, [String.t() | float()]}
  | {:auth, basic_auth()}
  | {:user_agent, String.t()}
  | {:notifications_minimum_severity, String.t()}
  | {:notifications_disabled_categories, [String.t()]}
  | {:ssl_opts, [:ssl.tls_client_option()]}
  | {:connect_timeout, timeout()}
  | {:recv_timeout, timeout()}
  | {:socket_options, [:gen_tcp.connect_option()]}
  | DBConnection.start_option()

Functions

child_spec(options)

@spec child_spec([start_option()]) :: :supervisor.child_spec()

Returns a supervisor child specification for a DBConnection pool.

connection_info(conn)

@spec connection_info(conn()) :: %{
  bolt_version: String.t(),
  server_version: String.t(),
  policy: Bolty.Policy.t()
}

Returns metadata about the negotiated connection.

Can be called with the conn passed into a transaction/3 callback or any checked-out connection.

Example

Bolty.transaction(Bolt, fn conn ->
  Bolty.connection_info(conn)
  # => %{bolt_version: "5.8", server_version: "Neo4j/5.26.27", policy: %Bolty.Policy{...}}
end)

query(conn, statement, params \\ %{}, opts \\ [])

@spec query(conn(), String.t(), map(), keyword()) ::
  {:ok, Bolty.Response.t()}
  | {:error, Bolty.Error.t() | DBConnection.ConnectionError.t()}

Executes a single query and returns the result.

Returns {:ok, %Bolty.Response{}} on success. On failure it returns {:error, %Bolty.Error{}} — every driver-side failure (connection, TLS, version negotiation, and Neo4j server errors) is surfaced as a Bolty.Error, so callers can match a single error shape. Pool checkout/timeout failures may additionally surface as DBConnection.ConnectionError.

Options

  • :db - Target database for multi-database routing (default: server default).

  • :mode - Access mode for this query, "w" (write, default) or "r" (read). Under server-side routing (a neo4j:// connection to a cluster — see start_link/1's :routing and the Clustering guide) the server uses this to forward the query to a writable or a read member. Without routing it is informational only.

  • :bookmarks - A list of bookmark strings for causal consistency: the server will not run this query until it has caught up to the transactions those bookmarks identify. Each Bolty.Response carries the bookmark of the transaction that produced it; thread the previous response's bookmark into the next query to chain reads-after-writes. There is no session abstraction, so bookmarks must be passed manually.

  • :recv_timeout - Overrides the connection's :recv_timeout for this query (see start_link/1). Pass :infinity for a query expected to run longer than the connection default before returning data.

Examples

{:ok, result} = Bolty.query(conn, "MATCH (n) RETURN n LIMIT 1")

{:ok, people} = Bolty.query(conn, "MATCH (n:PERSON) RETURN n", %{}, [db: "mydb"])

query!(conn, statement, params \\ %{}, opts \\ [])

@spec query!(conn(), String.t(), map(), keyword()) :: Bolty.Response.t()

Executes a single query and returns the result.

Examples

result = Bolty.query!(conn, "MATCH (n) RETURN n LIMIT 1")

people = Bolty.query!(conn, "MATCH (n:PERSON) RETURN n", %{}, [db: "mydb"])

query_many(conn, statement, params \\ %{}, opts \\ [])

@spec query_many(conn(), String.t(), map(), keyword()) ::
  {:ok, [Bolty.Response.t()]}
  | {:error, Bolty.Error.t() | DBConnection.ConnectionError.t()}

Executes a batch of ;-separated statements and returns a list of results.

The batch string is split into individual statements before each is run in turn against the same connection. Execution stops at the first failure and returns that {:error, %Bolty.Error{}}.

Statements are split on a top-level ; — one that is not inside a string literal, a comment (// or /* */), or a backtick-quoted identifier — so a ; hiding inside any of those is safe. The terminating ; is dropped (Bolt runs one bare statement at a time), and blank or comment-only segments are skipped rather than sent to the server.

query_many!(conn, statement, params \\ %{}, opts \\ [])

@spec query_many!(conn(), String.t(), map(), keyword()) :: [Bolty.Response.t()]

Same as query_many/4, but returns the list of Bolty.Response structs directly and raises the Bolty.Error from the first failing statement.

rollback(conn, reason)

@spec rollback(DBConnection.t(), any()) :: no_return()

See DBConnection.rollback/2.

start_link(options)

@spec start_link([start_option()]) :: GenServer.on_start()

Starts the connection process and connects to a Bolt/Neo4j server.

Options

  • :uri - Connection URI, of the form <SCHEME>://<HOST>[:<PORT>[?policy=<POLICY-NAME>]]. Explicit :hostname, :port and :scheme options take priority over the corresponding URI components.

  • :hostname - Server hostname (default: "localhost")

  • :port - Server port (default: 7687)

  • :scheme - Is one among neo4j, neo4j+s, neo4j+ssc, bolt, bolt+s, bolt+ssc (default: bolt+s). TLS is derived entirely from this: bolt/neo4j is plaintext, +s is full certificate verification against the OS trust store, +ssc is encrypted but trust-all (self-signed certs). The scheme also sets the default for :routing (below).

  • :routing - Whether to enable server-side routing (SSR) against a Neo4j Enterprise causal cluster, so the server forwards each statement to the right member based on the query's :mode/:bookmarks (see query/4). Defaults to true for neo4j* schemes and false for bolt* schemes, matching their meaning across the ecosystem; set it explicitly to override (routing: false for a neo4j:// URI used cosmetically against a single instance or proxy, routing: true to force it on a bolt:// scheme). See the Clustering guide for prerequisites and caveats.

  • :versions - List of Bolt versions to offer during negotiation, as strings (["5.4", "6.0"]) or {major, minor..minor} range tuples ([{5, 6..8}]). Floats ([5.4]) are deprecated — they can't distinguish 5.10 from 5.1 — and are still accepted with a one-time warning. (Default: the latest supported versions.)

  • :auth - The basic authentication scheme

  • :user_agent - Optionally override the default user agent name. (Default: 'bolty/<version>')

  • :notifications_minimum_severity - Set the minimum severity for notifications the server should send to the client. Disabling severities allows the server to skip analysis for those, which can speed up query execution. (default: nil) New in neo4j v5.7 and Bolt v5.2

  • :notifications_disabled_categories - Set categories of notifications the server should not send to the client. Disabling categories allows the server to skip analysis for those, which can speed up query execution. (default: nil) New in neo4j v5.7 and Bolt v5.2

  • :connect_timeout - Socket connect timeout in milliseconds (default: 15_000)

  • :recv_timeout - Per-read timeout in milliseconds for post-connect protocol reads (default: 15_000). This is a socket-inactivity timeout: it fires only if the server sends no bytes for this long, and the window resets on each chunk received, so streaming a large result is unaffected. A read that times out is surfaced as {:error, %Bolty.Error{code: :timeout}} and the connection is disconnected. Can be overridden per query via the same option to query/4; set to :infinity for queries that may compute for a long time before returning any data.

  • :ssl_opts - A list of :ssl.connect/2 options, merged over the strict, secure-by-default TLS options bolty derives from :scheme (default: []). An explicit verify:/cacertfile:/cacerts: here always overrides the scheme-derived default. Only applies when :scheme enables TLS (+s or +ssc); ignored for plain bolt/neo4j.

The given options are passed down to DBConnection, some of the most commonly used ones are documented below:

  • :after_connect - A function to run after the connection has been established, either a 1-arity fun, a {module, function, args} tuple, or nil (default: nil)

  • :pool - The pool module to use, defaults to built-in pool provided by DBconnection

  • :pool_size - The size of the pool

stream(conn, statement, params \\ %{}, opts \\ [])

@spec stream(DBConnection.t(), String.t(), map(), keyword()) ::
  DBConnection.Stream.t()

Lazily streams a query's result in batches, with server-side backpressure.

Returns a DBConnection stream that pulls records from the server in batches as it is enumerated, rather than materialising the whole result up front like query/4. Each element is a %Bolty.Response{} for one batch — its results and records are that batch's records; the summary metadata (stats, bookmark, …) lands on the final batch. Enumerating only part of the stream fetches only the batches consumed; the rest is discarded server-side.

Must be called inside a transaction/4 — streaming pages against the statement's server-assigned query id, which only exists within an explicit transaction. Enumerate the stream within the same transaction callback.

Unlike query/4, a failure mid-stream is surfaced by raising a Bolty.Error while the stream is being enumerated (the Enumerable protocol has no error-tuple channel), rather than returning {:error, _}. The error value is the same %Bolty.Error{}; the connection is RESET-recovered and returns clean to the pool, and the enclosing transaction rolls back.

Options

  • :fetch_size - Records to pull per batch (the Bolt PULL n). Default 1000. Smaller batches mean more round-trips but lower peak memory.

  • :mode, :bookmarks, :db - As documented on query/4, applied to the streamed statement's RUN.

Example

Bolty.transaction(conn, fn conn ->
  conn
  |> Bolty.stream("MATCH (n:Person) RETURN n", %{}, fetch_size: 500)
  |> Stream.flat_map(& &1.results)
  |> Enum.each(&process/1)
end)

transaction(conn, fun, opts \\ [], extra_parameters \\ %{})

@spec transaction(
  conn(),
  (DBConnection.t() -> result),
  [DBConnection.option()],
  map()
) ::
  {:ok, result} | {:error, any()}
when result: var

Runs fun inside a Bolt transaction, committing on return and rolling back on rollback/2 or a raised error.

opts are passed to DBConnection.transaction/3. extra_parameters is a map applied to the opening BEGIN for the whole transaction:

  • :mode - "w" (write, default) or "r" (read). Under server-side routing (a neo4j:// connection to a cluster — see start_link/1's :routing and the Clustering guide) the server routes the transaction to a writable or a read member accordingly.

  • :bookmarks - A list of bookmark strings for causal consistency; the transaction will not begin until the server has caught up to the transactions they identify. Read a completed transaction's bookmark from the last Bolty.Response it produced and thread it into the next one — there is no session abstraction, so this is manual.

  • :db - Target database for the transaction (default: server default).

Example

Bolty.transaction(conn, fn conn ->
  Bolty.query!(conn, "CREATE (:Person {name: $n})", %{n: "Ada"})
end, [], %{mode: "w"})