Erlang PostgreSQL Database Client

Asynchronous fork of wg/epgsql originally here: mabrek/epgsql and subsequently forked in order to provide a common fork for community development.

pgapp

If you want to get up to speed quickly with code that lets you run Postgres queries, you might consider trying epgsql/pgapp, which adds the following, on top of the epgsql driver:

A 'resource pool' (currently poolboy), which lets you decide how many Postgres workers you want to utilize.
Resilience against the database going down or other problems. The pgapp code will keep trying to reconnect to the database, but will not propagate the crash up the supervisor tree, so that, for instance, your web site will stay up even if the database is down for some reason. Erlang's "let it crash" is a good idea, but external resources going away might not be a good reason to crash your entire system.

Motivation

When you need to execute several queries, it involves a number network round-trips between the application and the database. The PostgreSQL frontend/backend protocol supports request pipelining. This means that you don't need to wait for the previous command to finish before sending the next command. This version of the driver makes full use of the protocol feature that allows faster execution.

Difference highlights

3 API sets:
- epgsql maintains backwards compatibility with the original driver API
- epgsqla delivers complete results as regular erlang messages
- epgsqli delivers results as messages incrementally (row by row) All API interfaces can be used with the same connection: eg, connection opened with epgsql can be queried with epgsql / epgsqla / epgsqli in any combinations.
internal queue of client requests, so you don't need to wait for the response to send the next request (pipelining)
single process to hold driver state and receive socket data
execution of several parsed statements as a batch
binding timestamps in erlang:now() format

see CHANGES for full list.

Differences between current epgsql and mabrek's original async fork:

Unnamed statements are used unless specified otherwise. This may cause problems for people attempting to use the same connection concurrently, which will no longer work.

Known problems

SSL performance can degrade if the driver process has a large inbox (thousands of messages).

Usage

Connect

connect(Opts) -> {ok, Connection :: epgsql:connection()} | {error, Reason :: epgsql:connect_error()}
    when
  Opts ::
    #{host :=     inet:ip_address() | inet:hostname(),
      username := iodata(),
      password => iodata() | fun( () -> iodata() ),
      database => iodata(),
      port =>     inet:port_number(),
      ssl =>      boolean() | required,
      ssl_opts => [ssl:tls_client_option()], % @see OTP ssl documentation
      socket_active => true | integer(), % @see "Active socket" section below
      tcp_opts => [gen_tcp:option()],    % @see OTP gen_tcp module documentation
      timeout =>  timeout(),             % socket connect timeout, default: 5000 ms
      async =>    pid() | atom(),        % process to receive LISTEN/NOTIFY msgs
      codecs =>   [{epgsql_codec:codec_mod(), any()}]}
      nulls =>    [any(), ...],          % NULL terms
      replication => Replication :: string()} % Pass "database" to connect in replication mode
    | list().

connect(Host, Username, Password, Opts) -> {ok, C} | {error, Reason}.

example:

{ok, C} = epgsql:connect(#{
    host => "localhost",
    username => "username",
    password => "psss",
    database => "test_db",
    timeout => 4000
}),
...
ok = epgsql:close(C).

Only host and username are mandatory, but most likely you would need database and password.

password - DB user password. It might be provided as string / binary or as a fun that returns string / binary. Internally, plain password is wrapped to anonymous fun before it is sent to connection process, so, if connect command crashes, plain password will not appear in crash logs.
timeout parameter will trigger an {error, timeout} result when the socket fails to connect within provided milliseconds.
ssl if set to true, perform an attempt to connect in ssl mode, but continue unencrypted if encryption isn't supported by server. if set to required connection will fail if encryption is not available.
ssl_opts will be passed as is to ssl:connect/3.
tcp_opts will be passed as is to gen_tcp:connect/3. Some options are forbidden, such as mode, packet, header, active. When tcp_opts is not provided, epgsql does some tuning (eg, sets TCP keepalive and auto-tunes buffer), but when tcp_opts is provided, no additional tweaks are added by epgsql itself, other than necessary ones (active, packet and mode).
async see Server notifications
codecs see Pluggable datatype codecs
nulls terms which will be used to represent SQL NULL. If any of those has been encountered in placeholder parameters ($1, $2 etc values), it will be interpreted as NULL. 1st element of the list will be used to represent NULLs received from the server. It's not recommended to use "string"s or lists. Try to keep this list short for performance! Default is [null, undefined], i.e. encode null or undefined in parameters as NULL and decode NULLs as atom null.
replication see Streaming replication protocol
application_name is an optional string parameter. It is usually set by an application upon connection to the server. The name will be displayed in the pg_stat_activity view and included in CSV log entries.
socket_active is an optional parameter, which can be true or an integer in the range -32768 to 32767 (inclusive, however only positive value make sense right now). This option is used to control the flow of incoming messages from the network socket to make sure huge query results won't result in epgsql process mailbox overflow. It affects the behaviour of some of the commands and interfaces (epgsqli and replication), so, use with caution! See Active socket for more details.

