use fallible_iterator::FallibleIterator;
use futures::executor;
use std::io::{BufRead, Read};
use tokio_postgres::types::{ToSql, Type};
use tokio_postgres::{Error, Row, SimpleQueryMessage};
use crate::copy_in_stream::CopyInStream;
use crate::copy_out_reader::CopyOutReader;
use crate::iter::Iter;
use crate::{Portal, Statement, ToStatement};
pub struct Transaction<'a>(tokio_postgres::Transaction<'a>);
impl<'a> Transaction<'a> {
pub(crate) fn new(transaction: tokio_postgres::Transaction<'a>) -> Transaction<'a> {
Transaction(transaction)
}
pub fn commit(self) -> Result<(), Error> {
executor::block_on(self.0.commit())
}
pub fn rollback(self) -> Result<(), Error> {
executor::block_on(self.0.rollback())
}
pub fn prepare(&mut self, query: &str) -> Result<Statement, Error> {
executor::block_on(self.0.prepare(query))
}
pub fn prepare_typed(&mut self, query: &str, types: &[Type]) -> Result<Statement, Error> {
executor::block_on(self.0.prepare_typed(query, types))
}
pub fn execute<T>(&mut self, query: &T, params: &[&(dyn ToSql + Sync)]) -> Result<u64, Error>
where
T: ?Sized + ToStatement,
{
executor::block_on(self.0.execute(query, params))
}
pub fn query<T>(&mut self, query: &T, params: &[&(dyn ToSql + Sync)]) -> Result<Vec<Row>, Error>
where
T: ?Sized + ToStatement,
{
executor::block_on(self.0.query(query, params))
}
pub fn query_raw<'b, T, I>(
&mut self,
query: &T,
params: I,
) -> Result<impl FallibleIterator<Item = Row, Error = Error>, Error>
where
T: ?Sized + ToStatement,
I: IntoIterator<Item = &'b dyn ToSql>,
I::IntoIter: ExactSizeIterator,
{
let stream = executor::block_on(self.0.query_raw(query, params))?;
Ok(Iter::new(stream))
}
pub fn bind<T>(&mut self, query: &T, params: &[&(dyn ToSql + Sync)]) -> Result<Portal, Error>
where
T: ?Sized + ToStatement,
{
executor::block_on(self.0.bind(query, params))
}
pub fn query_portal(&mut self, portal: &Portal, max_rows: i32) -> Result<Vec<Row>, Error> {
executor::block_on(self.0.query_portal(portal, max_rows))
}
pub fn query_portal_raw(
&mut self,
portal: &Portal,
max_rows: i32,
) -> Result<impl FallibleIterator<Item = Row, Error = Error>, Error> {
let stream = executor::block_on(self.0.query_portal_raw(portal, max_rows))?;
Ok(Iter::new(stream))
}
pub fn copy_in<T, R>(
&mut self,
query: &T,
params: &[&(dyn ToSql + Sync)],
reader: R,
) -> Result<u64, Error>
where
T: ?Sized + ToStatement,
R: Read + Unpin,
{
executor::block_on(self.0.copy_in(query, params, CopyInStream(reader)))
}
pub fn copy_out<T>(
&mut self,
query: &T,
params: &[&(dyn ToSql + Sync)],
) -> Result<impl BufRead, Error>
where
T: ?Sized + ToStatement,
{
let stream = executor::block_on(self.0.copy_out(query, params))?;
CopyOutReader::new(stream)
}
pub fn simple_query(&mut self, query: &str) -> Result<Vec<SimpleQueryMessage>, Error> {
executor::block_on(self.0.simple_query(query))
}
pub fn batch_execute(&mut self, query: &str) -> Result<(), Error> {
executor::block_on(self.0.batch_execute(query))
}
pub fn transaction(&mut self) -> Result<Transaction<'_>, Error> {
let transaction = executor::block_on(self.0.transaction())?;
Ok(Transaction(transaction))
}
}