Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Get for text results when querying #679

Closed
wants to merge 1 commit into from
Closed

Get for text results when querying #679

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 30 additions & 0 deletions postgres-types/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,7 @@ pub use postgres_protocol::Oid;

pub use crate::special::{Date, Timestamp};
use bytes::BytesMut;
use std::fmt::{Display, Formatter};

// Number of seconds from 1970-01-01 to 2000-01-01
const TIME_SEC_CONVERSION: u64 = 946_684_800;
Expand Down Expand Up @@ -542,6 +543,35 @@ impl<'a> FromSql<'a> for &'a str {
}
}

/// A newtype for String that accepts postgres textual data as
/// opposed to `String` that accepts only a limited set of types (TEXT, VARCHAR...)
///
pub struct AcceptEverythingString(String);

impl<'a> FromSql<'a> for AcceptEverythingString {
fn from_sql(_ty: &Type, raw: &'a [u8]) -> Result<Self, Box<dyn Error + Sync + Send>> {
types::text_from_sql(raw)
.map(ToString::to_string)
.map(Into::into)
}

fn accepts(_ty: &Type) -> bool {
true
}
}

impl From<String> for AcceptEverythingString {
fn from(value: String) -> Self {
AcceptEverythingString(value)
}
}

impl Display for AcceptEverythingString {
fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
Display::fmt(&self.0, formatter)
}
}