Options may be passed as proplist or as map with the same key names.

Asynchronous connect example (applies to epgsqli too):

  {ok, C} = epgsqla:start_link(),
  Ref = epgsqla:connect(C, "localhost", "username", "psss", #{database => "test_db"}),
  receive
    {C, Ref, connected} ->
        {ok, C};
    {C, Ref, Error = {error, _}} ->
        Error;
    {'EXIT', C, _Reason} ->
        {error, closed}
  end.

Simple Query

-include_lib("epgsql/include/epgsql.hrl").

-type query() :: string() | iodata().
-type squery_row() :: tuple() % tuple of binary().

-type ok_reply(RowType) ::
    {ok, ColumnsDescription :: [epgsql:column()], RowsValues :: [RowType]} |                            % select
    {ok, Count :: non_neg_integer()} |                                                            % update/insert/delete
    {ok, Count :: non_neg_integer(), ColumnsDescription :: [epgsql:column()], RowsValues :: [RowType]}. % update/insert/delete + returning
-type error_reply() :: {error, query_error()}.
-type reply(RowType) :: ok_reply() | error_reply().

-spec squery(connection(), query()) -> reply(squery_row()) | [reply(squery_row())].
%% @doc runs simple `SqlQuery' via given `Connection'
squery(Connection, SqlQuery) -> ...

examples:

epgsql:squery(C, "insert into account (name) values  ('alice'), ('bob')").
> {ok,2}

epgsql:squery(C, "select * from account").
> {ok,
    [#column{name = <<"id">>, type = int4, …},#column{name = <<"name">>, type = text, …}],
    [{<<"1">>,<<"alice">>},{<<"2">>,<<"bob">>}]
}

epgsql:squery(C,
    "insert into account(name)"
    "    values ('joe'), (null)"
    "    returning *").
> {ok,2,
    [#column{name = <<"id">>, type = int4, …}, #column{name = <<"name">>, type = text, …}],
    [{<<"3">>,<<"joe">>},{<<"4">>,null}]
}

epgsql:squery(C, "SELECT * FROM _nowhere_").
> {error,
   #error{severity = error,code = <<"42P01">>,
          codename = undefined_table,
          message = <<"relation \"_nowhere_\" does not exist">>,
          extra = [{file,<<"parse_relation.c">>},
                   {line,<<"1160">>},
                   {position,<<"15">>},
                   {routine,<<"parserOpenTable">>}]}}

The simple query protocol returns all columns as binary strings and does not support parameters binding.

Several queries separated by semicolon can be executed by squery.

[{ok, _, [{<<"1">>}]}, {ok, _, [{<<"2">>}]}] = epgsql:squery(C, "select 1; select 2").

epgsqla:squery/2 returns result as a single message:

  Ref = epgsqla:squery(C, Sql),
  receive
    {C, Ref, Result} -> Result
  end.

Result has the same format as return value of epgsql:squery/2.

epgsqli:squery/2 returns results incrementally for each query inside Sql and for each row:

Ref = epgsqli:squery(C, Sql),
receive
  {C, Ref, {columns, Columns}} ->
      %% columns description
      Columns;
  {C, Ref, {data, Row}} ->
      %% single data row
      Row;
  {C, Ref, {error, _E} = Error} ->
      Error;
  {C, Ref, {complete, {_Type, Count}}} ->
      %% execution of one insert/upd

Epgsql

Install / Use

README