use std::fmt::{self, Debug, Formatter};
use std::future::{self, Future};
use std::str::FromStr;
use std::time::Duration;
use log::LevelFilter;
use sqlx_core::connection::{ConnectOptions, Connection};
use sqlx_core::error::Error;
use sqlx_core::transaction::Transaction;
use sqlx_core::Url;
use crate::D1;
pub struct D1Connection {
pub(crate) db: worker::D1Database,
}
impl D1Connection {
#[must_use]
pub fn new(db: worker::D1Database) -> Self {
Self { db }
}
pub fn from_env(env: &worker::Env, binding: &str) -> worker::Result<Self> {
env.d1(binding).map(Self::new)
}
#[must_use]
pub fn database(&self) -> &worker::D1Database {
&self.db
}
#[must_use]
pub fn into_database(self) -> worker::D1Database {
self.db
}
}
impl Debug for D1Connection {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
f.debug_struct("D1Connection").finish_non_exhaustive()
}
}
impl Connection for D1Connection {
type Database = D1;
type Options = D1ConnectOptions;
fn close(self) -> impl Future<Output = Result<(), Error>> + Send + 'static {
future::ready(Ok(()))
}
fn close_hard(self) -> impl Future<Output = Result<(), Error>> + Send + 'static {
future::ready(Ok(()))
}
fn ping(&mut self) -> impl Future<Output = Result<(), Error>> + Send + '_ {
future::ready(Ok(()))
}
fn begin(&mut self) -> impl Future<Output = Result<Transaction<'_, D1>, Error>> + Send + '_ {
Transaction::begin(self, None)
}
fn shrink_buffers(&mut self) {}
fn flush(&mut self) -> impl Future<Output = Result<(), Error>> + Send + '_ {
future::ready(Ok(()))
}
fn should_flush(&self) -> bool {
false
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct D1ConnectOptions {
binding: String,
}
impl D1ConnectOptions {
#[must_use]
pub fn binding(&self) -> &str {
&self.binding
}
}
impl FromStr for D1ConnectOptions {
type Err = Error;
fn from_str(s: &str) -> Result<Self, Error> {
let url = Url::parse(s).map_err(|error| Error::Configuration(error.into()))?;
Self::from_url(&url)
}
}
impl ConnectOptions for D1ConnectOptions {
type Connection = D1Connection;
fn from_url(url: &Url) -> Result<Self, Error> {
if url.scheme() != "d1" {
return Err(Error::Configuration(
format!(
"expected a `d1://<binding>` URL, got scheme `{}`",
url.scheme()
)
.into(),
));
}
match url.host_str() {
Some(binding) if !binding.is_empty() => Ok(Self {
binding: binding.to_owned(),
}),
_ => Err(Error::Configuration(
"expected a `d1://<binding>` URL with a binding name".into(),
)),
}
}
fn to_url_lossy(&self) -> Url {
Url::parse(&format!("d1://{}", self.binding)).expect("a binding name parsed from a URL")
}
fn connect(&self) -> impl Future<Output = Result<D1Connection, Error>> + Send + '_ {
future::ready(Err(Error::Configuration(
format!(
"a D1 connection cannot be opened from a URL; get the `{}` binding from the \
Worker's environment with `D1Connection::from_env`",
self.binding
)
.into(),
)))
}
fn log_statements(self, _level: LevelFilter) -> Self {
self
}
fn log_slow_statements(self, _level: LevelFilter, _duration: Duration) -> Self {
self
}
}
#[cfg(test)]
mod tests {
use sqlx_core::connection::ConnectOptions;
use super::D1ConnectOptions;
#[test]
fn parses_a_binding_url() {
let options: D1ConnectOptions = "d1://DB".parse().unwrap();
assert_eq!(options.binding(), "DB");
assert_eq!(options.to_url_lossy().as_str(), "d1://DB");
}
#[test]
fn rejects_other_schemes_and_a_missing_binding() {
assert!("sqlite://DB".parse::<D1ConnectOptions>().is_err());
assert!("d1://".parse::<D1ConnectOptions>().is_err());
}
}