-module(sql_bridge_epgsql). -behaviour(sql_bridge_adapter). -include("compat.hrl"). -export([start/0, connect/5, query/4, schema_db_column/0, encode/1, start_transaction/1, commit_transaction/1, rollback_transaction/1, with_transaction/2, wrap_field/1 ]). start() -> application:start(poolboy), ok. wrap_field(V) -> sql_bridge_utils:to_string(V). connect(DB, User, Pass, Host, Port) when is_atom(DB) -> WorkerArgs = [ {database, atom_to_list(DB)}, {hostname, Host}, {username, User}, {password, Pass}, {port, Port} ], sql_bridge_utils:start_poolboy_pool(DB, WorkerArgs, sql_bridge_epgsql_worker), ok. start_transaction(DB) -> sql_bridge_utils:checkout_pool(DB), query(none, DB, "BEGIN", []), ok. rollback_transaction(DB) -> query(none, DB, "ROLLBACK", []), sql_bridge_utils:checkin_pool(DB), ok. commit_transaction(DB) -> query(none, DB, "COMMIT", []), sql_bridge_utils:checkin_pool(DB), ok. with_transaction(DB, Fun) -> start_transaction(DB), try Fun() of Res -> commit_transaction(DB), Res catch Error:Class:ST -> rollback_transaction(DB), {error, [{Error, Class}, ST]} end. query(Type, DB, Q, ParamList) -> try query_catched(Type, DB, Q, ParamList) catch exit:{noproc, _} -> {error, no_pool} end. query_catched(Type, DB, Q, ParamList) -> {Q2, ParamList2} = maybe_replace_tokens(Q, ParamList), ToRun = fun(Worker) -> %% calls sql_bridge_epgsql_worker:handle_call() InnerRes = gen_server:call(Worker, {equery, Q2, ParamList2}), case {Type, InnerRes} of {insert, {ok, _Count}} -> InsertRes = gen_server:call(Worker, {equery, <<"select lastval();">>, []}), {ok, format_insert_id(InsertRes)}; _ -> InnerRes end end, Res = sql_bridge_utils:with_poolboy_pool(DB, ToRun), case Res of {error, Error} -> error_logger:error_msg("Query Error~nDB: ~p~nQuery: ~s~nParams: ~p~nError: ~p~n",[DB, Q, ParamList, Error]), erlang:exit(query_error); _ -> {ok, format_result(Type, Res)} end. format_insert_id({ok, Columns, Rows}) -> ColTypes = columns_to_coltypes(Columns), case format_lists(ColTypes, Rows) of [[Insertid]] -> Insertid; _ -> undefined end; format_insert_id({error, {error, error, _, <<"lastval is not defined",_/binary>>, _}}) -> undefined; format_insert_id({error, Error}) -> {error, Error}. maybe_replace_tokens(Q, ParamList) -> case sql_bridge_utils:replacement_token() of postgres -> {Q, ParamList}; mysql -> sql_bridge_utils:token_mysql_to_postgres(Q, ParamList) end. format_result(none, _) -> ok; format_result(_Type, {ok, Count}) -> Count; format_result(insert, {ok, _Count, _Columns, Rows}) -> element(1,hd(Rows)); format_result(Type, {ok, _Count, Columns, Rows}) -> format_result(Type, {ok, Columns, Rows}); format_result(Type, {ok, Columns, Rows}) -> ColTypes = columns_to_coltypes(Columns), format_result_inner(Type, {ok, ColTypes, Rows}). format_result_inner(tuple, {ok, Columns, Rows}) -> format_tuples(Columns, Rows); format_result_inner(list, {ok, Columns, Rows}) -> format_lists(Columns, Rows); format_result_inner(proplist, {ok, Columns, Rows}) -> format_proplists(Columns, Rows); format_result_inner(dict, {ok, Columns, Rows}) -> format_dicts(Columns, Rows); format_result_inner(map, {ok, Columns, Rows}) -> format_maps(Columns, Rows). columns_to_coltypes(Columns) -> [{list_to_atom(binary_to_list(Col)), Type} || {column, Col, Type, _, _, _} <- Columns]. normalize_value(numeric, B) when is_binary(B) -> normalize_value(numeric, binary_to_list(B)); normalize_value(numeric, V) when is_list(V) -> try list_to_float(V) catch _:_ -> try list_to_integer(V) catch _:_ -> V end end; normalize_value(_Type, V) when is_tuple(V) -> sql_bridge_utils:format_datetime(V); normalize_value(_Type, V) -> sql_bridge_stringify:maybe_string(V). format_tuples(Columns, Rows) -> [format_tuple(Columns, Row) || Row <- Rows]. format_tuple(Columns, Row) -> list_to_tuple(format_list(Columns, Row)). format_lists(Columns, Rows) -> [format_list(Columns, Row) || Row <- Rows]. format_list(Columns, Row) when is_tuple(Row) -> format_list(Columns, tuple_to_list(Row)); format_list(Columns, Row) when is_list(Row) -> make_list(Columns, Row). make_list(Cols, Vals) -> ColVals = lists:zip(Cols, Vals), [normalize_value(Type, Val) || {{_Col, Type}, Val} <- ColVals]. format_proplists(Columns, Rows) -> [make_proplist(Columns, Row) || Row <- Rows]. format_dicts(Columns, Rows) -> [make_dict(Columns, Row) || Row <- Rows]. make_dict(Cols, Row) when is_tuple(Row) -> make_dict(Cols, tuple_to_list(Row), dict:new()). make_dict([], [], Dict) -> Dict; make_dict([{Col, Type}|Cols], [Val|Vals], Dict) -> Val2 = normalize_value(Type, Val), NewDict = dict:store(Col, Val2, Dict), make_dict(Cols, Vals, NewDict). make_proplist(Columns, Row) when is_tuple(Row) -> make_proplist(Columns, tuple_to_list(Row)); make_proplist([{Col, Type}|Cols], [Val|Vals]) -> Val2 = normalize_value(Type, Val), [{Col, Val2} | make_proplist(Cols, Vals)]; make_proplist([], []) -> []. -ifdef(has_maps). format_maps(Columns, Rows) -> [make_map(Columns, Row) || Row <- Rows]. make_map(Cols, Row) -> make_map(Cols, tuple_to_list(Row), maps:new()). make_map([], [], Map) -> Map; make_map([{Col, Type}|Cols],[Val|Vals], Map) -> Val2 = normalize_value(Type, Val), NewMap = maps:put(Col, Val2, Map), make_map(Cols, Vals, NewMap). -else. format_maps(_,_) -> throw(maps_not_supported). -endif. schema_db_column() -> "table_catalog". encode(A) when is_atom(A) -> encode(atom_to_list(A)); encode(I) when is_integer(I) -> list_to_binary(integer_to_list(I)); encode(F) when is_float(F) -> list_to_binary(float_to_list(F)); encode(B) when is_binary(B) -> <<"'",(escape_binary(B, <<>>))/binary,"'">>; encode(L) when is_list(L) -> encode(list_to_binary(L)). escape_binary(<>, Acc) when X==$' -> escape_binary(Rest, <>); escape_binary(<>, Acc) -> escape_binary(Rest, <>); escape_binary(<<>>, Acc) -> Acc.