graphile_worker 0.13.3

High performance Rust/PostgreSQL job queue (also suitable for getting jobs generated by PostgreSQL triggers/functions out into a different work queue)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

use graphile_worker::{DbExecutorArg, DbParams, DbValue};

use super::support::{connect_client, deadpool_pool};
use crate::helpers::with_test_db;

#[tokio::test]
async fn tokio_postgres_executor_args_support_direct_queries() {
    with_test_db(|test_db| async move {
        let mut client = connect_client(&test_db.name).await;
        let mut client_executor = &client;

        DbExecutorArg::execute(
            &mut client_executor,
            "SELECT $1::int",
            vec![DbValue::I32(1)].into(),
        )
        .await
        .expect("tokio-postgres client execute should succeed");

        let client_rows = DbExecutorArg::fetch_all(
            &mut client_executor,
            "SELECT $1::int AS value",
            vec![DbValue::I32(2)].into(),
        )
        .await
        .expect("tokio-postgres client fetch_all should succeed");
        assert_eq!(client_rows[0].try_get::<i32>("value").unwrap(), 2);

        let empty_error = DbExecutorArg::fetch_one(
            &mut client_executor,
            "SELECT $1::int AS value WHERE false",
            vec![DbValue::I32(3)].into(),
        )
        .await
        .expect_err("tokio-postgres client fetch_one should reject empty results");
        assert!(empty_error.to_string().contains("query returned no rows"));

        let tx = client
            .transaction()
            .await
            .expect("Failed to begin tokio-postgres transaction");
        let mut tx_executor = &tx;

        DbExecutorArg::execute(
            &mut tx_executor,
            "SELECT $1::int",
            vec![DbValue::I32(4)].into(),
        )
        .await
        .expect("tokio-postgres transaction execute should succeed");

        let tx_rows = DbExecutorArg::fetch_all(
            &mut tx_executor,
            "SELECT $1::int AS value",
            vec![DbValue::I32(5)].into(),
        )
        .await
        .expect("tokio-postgres transaction fetch_all should succeed");
        assert_eq!(tx_rows[0].try_get::<i32>("value").unwrap(), 5);

        tx.rollback()
            .await
            .expect("Failed to roll back tokio-postgres transaction");

        let mut tx = client
            .transaction()
            .await
            .expect("Failed to begin mutable tokio-postgres transaction");
        let mut tx_executor = &mut tx;

        DbExecutorArg::execute(
            &mut tx_executor,
            "SELECT $1::int",
            vec![DbValue::I32(6)].into(),
        )
        .await
        .expect("mutable tokio-postgres transaction execute should succeed");

        let tx_rows = DbExecutorArg::fetch_all(
            &mut tx_executor,
            "SELECT $1::int AS value",
            vec![DbValue::I32(7)].into(),
        )
        .await
        .expect("mutable tokio-postgres transaction fetch_all should succeed");
        assert_eq!(tx_rows[0].try_get::<i32>("value").unwrap(), 7);

        tx.rollback()
            .await
            .expect("Failed to roll back mutable tokio-postgres transaction");

        let pool = deadpool_pool(&test_db.name);
        let mut pool_executor = &pool;

        DbExecutorArg::execute(
            &mut pool_executor,
            "SELECT $1::int",
            vec![DbValue::I32(8)].into(),
        )
        .await
        .expect("deadpool pool execute should succeed");

        let pool_rows = DbExecutorArg::fetch_all(
            &mut pool_executor,
            "SELECT $1::int AS value",
            vec![DbValue::I32(9)].into(),
        )
        .await
        .expect("deadpool pool fetch_all should succeed");
        assert_eq!(pool_rows[0].try_get::<i32>("value").unwrap(), 9);

        let deadpool_client = pool.get().await.expect("Failed to get deadpool client");
        let mut deadpool_client_executor = &deadpool_client;

        DbExecutorArg::execute(
            &mut deadpool_client_executor,
            "SELECT $1::int",
            vec![DbValue::I32(10)].into(),
        )
        .await
        .expect("deadpool client execute should succeed");

        let deadpool_client_rows = DbExecutorArg::fetch_all(
            &mut deadpool_client_executor,
            "SELECT $1::int AS value",
            vec![DbValue::I32(11)].into(),
        )
        .await
        .expect("deadpool client fetch_all should succeed");
        assert_eq!(deadpool_client_rows[0].try_get::<i32>("value").unwrap(), 11);

        let params = DbParams::new();
        DbExecutorArg::execute(&mut deadpool_client_executor, "SELECT 1", params)
            .await
            .expect("deadpool client execute without params should succeed");
    })
    .await;
}