macro_rules! simple_from {
($t:ty, $f:ident, $($expected:ident),+) => {
impl<'a> FromSql<'a> for $t {
Expand Down
17 changes: 17 additions & 0 deletions postgres/src/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,23 @@ impl Client {
self.connection.block_on(self.client.query(query, params))
}

/// Same as `query` but returns text results instead of binary
///
/// We introduce an additional method rather than parameters not to break all
/// the conversions and marshalling supplied by `postgres-types` and feature crates and
/// inasmuch as this allows us to maintain API compatibility with upstream
pub fn query_with_text_results<T>(
&mut self,
query: &T,
params: &[&(dyn ToSql + Sync)],
) -> Result<Vec<Row>, Error>
where
T: ?Sized + ToStatement,
{
self.connection
.block_on(self.client.query_with_text_results(query, params))
}

/// Executes a statement which returns a single row, returning it.
///
/// Returns an error if the query does not return exactly one row.
Expand Down
35 changes: 34 additions & 1 deletion tokio-postgres/src/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -226,6 +226,25 @@ impl Client {
.await
}

/// Same as `query` but returns text results instead of binary
///
/// We introduce an additional method rather than parameters not to break all
/// the conversions and marshalling supplied by `postgres-types` and feature crates and
/// inasmuch as this allows us to maintain API compatibility with upstream
pub async fn query_with_text_results<T>(
&self,
statement: &T,
params: &[&(dyn ToSql + Sync)],
) -> Result<Vec<Row>, Error>
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This shouldn't use the Row type since FromSql will not work with almost any of the types returned in text mode.

Copy link
Author

@EPashkin EPashkin Oct 20, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You right, in your project it used with proxy function, but in general case it need more thoughts.

decode_column_value_common::<postgres_types::AcceptEverythingString>(row, i)


fn decode_column_value_common<'a, T: postgres::types::FromSql<'a> + std::fmt::Display>(
    row: &'a Row,
    i: usize,
) -> String {
    match row.try_get::<usize, T>(i) {
        Ok(data) => format!("{}", data),
        Err(_) => "<null>".to_string(),
    }
}

where
T: ?Sized + ToStatement,
{
self.query_raw_common(statement, slice_iter(params), false)
.await?
.try_collect()
.await
}

/// Executes a statement which returns a single row, returning it.
///
/// Returns an error if the query does not return exactly one row.
Expand Down Expand Up @@ -343,13 +362,27 @@ impl Client {
/// # }
/// ```
pub async fn query_raw<'a, T, I>(&self, statement: &T, params: I) -> Result<RowStream, Error>
where
T: ?Sized + ToStatement,
I: IntoIterator<Item = &'a dyn ToSql>,
I::IntoIter: ExactSizeIterator,
{
self.query_raw_common(statement, params, true).await
}

async fn query_raw_common<'a, T, I>(
&self,
statement: &T,
params: I,
binary_results: bool,
) -> Result<RowStream, Error>
where
T: ?Sized + ToStatement,
I: IntoIterator<Item = &'a dyn ToSql>,
I::IntoIter: ExactSizeIterator,
{
let statement = statement.__convert().into_statement(self).await?;
query::query(&self.inner, statement, params).await
query::query_common(&self.inner, statement, params, binary_results).await
}

/// Executes a statement, returning the number of rows modified.
Expand Down
51 changes: 46 additions & 5 deletions tokio-postgres/src/query.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,19 @@ pub async fn query<'a, I>(
statement: Statement,
params: I,
) -> Result<RowStream, Error>
where
I: IntoIterator<Item = &'a dyn ToSql>,
I::IntoIter: ExactSizeIterator,
{
query_common(client, statement, params, true).await
}

pub async fn query_common<'a, I>(
client: &InnerClient,
statement: Statement,
params: I,
binary_results: bool,
) -> Result<RowStream, Error>
where
I: IntoIterator<Item = &'a dyn ToSql>,
I::IntoIter: ExactSizeIterator,
Expand All @@ -29,9 +42,9 @@ where
statement.name(),
params,
);
encode(client, &statement, params)?
encode_common(client, &statement, params, binary_results)?
} else {
encode(client, &statement, params)?
encode_common(client, &statement, params, binary_results)?
};
let responses = start(client, buf).await?;
Ok(RowStream {
Expand Down Expand Up @@ -115,23 +128,37 @@ async fn start(client: &InnerClient, buf: Bytes) -> Result<Responses, Error> {
}

pub fn encode<'a, I>(client: &InnerClient, statement: &Statement, params: I) -> Result<Bytes, Error>
where
I: IntoIterator<Item = &'a dyn ToSql>,
I::IntoIter: ExactSizeIterator,
{
encode_common(client, statement, params, true)
}

pub fn encode_common<'a, I>(
client: &InnerClient,
statement: &Statement,
params: I,
binary_results: bool,
) -> Result<Bytes, Error>
where
I: IntoIterator<Item = &'a dyn ToSql>,
I::IntoIter: ExactSizeIterator,
{
client.with_buf(|buf| {
encode_bind(statement, params, "", buf)?;
encode_bind_common(statement, params, "", buf, binary_results)?;
frontend::execute("", 0, buf).map_err(Error::encode)?;
frontend::sync(buf);
Ok(buf.split().freeze())
})
}

pub fn encode_bind<'a, I>(
pub fn encode_bind_common<'a, I>(
statement: &Statement,
params: I,
portal: &str,
buf: &mut BytesMut,
binary_results: bool,
) -> Result<(), Error>
where
I: IntoIterator<Item = &'a dyn ToSql>,
Expand All @@ -145,6 +172,7 @@ where
statement.params().len(),
params.len()
);
let binary_results = if binary_results { 1 } else { 0 };

let mut error_idx = 0;
let r = frontend::bind(
Expand All @@ -160,7 +188,7 @@ where
Err(e)
}
},
Some(1),
Some(binary_results),
buf,
);
match r {
Expand All @@ -170,6 +198,19 @@ where
}
}

pub fn encode_bind<'a, I>(
statement: &Statement,
params: I,
portal: &str,
buf: &mut BytesMut,
) -> Result<(), Error>
where
I: IntoIterator<Item = &'a dyn ToSql>,
I::IntoIter: ExactSizeIterator,
{
encode_bind_common(statement, params, portal, buf, true)
}

pin_project! {
/// A stream of table rows.
pub struct RowStream {
Expand Down