Skip to content

Commit

Permalink
Replace the Pgx_async example with a better one
Browse files Browse the repository at this point in the history 
This focuses on the parts of the interface that people should actually use,
and has a somewhat-reasonable organization for the code.

Also made some notes about how to apply that example to Pgx_lwt and Pgx_unix.
  • Loading branch information
@brendanlong
brendanlong committed Jan 22, 2019
1 parent 1fba23a commit 8f5ffa9
Show file tree
Hide file tree
Showing 2 changed files with 125 additions and 166 deletions.
19 changes: 18 additions & 1 deletion README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,4 +36,21 @@ opam pin add pgx https://github.com/arenadotio/pgx.git

## Examples

See [pgx_async/bin/pgx_async_example.ml](pgx_async/bin/pgx_async_example.ml).
See [pgx_async/bin/pgx_async_example.ml](pgx_async/bin/pgx_async_example.ml) for
a complete example of the high-level functional interface. To translate the
example to Lwt, replace `Pgx_async` with `Pgx_lwt` and `>>| with `>|=`. To
translate it to synchronous IO / standard-library-only, use `Pgx_unix` and
replace both `>>|` and `>>=` with `|>`, or just replace `>>| fun () ->` with `;`.

I.e. in `Pgx_unix`, you can replace:

```ocaml
Pgx_async.execute ~params "INSERT INTO ..."
>>| fun () ->
```

... with:

```ocaml
Pgx_unix.execute ~params "INSERT INTO ...";
```
272 changes: 107 additions & 165 deletions pgx_async/bin/pgx_async_example.ml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,174 +2,116 @@
open Core
open Async

let print_table dbh =
let query_all = "SELECT * FROM Company" in
Pgx_async.simple_query dbh query_all
>>| fun x -> printf !"%{sexp: Pgx.row list list}\n" x
module Employee = struct
let create db =
Pgx_async.simple_query db {|
CREATE TEMPORARY TABLE Employee (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL UNIQUE);
|}
|> Deferred.ignore

(* This function lets us insert multiple users relatively efficiently *)
let insert_many db names =
let params =
List.map names ~f:(fun name ->
Pgx_async.Value.[ of_string name ])
in
Pgx_async.execute_many db ~params ~query:{|
INSERT INTO Employee (name)
VALUES ($1)
RETURNING id
|}
>>| List.map ~f:(function
| [[ id ]] -> Pgx.Value.to_int_exn id
| _ -> assert false)

let insert ~name db =
insert_many db [ name ]
>>| List.hd_exn
end

module Facility = struct
let create db =
Pgx_async.simple_query db {|
CREATE TEMPORARY TABLE Facility (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL UNIQUE,
director_id INT REFERENCES Employee(id) ON DELETE SET NULL);

CREATE INDEX facility_director_id ON Facility (director_id);
|}
|> Deferred.ignore

let insert ~name ?director_id db =
let params = Pgx_async.Value.[ of_string name ; opt of_int director_id ] in
Pgx_async.execute db ~params {|
INSERT INTO Facility (name, director_id)
VALUES ($1, $2)
RETURNING id
|}
>>| function
| [[ id ]] -> Pgx.Value.to_int_exn id
| _ -> assert false

let all_name_and_director_name db =
Pgx_async.execute db {|
SELECT f.name, e.name
FROM Facility f
LEFT JOIN Employee e ON e.id = f.director_id
|}
>>| List.map ~f:(function
| [ name ; director_name ] ->
Pgx.Value.(to_string_exn name, to_string director_name)
| _ -> assert false)

let reassign_director db ~director_id ~from_facility_id ~to_facility_id =
(* Note: with_transaction doesn't currently have any special handling
for concurrent queries *)
Pgx_async.with_transaction db @@ fun db ->
let params = Pgx.Value.[ of_int director_id ; of_int from_facility_id ] in
Pgx_async.execute db ~params {|
UPDATE Facility SET director_id = NULL WHERE id = $2 AND director_id = $1
|}
>>= fun _ ->
let params = Pgx.Value.[ of_int director_id ; of_int to_facility_id ] in
Pgx_async.execute db ~params {|
UPDATE Facility SET director_id = $1 WHERE id = $2
|}
|> Deferred.ignore
end

let setup db =
Employee.create db
>>= fun () ->
Facility.create db

let main () =

Pgx_async_test.with_temp_db
(fun dbh ~db_name:_ ->
(** Create the table *)
print_endline "creating table";
Pgx_async.simple_query dbh
"CREATE TABLE Company( \
id INT PRIMARY KEY NOT NULL, \
name TEXT NOT NULL, \
age INT NOT NULL, \
address VARCHAR(50), \
salary REAL)"

(** Inserting one row with prepare and execute_prepared *)
>>= fun _ ->
print_endline "inserting 1 row";
let insert_query = "INSERT INTO Company (id, name, age, address, salary) \
VALUES ($1, $2, $3, $4, $5)" in
let params = [ Some "2" ; Some "John" ; Some "21" ; Some "Pittsburgh" ; Some "100000.0" ] in
Pgx_async.Prepared.(with_prepare dbh ~query:insert_query
~f:(execute ~params))

(** Inserting multiple rows with prepare_execute_prepared_many *)
>>= fun _ ->
print_endline "inserting 2 rows at the same time";
let params = [[ Some "4" ; Some "Robert" ; Some "21" ; Some "Pittsburgh" ; Some "100000.0"]
;[ Some "3" ; Some "John" ; Some "23" ; Some "New York" ; Some "30000.0" ]] in
Pgx_async.execute_many dbh ~query:insert_query ~params

(** Select a single row, with id = 3*)
>>= fun _ ->
print_endline "selecting name with id = 3";
let query_select_id = "SELECT name FROM Company WHERE id = $1" in
let params = [ Some "3" ] in
Pgx_async.execute dbh query_select_id ~params
>>= fun rows ->
printf !"%{sexp: Pgx.row list}\n" rows;

(** Alter table by deleting the column address *)
print_endline "altering table by dropping column address";
let query = "ALTER TABLE Company
DROP COLUMN address" in
Pgx_async.simple_query dbh query

>>= fun _ -> print_table dbh

(** adding a new column gender *)
>>= fun _ ->
print_endline "altering table by adding gender column";
let query = "ALTER TABLE Company
ADD COLUMN gender TEXT" in
Pgx_async.simple_query dbh query

>>= fun _ -> print_table dbh

>>= fun _ ->
print_endline "updating John's gender";
let query = "UPDATE Company
SET gender = $1
WHERE name = $2" in
let params = [ Some "male" ; Some "John" ] in
Pgx_async.execute dbh query ~params

>>= fun _ -> print_table dbh

(** Update a row *)
>>= fun _ ->
print_endline "updating person with id = 4";
let query = "UPDATE Company
SET name = $1, age = $2
WHERE id = $3" in
let params = [ Some "Roger" ; Some "34" ; Some "4" ] in
Pgx_async.execute dbh query ~params

>>= fun _ -> print_table dbh

(** Delete a row *)
>>= fun _ ->
print_endline "deleting person with id = 4";
let query = "DELETE FROM Company
WHERE id= $1" in
let params = [ Some "4" ] in
Pgx_async.execute dbh query ~params

>>= fun _ -> print_table dbh

(** Begin Transaction using Pgx_async.transact *)
>>= fun () ->
print_endline "start transaction";
Monitor.try_with (fun () ->
Pgx_async.with_transaction dbh
(fun dbh ->
(** Deleting id = 2 *)
print_endline "deleting person with id = 2";
let query = "DELETE FROM Company WHERE id = $1" in
let params = [ Some "2" ] in
Pgx_async.execute dbh query ~params

>>= fun _ -> print_table dbh

(** if an exception is thrown, rollback will be done *)
>>= fun () -> failwith "example exception"
)
>>| function
| Ok v -> v
| Error _e -> print_endline "caught exception"
)

>>= fun _ ->
print_endline "exception was thrown in transaction, so rollback executed";
print_table dbh

(** Begin Transaction using manual transaction *)
>>= fun () -> Pgx_async.begin_work dbh
>>= fun tx ->
print_endline "start transaction";

(** Add a guy with id = 5 *)
print_endline "inserting person with id = 5";
let query = "INSERT INTO Company (id, name, age, salary) VALUES ($1, $2, $3, $4)" in
let params = [ Some "5" ; Some "Alex" ; Some "15" ; Some "60000.0" ] in
Pgx_async.execute dbh query ~params

>>= fun _ ->
print_endline "commit to inserting person with id = 5";

Pgx_async.commit tx

(** rollback and commit both automatically exit the transaction *)
>>= fun _ -> print_table dbh

(** begin work again *)
>>= fun () -> Pgx_async.begin_work dbh
>>= fun _ ->
print_endline "start transaction";

(** update alex's salary by increasing it *)
print_endline "update alex's salary";
let query = "UPDATE Company
SET salary = $1
WHERE name = $2" in
let params = [ Some "70000.0" ; Some "Alex" ] in
Pgx_async.execute dbh query ~params

>>= fun _ -> print_table dbh

(** rollback to alex's original salary *)
>>= fun _ ->
print_endline "rollback to alex's original salary";
Pgx_async.rollback dbh

>>= fun _ -> print_table dbh

(** select multiple rows using execute_many *)
>>= fun _ ->
print_endline "selecting multiple lines at same time using execute_many. Make sure order is correct!";
let params = [[Some "John"; Some "50000"];[Some "Alex"; Some "0"]] in
Pgx_async.execute_many dbh ~query:"SELECT * FROM Company WHERE name = $1 and salary >= $2" ~params
>>| fun x -> printf !"%{sexp: Pgx.row list list}\n" x
)
Pgx_async.with_conn @@ fun db ->
setup db
>>= fun () ->
let%bind steve_id = Employee.insert ~name:"Steve" db in
(* Parallel queries are not an error, but will execute in serial *)
[ Facility.insert ~name:"Headquarters" ~director_id:steve_id db
; Facility.insert ~name:"New Office" db ]
|> Deferred.all
>>= function
| [ headquarters_id ; new_office_id ] ->
Facility.all_name_and_director_name db
>>| List.iter ~f:(fun (name, director_name) ->
let director_name = Option.value director_name ~default:"(none)" in
printf "The director of %s is %s\n" name director_name)
>>= fun () ->
print_endline "Re-assigning Steve to the New Office";
Facility.reassign_director db ~director_id:steve_id ~from_facility_id:headquarters_id ~to_facility_id:new_office_id
>>= fun () ->
Facility.all_name_and_director_name db
>>| List.iter ~f:(fun (name, director_name) ->
let director_name = Option.value director_name ~default:"(none)" in
printf "The director of %s is %s\n" name director_name)
| _ -> assert false

let () =
let summary = "example command" in
let summary = "Pgx_async example" in
Command.async ~summary (Command.Param.return main)
|> Command.run

0 comments on commit 8f5ffa9

Please sign in to comment.