-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, primary_key/2, is_auto_increment/3, field_type/3 ]). 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}), %% TODO ,remove the lastval() thing - that's a mysql thing 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) -> lists:filtermap(fun %% At some point (not sure when), the tuple returned was converted from %% a 6-tuple to an 8-tuple (probably an epgsql record). This %% accommodates both versions. ({column, Col, Type, _, _, _, _, _, _}) -> {true, {binary_to_atom(Col), Type}}; ({column, Col, Type, _, _, _}) -> {true, {binary_to_atom(Col), Type}}; (_) -> false end, 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([], []) -> []. 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). 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. primary_key(DB, Table) -> [T1, T2] = sql_bridge_utils:create_placeholders(2), SQL = [<<"SELECT column_name from information_schema.key_column_usage where table_catalog = ">>,T1, <<" and table_name = ">>,T2], case sql_bridge:ffl(SQL, [DB, Table]) of [F] -> sql_bridge_utils:to_atom(F); _ -> undefined end. is_auto_increment(DB, Table, Field) -> [T1, T2, T3] = sql_bridge_utils:create_placeholders(3), SQL = [<<"SELECT column_name, data_type, column_default, is_identity, identity_generation FROM information_schema.columns WHERE table_catalog = ">>,T1, <<" and table_name = ">>,T2, <<" and column_name = ">>,T3, <<" and (">>, %% PGsql < 10 <<"column_default LIKE 'nextval%' ">>, %%-- PGsql 10+ <<"OR is_identity = 'YES'">>, <<");">>], case sql_bridge:fffr(SQL, [DB, Table, Field]) of not_found -> false; _ -> true end. -spec field_type(sql_bridge:db(), sql_bridge:table(), sql_bridge:field()) -> sql_bridge:field_type(). field_type(DB, Table, Field) -> [T1, T2, T3] = sql_bridge_utils:create_placeholders(3), SQL = [ <<"select udt_name, character_maximum_length from information_schema.columns where table_catalog=">>,T1, <<" and table_name=">>,T2, <<" and column_name=">>,T3 ], Res = sql_bridge:tfr(SQL, [DB, Table, Field]), io:format("Postgres Field Type Result: (~p.~p.~p: ~p~n",[DB, Table, Field, Res]), case Res of {Type, Len} when Type=="varchar"; Type=="bpchar"; Type=="bytea"; Type=="text" -> {text, Len}; {Type, _} when Type=="int2"; Type=="int4"; Type=="int8" -> AType = list_to_atom(Type), {integer, sql_bridge_utils:int_ranges(AType)}; {Type, _} when Type=="uuid" -> {uuid, undefined} end.