elephantry 5.1.1

Object model manager for PostgreSQL
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

/**
 * This `struct` is created by the [`Connection::async`] method.
 *
 * [`Connection::async`]: crate::Connection::async
 */
#[derive(Debug)]
pub struct Async<'c> {
    last_result: Option<crate::Result<crate::pq::Result>>,
    connection: &'c std::sync::Mutex<libpq::Connection>,
}

impl std::future::Future for Async<'_> {
    type Output = crate::Result<crate::pq::Result>;

    fn poll(
        mut self: std::pin::Pin<&mut Self>,
        ctx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Self::Output> {
        let connection = match self.connection.lock() {
            Ok(connection) => connection,
            Err(_) => return std::task::Poll::Pending,
        };

        if let Some(result) = connection.result() {
            self.last_result = Some(result.try_into());
        } else {
            let last_result = self.last_result.take();

            if let Some(result) = last_result {
                return std::task::Poll::Ready(result);
            }
        }

        ctx.waker().wake_by_ref();
        std::task::Poll::Pending
    }
}

impl<'c> Async<'c> {
    pub(crate) fn new(connection: &'c std::sync::Mutex<libpq::Connection>) -> Self {
        Self {
            last_result: None,
            connection,
        }
    }

    /**
     * Async version of [`Connection::execute`].
     *
     * [`Connection::execute`]: crate::Connection::execute
     */
    pub async fn execute(self, query: &str) -> crate::Result<crate::pq::Result> {
        self.connection
            .lock()
            .map_err(|e| crate::Error::Mutex(e.to_string()))?
            .send_query(query)
            .map_err(crate::Error::Async)?;

        self.await
    }

    /**
     * Async version of [`Connection::query`].
     *
     * [`Connection::query`]: crate::Connection::query
     */
    pub async fn query<E: crate::Entity>(
        self,
        query: &str,
        params: &[&dyn crate::ToSql],
    ) -> crate::Result<crate::Rows<E>> {
        Ok(self.send_query(query, params).await?.into())
    }

    /**
     * Async version of [`Connection::query_one`].
     *
     * [`Connection::query_one`]: crate::Connection::query_one
     */
    pub async fn query_one<E: crate::Entity>(
        self,
        query: &str,
        params: &[&dyn crate::ToSql],
    ) -> crate::Result<E> {
        match self.query(query, params).await?.try_get(0) {
            Some(e) => Ok(e),
            None => Err(crate::Error::MissingField("0".to_string())),
        }
    }

    async fn send_query(
        self,
        query: &str,
        params: &[&dyn crate::ToSql],
    ) -> crate::Result<crate::pq::Result> {
        let param = crate::Connection::transform_params(params)?;

        self.connection
            .lock()
            .map_err(|e| crate::Error::Mutex(e.to_string()))?
            .send_query_params(
                query,
                &param.types,
                &param
                    .values
                    .iter()
                    .map(|x| x.as_deref())
                    .collect::<Vec<_>>(),
                &[],
                crate::pq::Format::Binary,
            )
            .map_err(crate::Error::Async)?;

        self.await
    }
}