%% @author Maas-Maarten Zeeman <mmzeeman@xs4all.nl>
|
%% @copyright 2011 - 2017 Maas-Maarten Zeeman
|
|
%% @doc Erlang API for sqlite3 databases
|
|
%% Copyright 2011 - 2017 Maas-Maarten Zeeman
|
%%
|
%% Licensed under the Apache License, Version 2.0 (the "License");
|
%% you may not use this file except in compliance with the License.
|
%% You may obtain a copy of the License at
|
%%
|
%% http://www.apache.org/licenses/LICENSE-2.0
|
%%
|
%% Unless required by applicable law or agreed to in writing, software
|
%% distributed under the License is distributed on an "AS IS" BASIS,
|
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
%% See the License for the specific language governing permissions and
|
%% limitations under the License.
|
|
-module(esqlite3).
|
-author("Maas-Maarten Zeeman <mmzeeman@xs4all.nl>").
|
|
%% higher-level export
|
-compile(export_all).
|
|
-define(DEFAULT_TIMEOUT, 5000).
|
-define(DEFAULT_CHUNK_SIZE, 5000).
|
|
%%
|
-type connection() :: {connection, reference(), term()}.
|
-type statement() :: {statement, term(), connection()}.
|
-type sql() :: iodata().
|
|
%% @doc Opens a sqlite3 database mentioned in Filename.
|
%%
|
-spec open(FileName) -> {ok, connection()} | {error, _} when
|
FileName :: string().
|
open(Filename) ->
|
open(Filename, ?DEFAULT_TIMEOUT).
|
|
%% @doc Open a database connection
|
%%
|
-spec open(Filename, timeout()) -> {ok, connection()} | {error, _} when
|
Filename :: string().
|
open(Filename, Timeout) ->
|
{ok, Connection} = esqlite3_nif:start(),
|
|
Ref = make_ref(),
|
ok = esqlite3_nif:open(Connection, Ref, self(), Filename),
|
case receive_answer(Ref, Timeout) of
|
ok ->
|
{ok, {connection, make_ref(), Connection}};
|
{error, _Msg}=Error ->
|
Error
|
end.
|
|
%% @doc Subscribe to database notifications
|
%% Messages will come in the shape {action, table, id}
|
%% Where action will be insert | update | delete
|
%% and table will be a string
|
%% and id will be an integer
|
%%
|
-spec set_update_hook(pid(), connection()) -> ok | {error, term()}.
|
set_update_hook(Pid, Connection) ->
|
set_update_hook(Pid, Connection, ?DEFAULT_TIMEOUT).
|
|
-spec set_update_hook(pid(), connection(), timeout()) -> ok | {error, term()}.
|
set_update_hook(Pid, {connection, _Ref, Connection}, Timeout) ->
|
Ref = make_ref(),
|
ok = esqlite3_nif:set_update_hook(Connection, Ref, self(), Pid),
|
receive_answer(Ref, Timeout).
|
|
%% @doc Execute a sql statement, returns a list with tuples.
|
-spec q(sql(), connection()) -> list(tuple()) | {error, term()}.
|
q(Sql, Connection) ->
|
q(Sql, [], Connection).
|
|
%% @doc Execute statement, bind args and return a list with tuples as result.
|
-spec q(sql(), list(), connection()) -> list(tuple()) | {error, term()}.
|
q(Sql, [], Connection) ->
|
case prepare(Sql, Connection) of
|
{ok, Statement} ->
|
fetchall(Statement);
|
{error, _Msg}=Error ->
|
throw(Error)
|
end;
|
q(Sql, Args, Connection) ->
|
case prepare(Sql, Connection) of
|
{ok, Statement} ->
|
ok = bind(Statement, Args),
|
fetchall(Statement);
|
{error, _Msg}=Error ->
|
throw(Error)
|
end.
|
|
%% @doc
|
-spec map(F, sql(), connection()) -> list(Type) when
|
F :: fun((Row) -> Type) | fun((ColumnNames, Row) -> Type),
|
Row :: tuple(),
|
ColumnNames :: tuple(),
|
Type :: any().
|
map(F, Sql, Connection) ->
|
case prepare(Sql, Connection) of
|
{ok, Statement} ->
|
map_s(F, Statement);
|
{error, _Msg}=Error ->
|
throw(Error)
|
end.
|
|
%% @doc
|
-spec foreach(F, sql(), connection()) -> ok when
|
F :: fun((Row) -> any()) | fun((ColumnNames, Row) -> any()),
|
Row :: tuple(),
|
ColumnNames :: tuple().
|
foreach(F, Sql, Connection) ->
|
case prepare(Sql, Connection) of
|
{ok, Statement} ->
|
foreach_s(F, Statement);
|
{error, _Msg}=Error ->
|
throw(Error)
|
end.
|
|
%%
|
-spec foreach_s(F, statement()) -> ok when
|
F :: fun((Row) -> any()) | fun((ColumnNames, Row) -> any()),
|
Row :: tuple(),
|
ColumnNames :: tuple().
|
foreach_s(F, Statement) when is_function(F, 1) ->
|
case try_multi_step(Statement, 1, [], 0) of
|
{'$done', []} -> ok;
|
{error, _} = E -> F(E);
|
{rows, [Row | []]} ->
|
F(Row),
|
foreach_s(F, Statement)
|
end;
|
foreach_s(F, Statement) when is_function(F, 2) ->
|
ColumnNames = column_names(Statement),
|
case try_multi_step(Statement, 1, [], 0) of
|
{'$done', []} -> ok;
|
{error, _} = E -> F([], E);
|
{rows, [Row | []]} ->
|
F(ColumnNames, Row),
|
foreach_s(F, Statement)
|
end.
|
|
%%
|
-spec map_s(F, statement()) -> list(Type) when
|
F :: fun((Row) -> Type) | fun((ColumnNames, Row) -> Type),
|
Row :: tuple(),
|
ColumnNames :: tuple(),
|
Type :: term().
|
map_s(F, Statement) when is_function(F, 1) ->
|
case try_multi_step(Statement, 1, [], 0) of
|
{'$done', []} -> [];
|
{error, _} = E -> F(E);
|
{rows, [Row | []]} ->
|
[F(Row) | map_s(F, Statement)]
|
end;
|
map_s(F, Statement) when is_function(F, 2) ->
|
ColumnNames = column_names(Statement),
|
case try_multi_step(Statement, 1, [], 0) of
|
{'$done', []} -> [];
|
{error, _} = E -> F([], E);
|
{rows, [Row | []]} ->
|
[F(ColumnNames, Row) | map_s(F, Statement)]
|
end.
|
|
%%
|
%%-spec fetchone(statement()) -> tuple().
|
fetchone(Statement) ->
|
case try_multi_step(Statement, 1, [], 0) of
|
{'$done', []} -> ok;
|
{error, _} = E -> E;
|
{rows, [Row | []]} -> Row
|
end.
|
|
%% @doc Fetch all records
|
%% @param Statement is prepared sql statement
|
%% @spec fetchall(statement()) -> list(tuple()) | {error, term()}.
|
-spec fetchall(statement()) ->
|
list(tuple()) |
|
{error, term()}.
|
fetchall(Statement) ->
|
fetchall(Statement, ?DEFAULT_CHUNK_SIZE).
|
|
%% @doc Fetch all records
|
%% @param Statement is prepared sql statement
|
%% @param ChunkSize is a count of rows to read from sqlite and send to erlang process in one bulk.
|
%% Decrease this value if rows are heavy. Default value is 5000 (DEFAULT_CHUNK_SIZE).
|
%% @spec fetchall(statement()) -> list(tuple()) | {error, term()}.
|
-spec fetchall(statement(), pos_integer()) ->
|
list(tuple()) |
|
{error, term()}.
|
fetchall(Statement, ChunkSize) ->
|
case fetchall_internal(Statement, ChunkSize, []) of
|
{'$done', Rows} -> lists:reverse(Rows);
|
{error, _} = E -> E
|
end.
|
|
%% return rows in revers order
|
-spec fetchall_internal(statement(), pos_integer(), list(tuple())) ->
|
{'$done', list(tuple())} |
|
{error, term()}.
|
fetchall_internal(Statement, ChunkSize, Rest) ->
|
case try_multi_step(Statement, ChunkSize, Rest, 0) of
|
{rows, Rows} -> fetchall_internal(Statement, ChunkSize, Rows);
|
Else -> Else
|
end.
|
|
%% Try a number of steps, when the database is busy,
|
%% return rows in revers order
|
-spec try_multi_step(statement(), pos_integer(), list(tuple()), non_neg_integer()) ->
|
{rows, list(tuple())} |
|
{'$done', list(tuple())} |
|
{error, term()}.
|
try_multi_step(_Statement, _ChunkSize, _Rest, Tries) when Tries > 5 ->
|
throw(too_many_tries);
|
try_multi_step(Statement, ChunkSize, Rest, Tries) ->
|
case multi_step(Statement, ChunkSize) of
|
{'$busy', Rows} -> %% core can fetch a number of rows (rows < ChunkSize) per 'multi_step' call and then get busy...
|
erlang:display({"busy", Tries}),
|
timer:sleep(100 * Tries),
|
try_multi_step(Statement, ChunkSize, Rows ++ Rest, Tries + 1);
|
{rows, Rows} ->
|
{rows, Rows ++ Rest};
|
{'$done', Rows} ->
|
{'$done', Rows ++ Rest};
|
Else -> Else
|
end.
|
|
%% @doc Execute Sql statement, returns the number of affected rows.
|
%%
|
%% @spec exec(iolist(), connection()) -> integer() | {error, error_message()}
|
exec(Sql, Connection) ->
|
exec("begin", Connection, ?DEFAULT_TIMEOUT),
|
exec(Sql, Connection, ?DEFAULT_TIMEOUT),
|
exec("commit", Connection, ?DEFAULT_TIMEOUT).
|
|
%% @doc Execute
|
%%
|
%% @spec exec(iolist(), connection(), timeout()) -> integer() | {error, error_message()}
|
exec(Sql, {connection, _Ref, Connection}, Timeout) ->
|
Ref = make_ref(),
|
ok = esqlite3_nif:exec(Connection, Ref, self(), Sql),
|
receive_answer(Ref, Timeout);
|
|
%% @spec exec(iolist(), list(term()), connection()) -> integer() | {error, error_message()}
|
exec(Sql, Params, {connection, _, _}=Connection) when is_list(Params) ->
|
exec(Sql, Params, Connection, ?DEFAULT_TIMEOUT).
|
|
%% @spec exec(iolist(), list(term()), connection(), timeout()) -> integer() | {error, error_message()}
|
exec(Sql, Params, {connection, _, _}=Connection, Timeout) when is_list(Params) ->
|
{ok, Statement} = prepare(Sql, Connection, Timeout),
|
bind(Statement, Params),
|
step(Statement, Timeout).
|
|
|
%% @doc Return the number of affected rows of last statement.
|
changes(Connection) ->
|
changes(Connection, ?DEFAULT_TIMEOUT).
|
|
%% @doc Return the number of affected rows of last statement.
|
changes({connection, _Ref, Connection}, Timeout) ->
|
Ref = make_ref(),
|
ok = esqlite3_nif:changes(Connection, Ref, self()),
|
receive_answer(Ref, Timeout).
|
|
%% @doc Insert records, returns the last rowid.
|
%%
|
%% @spec insert(iolist(), connection()) -> {ok, integer()} | {error, error_message()}
|
insert(Sql, Connection) ->
|
insert(Sql, Connection, ?DEFAULT_TIMEOUT).
|
|
%% @doc Insert
|
%%
|
%% @spec insert(iolist(), connection(), timeout()) -> {ok, integer()} | {error, error_message()}
|
insert(Sql, {connection, _Ref, Connection}, Timeout) ->
|
Ref = make_ref(),
|
ok = esqlite3_nif:insert(Connection, Ref, self(), Sql),
|
receive_answer(Ref, Timeout).
|
|
%% @doc Get autocommit
|
%%
|
%% @spec get_autocommit(connection) -> true | false
|
get_autocommit(Connection) ->
|
get_autocommit(Connection, ?DEFAULT_TIMEOUT).
|
|
get_autocommit({connection, _Ref, Connection}, Timeout) ->
|
Ref = make_ref(),
|
ok = esqlite3_nif:get_autocommit(Connection, Ref, self()),
|
receive_answer(Ref, Timeout).
|
|
%% @doc Prepare a statement
|
%%
|
%% @spec prepare(iolist(), connection()) -> {ok, prepared_statement()} | {error, error_message()}
|
prepare(Sql, Connection) ->
|
prepare(Sql, Connection, ?DEFAULT_TIMEOUT).
|
|
%% @doc
|
%%
|
%% @spec(iolist(), connection(), timeout()) -> {ok, prepared_statement()} | {error, error_message()}
|
prepare(Sql, {connection, _Ref, Connection}=C, Timeout) ->
|
Ref = make_ref(),
|
ok = esqlite3_nif:prepare(Connection, Ref, self(), Sql),
|
case receive_answer(Ref, Timeout) of
|
{ok, Stmt} -> {ok, {statement, Stmt, C}};
|
Else -> Else
|
end.
|
|
%% @doc Step
|
%%
|
%% @spec step(prepared_statement()) -> tuple()
|
step(Stmt) ->
|
step(Stmt, ?DEFAULT_TIMEOUT).
|
|
%% @doc
|
%%
|
%% @spec step(prepared_statement(), timeout()) -> tuple()
|
-spec step(term(), timeout()) -> tuple() | '$busy' | '$done'.
|
step({statement, Stmt, {connection, _, Conn}}, Timeout) ->
|
Ref = make_ref(),
|
ok = esqlite3_nif:multi_step(Conn, Stmt, 1, Ref, self()),
|
case receive_answer(Ref, Timeout) of
|
{rows, [Row | []]} -> {row, Row};
|
{'$done', []} -> '$done';
|
{'$busy', []} -> '$busy';
|
Else -> Else
|
end.
|
|
%% make multiple sqlite steps per call
|
%% return rows in reverse order
|
multi_step(Stmt, ChunkSize) ->
|
multi_step(Stmt, ChunkSize, ?DEFAULT_TIMEOUT).
|
|
%% make multiple sqlite steps per call
|
%% return rows in reverse order
|
-spec multi_step(term(), pos_integer(), timeout()) ->
|
{rows, list(tuple())} |
|
{'$busy', list(tuple())} |
|
{'$done', list(tuple())} |
|
{error, term()}.
|
multi_step({statement, Stmt, {connection, _, Conn}}, ChunkSize, Timeout) ->
|
Ref = make_ref(),
|
ok = esqlite3_nif:multi_step(Conn, Stmt, ChunkSize, Ref, self()),
|
receive_answer(Ref, Timeout).
|
|
%% @doc Reset the prepared statement back to its initial state.
|
%%
|
%% @spec reset(prepared_statement()) -> ok | {error, error_message()}
|
reset({statement, Stmt, {connection, _, Conn}}) ->
|
Ref = make_ref(),
|
ok = esqlite3_nif:reset(Conn, Stmt, Ref, self()),
|
receive_answer(Ref, ?DEFAULT_TIMEOUT).
|
|
%% @doc Bind values to prepared statements
|
%%
|
%% @spec bind(prepared_statement(), value_list()) -> ok | {error, error_message()}
|
bind(Stmt, Args) ->
|
bind(Stmt, Args, ?DEFAULT_TIMEOUT).
|
|
%% @doc Bind values to prepared statements
|
%%
|
%% @spec bind(prepared_statement(), [], timeout()) -> ok | {error, error_message()}
|
bind({statement, Stmt, {connection, _, Conn}}, Args, Timeout) ->
|
Ref = make_ref(),
|
ok = esqlite3_nif:bind(Conn, Stmt, Ref, self(), Args),
|
receive_answer(Ref, Timeout).
|
|
%% @doc Return the column names of the prepared statement.
|
%%
|
-spec column_names(statement()) -> {atom()}.
|
column_names(Stmt) ->
|
column_names(Stmt, ?DEFAULT_TIMEOUT).
|
|
-spec column_names(statement(), timeout()) -> {atom()}.
|
column_names({statement, Stmt, {connection, _, Conn}}, Timeout) ->
|
Ref = make_ref(),
|
ok = esqlite3_nif:column_names(Conn, Stmt, Ref, self()),
|
receive_answer(Ref, Timeout).
|
|
%% @doc Return the column types of the prepared statement.
|
%%
|
-spec column_types(statement()) -> {atom()}.
|
column_types(Stmt) ->
|
column_types(Stmt, ?DEFAULT_TIMEOUT).
|
|
-spec column_types(statement(), timeout()) -> {atom()}.
|
column_types({statement, Stmt, {connection, _, Conn}}, Timeout) ->
|
Ref = make_ref(),
|
ok = esqlite3_nif:column_types(Conn, Stmt, Ref, self()),
|
receive_answer(Ref, Timeout).
|
|
%% @doc Close the database
|
%%
|
%% @spec close(connection()) -> ok | {error, error_message()}
|
-spec close(connection()) -> ok | {error, _}.
|
close(Connection) ->
|
close(Connection, ?DEFAULT_TIMEOUT).
|
|
%% @doc Close the database
|
%%
|
%% @spec close(connection(), integer()) -> ok | {error, error_message()}
|
-spec close(connection(), timeout()) -> ok | {error, _}.
|
close({connection, _Ref, Connection}, Timeout) ->
|
Ref = make_ref(),
|
ok = esqlite3_nif:close(Connection, Ref, self()),
|
receive_answer(Ref, Timeout).
|
|
%% Internal functions
|
|
receive_answer(Ref, Timeout) ->
|
Start = os:timestamp(),
|
receive
|
{esqlite3, Ref, Resp} ->
|
Resp;
|
{esqlite3, _, _}=StaleAnswer ->
|
error_logger:warning_msg("Esqlite3: Ignoring stale answer ~p~n", [StaleAnswer]),
|
PassedMics = timer:now_diff(os:timestamp(), Start) div 1000,
|
NewTimeout = case Timeout - PassedMics of
|
Passed when Passed < 0 -> 0;
|
TO -> TO
|
end,
|
receive_answer(Ref, NewTimeout)
|
after Timeout ->
|
throw({error, timeout, Ref})
|
end.
|
