use std::time::Duration;
use sqlx::Database as SqlxDatabase;
#[cfg(feature = "mysql")]
use sqlx::MySql;
#[cfg(feature = "mysql")]
use sqlx::MySqlPool;
#[cfg(feature = "postgres")]
use sqlx::PgPool;
#[cfg(feature = "postgres")]
use sqlx::Postgres;
#[cfg(feature = "sqlite")]
use sqlx::Sqlite;
#[cfg(feature = "sqlite")]
use sqlx::SqlitePool;
use crate::transaction::Target;
use crate::{Driver, Error, Transaction};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PoolOptions {
pub max_connections: u32,
pub acquire_timeout: Duration,
pub min_connections: u32,
}
impl Default for PoolOptions {
fn default() -> Self {
Self {
max_connections: 5,
acquire_timeout: Duration::from_secs(10),
min_connections: 0,
}
}
}
#[derive(Debug)]
#[doc(hidden)]
pub enum Pool {
#[cfg(feature = "mysql")]
MySql(MySqlPool),
#[cfg(feature = "postgres")]
Postgres(PgPool),
#[cfg(feature = "sqlite")]
Sqlite(SqlitePool),
}
impl Pool {
pub(crate) fn driver(&self) -> Driver {
match self {
#[cfg(feature = "mysql")]
Self::MySql(_) => Driver::MySql,
#[cfg(feature = "postgres")]
Self::Postgres(_) => Driver::Postgres,
#[cfg(feature = "sqlite")]
Self::Sqlite(_) => Driver::Sqlite,
}
}
}
#[derive(Debug)]
pub struct Database {
pub(crate) pool: Pool,
}
fn driver_for(url: &str) -> Result<Driver, Error> {
let scheme = url.split_once(':').map_or(url, |(scheme, _)| scheme);
#[cfg(feature = "mysql")]
if <MySql as SqlxDatabase>::URL_SCHEMES.contains(&scheme) {
return Ok(Driver::MySql);
}
#[cfg(feature = "postgres")]
if <Postgres as SqlxDatabase>::URL_SCHEMES.contains(&scheme) {
return Ok(Driver::Postgres);
}
#[cfg(feature = "sqlite")]
if <Sqlite as SqlxDatabase>::URL_SCHEMES.contains(&scheme) {
return Ok(Driver::Sqlite);
}
Err(Error::UnsupportedDriver {
scheme: scheme.to_owned(),
})
}
impl Database {
pub async fn connect(url: &str) -> Result<Self, Error> {
Self::connect_with(url, PoolOptions::default()).await
}
pub async fn connect_with(url: &str, options: PoolOptions) -> Result<Self, Error> {
let pool = match driver_for(url)? {
#[cfg(feature = "mysql")]
Driver::MySql => Pool::MySql(options.apply::<MySql>().connect(url).await?),
#[cfg(feature = "postgres")]
Driver::Postgres => Pool::Postgres(options.apply::<Postgres>().connect(url).await?),
#[cfg(feature = "sqlite")]
Driver::Sqlite => Pool::Sqlite(options.apply::<Sqlite>().connect(url).await?),
};
Ok(Self { pool })
}
pub fn driver(&self) -> Driver {
self.pool.driver()
}
#[cfg(feature = "mysql")]
pub fn pool(&self) -> &MySqlPool {
match &self.pool {
Pool::MySql(pool) => pool,
#[cfg(any(feature = "postgres", feature = "sqlite"))]
_ => panic!(
"`pool` is for MySQL; this database is {}",
self.driver().name()
),
}
}
#[cfg(feature = "postgres")]
pub fn postgres_pool(&self) -> Option<&PgPool> {
match &self.pool {
Pool::Postgres(pool) => Some(pool),
#[cfg(any(feature = "mysql", feature = "sqlite"))]
_ => None,
}
}
#[cfg(feature = "sqlite")]
pub fn sqlite_pool(&self) -> Option<&SqlitePool> {
match &self.pool {
Pool::Sqlite(pool) => Some(pool),
#[cfg(any(feature = "mysql", feature = "postgres"))]
_ => None,
}
}
pub async fn ping(&self) -> Result<(), Error> {
crate::backend::execute(Target::Pool(&self.pool), "SELECT 1", &[]).await?;
Ok(())
}
pub async fn transaction<T, E, F>(&self, f: F) -> Result<T, E>
where
E: From<Error>,
F: AsyncFnOnce(&mut Transaction) -> Result<T, E>,
{
let mut tx = Transaction::begin(&self.pool).await?;
match f(&mut tx).await {
Ok(value) => {
tx.commit().await?;
Ok(value)
}
Err(error) => {
let _ = tx.rollback().await;
Err(error)
}
}
}
pub async fn close(&self) {
match &self.pool {
#[cfg(feature = "mysql")]
Pool::MySql(pool) => pool.close().await,
#[cfg(feature = "postgres")]
Pool::Postgres(pool) => pool.close().await,
#[cfg(feature = "sqlite")]
Pool::Sqlite(pool) => pool.close().await,
}
}
}
impl PoolOptions {
fn apply<DB: SqlxDatabase>(self) -> sqlx::pool::PoolOptions<DB> {
sqlx::pool::PoolOptions::<DB>::new()
.max_connections(self.max_connections)
.min_connections(self.min_connections)
.acquire_timeout(self.acquire_timeout)
}
}