helge 3.0.0

Tiny wrapper around r2d2::Pool and diesel ConnectionManager to ease use in async contexts
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

//! Helge is a tiny wrapper around r2d2::Pool and diesel ConnectionManager
//! to provide a simple way to use diesel postgres with r2d2 in an async context.
//!
//! <br>
//!
//! ## Example

//! ```rust,no_run
//! use diesel::prelude::*;
//! use helge::{Error, Helge};
//!
//! diesel::table! {
//!     users {
//! 	    id -> Integer,
//! 	    name -> Varchar,
//!     }
//! }
//!
//! #[derive(Debug, Insertable)]
//! #[diesel(table_name = users)]
//! struct NewUser {
//!     name: String,
//! }
//!
//!
//! async fn query() -> Result<(), helge::Error> {
//!   let helge = Helge::<diesel::PgConnection>::new("postgres://localhost/somedatabase")
//!                            .expect("connecting Helge");
//!   helge.query(|conn| {
//!       diesel::insert_into(users::table)
//!           .values(&NewUser {
//!               name: String::from("Helge"),
//!           })
//!           .execute(conn)
//!   }).await;
//!   Ok(())
//! }
//!
//! ```

use diesel::r2d2::{self, ConnectionManager, ManageConnection, R2D2Connection};

mod error;
pub use error::{ConnectionError, Error};

/// The main wrapper, simply contains an r2d2::Pool
pub struct Helge<C>
where
    C: R2D2Connection + Send + 'static,
{
    pool: r2d2::Pool<ConnectionManager<C>>,
}

impl<C> Clone for Helge<C>
where
    C: R2D2Connection + Send + 'static,
{
    fn clone(&self) -> Self {
        Self {
            pool: self.pool.clone(),
        }
    }
}

impl<C> Helge<C>
where
    C: R2D2Connection + Send + 'static,
{
    /// Create a new Helge with default settings
    pub fn new(database_uri: impl Into<String>) -> Result<Self, ConnectionError> {
        let manager = ConnectionManager::<C>::new(database_uri);

        let _ = manager.connect().map_err(|err| match err {
            diesel::r2d2::Error::ConnectionError(err) => ConnectionError::Connection(err),
            diesel::r2d2::Error::QueryError(err) => ConnectionError::PingFailed(err),
        })?;

        let pool = diesel::r2d2::Builder::new()
            .max_size(5)
            .build(manager)
            .map_err(|err| ConnectionError::PoolSettings(err.to_string()))?;

        Ok(Self { pool })
    }

    pub fn from_pool(pool: r2d2::Pool<diesel::r2d2::ConnectionManager<C>>) -> Self {
        Self { pool }
    }

    pub fn get_conn(
        &self,
    ) -> Result<r2d2::PooledConnection<diesel::r2d2::ConnectionManager<C>>, Error> {
        self.pool.get().map_err(|err| Error::Pool(err.to_string()))
    }

    pub async fn query<T, F>(&self, f: F) -> std::result::Result<T, Error>
    where
        T: Send + 'static,
        F: FnOnce(&mut C) -> std::result::Result<T, diesel::result::Error> + Send + 'static,
    {
        let pool = self.pool.clone();

        tokio::task::spawn_blocking(move || {
            let mut conn = pool.get().map_err(|err| Error::Pool(err.to_string()))?;

            f(&mut conn).map_err(Error::Query)
        })
        .await
        .map_err(Error::Runtime)?
    }

    pub async fn run<T, E, F>(&self, f: F) -> std::result::Result<T, E>
    where
        T: Send + 'static,
        E: From<Error> + Send + 'static,
        F: FnOnce(&mut C) -> std::result::Result<T, E> + Send + 'static,
    {
        let pool = self.pool.clone();

        tokio::task::spawn_blocking(move || {
            let mut conn = pool
                .get()
                .map_err(|err| E::from(Error::Pool(err.to_string())))?;

            f(&mut conn)
        })
        .await
        .map_err(|err: tokio::task::JoinError| E::from(Error::Runtime(err)))?
    }
}