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
use std::time::Duration;
use crate::{
config::{Config, ConfigBuilder, Database, LiveConfig},
errors::Result,
pool::{create_pool, ConnectionPool},
query::Query,
stream::DetachedRowStream,
txn::Txn,
};
/// A neo4j database abstraction.
/// This type can be cloned and shared across threads, internal resources
/// are reference-counted.
#[derive(Clone)]
pub struct Graph {
config: LiveConfig,
pool: ConnectionPool,
}
/// Returns a [`Query`] which provides methods like [`Query::param`] to add parameters to the query
pub fn query(q: &str) -> Query {
Query::new(q.to_owned())
}
impl Graph {
/// Connects to the database with configurations provided.
///
/// You can build a config using [`ConfigBuilder::default()`].
pub async fn connect(config: Config) -> Result<Self> {
let pool = create_pool(&config).await?;
let config = config.into_live_config();
Ok(Graph { config, pool })
}
/// Connects to the database with default configurations
pub async fn new(
uri: impl Into<String>,
user: impl Into<String>,
password: impl Into<String>,
) -> Result<Self> {
let config = ConfigBuilder::default()
.uri(uri)
.user(user)
.password(password)
.build()?;
Self::connect(config).await
}
/// Starts a new transaction on the configured database.
/// All queries that needs to be run/executed within the transaction
/// should be executed using either [`Txn::run`] or [`Txn::execute`]
///
/// Transactions will not be automatically retried on any failure.
pub async fn start_txn(&self) -> Result<Txn> {
self.start_txn_on(self.config.db.clone()).await
}
/// Starts a new transaction on the provided database.
/// All queries that needs to be run/executed within the transaction
/// should be executed using either [`Txn::run`] or [`Txn::execute`]
///
/// Transactions will not be automatically retried on any failure.
pub async fn start_txn_on(&self, db: impl Into<Database>) -> Result<Txn> {
let connection = self.pool.get().await?;
Txn::new(db.into(), self.config.fetch_size, connection).await
}
/// Runs a query on the configured database using a connection from the connection pool,
/// It doesn't return any [`RowStream`] as the `run` abstraction discards any stream.
///
/// This operation retires the query on certain failures.
/// All errors with the `Transient` error class as well as a few other error classes are considered retryable.
/// This includes errors during a leader election or when the transaction resources on the server (memory, handles, ...) are exhausted.
/// Retries happen with an exponential backoff until a retry delay exceeds 60s, at which point the query fails with the last error as it would without any retry.
///
/// Use [`Graph::run`] for cases where you just want a write operation
///
/// use [`Graph::execute`] when you are interested in the result stream
pub async fn run(&self, q: Query) -> Result<()> {
self.run_on(&self.config.db, q).await
}
/// Runs a query on the provided database using a connection from the connection pool.
/// It doesn't return any [`RowStream`] as the `run` abstraction discards any stream.
///
/// This operation retires the query on certain failures.
/// All errors with the `Transient` error class as well as a few other error classes are considered retryable.
/// This includes errors during a leader election or when the transaction resources on the server (memory, handles, ...) are exhausted.
/// Retries happen with an exponential backoff until a retry delay exceeds 60s, at which point the query fails with the last error as it would without any retry.
///
/// Use [`Graph::run`] for cases where you just want a write operation
///
/// use [`Graph::execute`] when you are interested in the result stream
pub async fn run_on(&self, db: &str, q: Query) -> Result<()> {
backoff::future::retry_notify(
self.pool.manager().backoff(),
|| {
let pool = &self.pool;
let query = &q;
async move {
let mut connection = pool.get().await.map_err(crate::Error::from)?;
query.run_retryable(db, &mut connection).await
}
},
Self::log_retry,
)
.await
}
/// Executes a query on the configured database and returns a [`DetachedRowStream`]
///
/// This operation retires the query on certain failures.
/// All errors with the `Transient` error class as well as a few other error classes are considered retryable.
/// This includes errors during a leader election or when the transaction resources on the server (memory, handles, ...) are exhausted.
/// Retries happen with an exponential backoff until a retry delay exceeds 60s, at which point the query fails with the last error as it would without any retry.
pub async fn execute(&self, q: Query) -> Result<DetachedRowStream> {
self.execute_on(&self.config.db, q).await
}
/// Executes a query on the provided database and returns a [`DetaRowStream`]
///
/// This operation retires the query on certain failures.
/// All errors with the `Transient` error class as well as a few other error classes are considered retryable.
/// This includes errors during a leader election or when the transaction resources on the server (memory, handles, ...) are exhausted.
/// Retries happen with an exponential backoff until a retry delay exceeds 60s, at which point the query fails with the last error as it would without any retry.
pub async fn execute_on(&self, db: &str, q: Query) -> Result<DetachedRowStream> {
backoff::future::retry_notify(
self.pool.manager().backoff(),
|| {
let pool = &self.pool;
let fetch_size = self.config.fetch_size;
let query = &q;
async move {
let connection = pool.get().await.map_err(crate::Error::from)?;
query.execute_retryable(db, fetch_size, connection).await
}
},
Self::log_retry,
)
.await
}
fn log_retry(e: crate::Error, delay: Duration) {
let level = match delay.as_millis() {
0..=499 => log::Level::Debug,
500..=4999 => log::Level::Info,
_ => log::Level::Warn,
};
log::log!(level, "Retrying query in {delay:?} due to error: {e}");
}
}
const _: () = {
const fn assert_send_sync<T: ?Sized + Send + Sync>() {}
assert_send_sync::<Graph>();
};