bsql 0.7.0

Safe SQL for Rust — if it compiles, the SQL is correct
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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162

//! Integration tests: fetch_stream — streaming query results.
//!
//! Requires a running PostgreSQL with the test schema.
//! Set BSQL_DATABASE_URL=postgres://bsql:bsql@localhost/bsql_test

use bsql::Pool;
use tokio_stream::StreamExt;

async fn pool() -> Pool {
    Pool::connect("postgres://bsql:bsql@localhost/bsql_test")
        .await
        .expect("Failed to connect to test database. Is PostgreSQL running?")
}

#[tokio::test]
async fn stream_multiple_rows() {
    let pool = pool().await;
    let mut stream = bsql::query!("SELECT id, login FROM users WHERE active = true ORDER BY id")
        .fetch_stream(&pool)
        .await
        .unwrap();

    let mut rows = Vec::new();
    while let Some(row) = stream.next().await {
        rows.push(row.unwrap());
    }

    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0].login, "alice");
    assert_eq!(rows[1].login, "bob");
}

#[tokio::test]
async fn stream_single_row() {
    let pool = pool().await;
    let id = 1i32;
    let mut stream = bsql::query!("SELECT id, login FROM users WHERE id = $id: i32")
        .fetch_stream(&pool)
        .await
        .unwrap();

    let row = stream.next().await.unwrap().unwrap();
    assert_eq!(row.id, 1);
    assert_eq!(row.login, "alice");

    // No more rows
    assert!(stream.next().await.is_none());
}

#[tokio::test]
async fn stream_zero_rows() {
    let pool = pool().await;
    let login = "nonexistent_user_for_stream_test";
    let mut stream = bsql::query!("SELECT id, login FROM users WHERE login = $login: &str")
        .fetch_stream(&pool)
        .await
        .unwrap();

    assert!(stream.next().await.is_none());
}

#[tokio::test]
async fn stream_with_nullable_column() {
    let pool = pool().await;
    let mut stream = bsql::query!("SELECT id, middle_name FROM users ORDER BY id")
        .fetch_stream(&pool)
        .await
        .unwrap();

    let row = stream.next().await.unwrap().unwrap();
    assert_eq!(row.id, 1);
    assert!(row.middle_name.is_none());
}

#[tokio::test]
async fn stream_connection_returns_to_pool_on_drop() {
    let pool = pool().await;
    let before = pool.status();

    {
        let mut stream = bsql::query!("SELECT id, login FROM users ORDER BY id")
            .fetch_stream(&pool)
            .await
            .unwrap();

        // Consume one row then drop the stream
        let _row = stream.next().await.unwrap().unwrap();
        // stream drops here
    }

    // Connection should be returned to the pool
    let after = pool.status();
    assert!(
        after.available >= before.available,
        "connection not returned: before={}, after={}",
        before.available,
        after.available,
    );
}

#[tokio::test]
async fn stream_collect_all() {
    let pool = pool().await;
    let stream = bsql::query!("SELECT id, login FROM users WHERE active = true ORDER BY id")
        .fetch_stream(&pool)
        .await
        .unwrap();

    let rows: Vec<_> = stream
        .collect::<Vec<_>>()
        .await
        .into_iter()
        .collect::<Result<Vec<_>, _>>()
        .unwrap();

    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0].login, "alice");
    assert_eq!(rows[1].login, "bob");
}

#[tokio::test]
async fn stream_with_params_and_join() {
    let pool = pool().await;
    let id = 1i32;
    let mut stream = bsql::query!(
        "SELECT t.id as ticket_id, t.title, u.login as creator
         FROM tickets t
         JOIN users u ON u.id = t.created_by_user_id
         WHERE t.id = $id: i32"
    )
    .fetch_stream(&pool)
    .await
    .unwrap();

    let row = stream.next().await.unwrap().unwrap();
    assert_eq!(row.ticket_id, 1);
    assert_eq!(row.title, "Fix login bug");
    assert_eq!(row.creator, "alice");
    assert!(stream.next().await.is_none());
}

#[tokio::test]
async fn stream_with_optional_clause() {
    let pool = pool().await;
    let dept: Option<i32> = None;
    let mut stream = bsql::query!(
        "SELECT id, title FROM tickets
         WHERE deleted_at IS NULL
         [AND department_id = $dept: Option<i32>]
         ORDER BY id"
    )
    .fetch_stream(&pool)
    .await
    .unwrap();

    let mut count = 0;
    while let Some(row) = stream.next().await {
        let _ = row.unwrap();
        count += 1;
    }
    assert!(count >= 2);
}