better_duck_core/asynchronous/
database.rs

1use std::path::Path;
2
3use crate::{
4    asynchronous::connection::AsyncConnection,
5    database::Database,
6    error::{Error, Result},
7};
8
9/// An async facade over a [`Database`].
10///
11/// `open`/`open_in_memory` and `connect` dispatch to `tokio::task::spawn_blocking`.
12#[derive(Clone, Debug)]
13pub struct AsyncDatabase {
14    inner: Database,
15}
16
17impl AsyncDatabase {
18    /// Wraps an existing [`Database`] for async use.
19    pub fn from_database(db: Database) -> AsyncDatabase {
20        AsyncDatabase { inner: db }
21    }
22
23    /// Returns the underlying [`Database`] handle.
24    pub fn database(&self) -> &Database {
25        &self.inner
26    }
27
28    /// Opens a database at the given file path.
29    ///
30    /// # Errors
31    ///
32    /// Returns an error if the database cannot be opened.
33    pub async fn open<P>(path: P) -> Result<AsyncDatabase>
34    where
35        P: AsRef<Path> + Send + 'static,
36    {
37        let db = tokio::task::spawn_blocking(move || Database::open(path))
38            .await
39            .map_err(|e| Error::BackgroundTaskFailed(e.to_string()))??;
40        Ok(AsyncDatabase::from_database(db))
41    }
42
43    /// Opens an in-memory database.
44    ///
45    /// # Errors
46    ///
47    /// Returns an error if the database cannot be opened.
48    pub async fn open_in_memory() -> Result<AsyncDatabase> {
49        let db = tokio::task::spawn_blocking(Database::open_in_memory)
50            .await
51            .map_err(|e| Error::BackgroundTaskFailed(e.to_string()))??;
52        Ok(AsyncDatabase::from_database(db))
53    }
54
55    /// Opens a new connection to this database.
56    ///
57    /// # Errors
58    ///
59    /// Returns an error if the connection cannot be established.
60    pub async fn connect(&self) -> Result<AsyncConnection> {
61        let db = self.inner.clone();
62        let conn = tokio::task::spawn_blocking(move || db.connect())
63            .await
64            .map_err(|e| Error::BackgroundTaskFailed(e.to_string()))??;
65        Ok(AsyncConnection::new(conn))
66    }
67}
68
69#[cfg(test)]
70mod tests {
71    use super::*;
72
73    #[tokio::test]
74    async fn async_database_parallel_connections() {
75        let db = AsyncDatabase::open_in_memory().await.unwrap();
76        let a = db.connect().await.unwrap();
77        a.execute_batch("CREATE TABLE t (id INTEGER)").await.unwrap();
78
79        let mut handles = Vec::new();
80        for i in 0..8 {
81            let db = db.clone();
82            handles.push(tokio::spawn(async move {
83                let conn = db.connect().await.unwrap();
84                conn.execute_batch(format!("INSERT INTO t VALUES ({i})")).await.unwrap();
85            }));
86        }
87        for h in handles {
88            h.await.unwrap();
89        }
90
91        let result = a.execute("SELECT count(*) AS c FROM t").await.unwrap();
92        match result.rows()[0].get("c").unwrap() {
93            crate::types::value::DuckValue::BigInt(n) => assert_eq!(*n, 8),
94            other => panic!("expected BigInt, got {other:?}"),
95        }
96    }
97}
better_duck_core/asynchronous/database.rs

better_duck_core/asynchronous/
database.rs
