Struct chewdata::connector::psql::Psql

pub struct Psql {
    pub endpoint: String,
    pub database: String,
    pub collection: String,
    pub parameters: Value,
    pub query: Option<String>,
    pub paginator_type: PaginatorType,
    pub counter_type: CounterType,
    pub max_connections: u32,
}

Fields

endpoint: String

database: String

collection: String

parameters: Value

query: Option<String>

paginator_type: PaginatorType

counter_type: CounterType

max_connections: u32

Implementations

impl Psql

pub fn query_sanitized( &self, query: &str, parameters: &Value ) -> Result<(String, PgArguments)>

Transform mustache query into sanitized psql query with his arguments Query: SELECT * FROM {{ collection }} WHERE “a”={{ a }}; Return: (SELECT * FROM collection WHERE “a” = $1, “a”)

pub async fn client(&self) -> Result<Pool<Postgres>>

Get the current client

Trait Implementations

impl Clone for Psql

fn clone(&self) -> Psql

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Connector for Psql

fn path(&self) -> String

See Connector::path for more details.

fn set_parameters(&mut self, parameters: Value)

See Connector::set_parameters for more details.

fn is_variable(&self) -> bool

See Connector::is_variable for more details.

fn is_resource_will_change(&self, _new_parameters: Value) -> Result<bool>

See Connector::is_resource_will_change for more details.

fn len<'life0, 'async_trait>( &'life0 self ) -> Pin<Box<dyn Future<Output = Result<usize>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

See Connector::len for more details.

Examples

use chewdata::connector::psql::Psql;
use chewdata::document::json::Json;
use chewdata::connector::Connector;
use async_std::prelude::*;
use std::io;

#[async_std::main]
async fn main() -> io::Result<()> {
    let mut connector = Psql::default();
    connector.endpoint = "psql://admin:admin@localhost:5432".into();
    connector.database = "postgres".into();
    connector.collection = "public.read".into();
    let len = connector.len().await.unwrap();
    assert!(0 < len, "The connector should have a size upper than zero");

    Ok(())
}

fn fetch<'life0, 'async_trait>( &'life0 mut self ) -> Pin<Box<dyn Future<Output = Result<Option<DataStream>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

See Connector::fetch for more details.

Examples

use chewdata::connector::psql::Psql;
use chewdata::document::json::Json;
use chewdata::connector::Connector;
use serde_json::Value;
use async_std::prelude::*;
use std::io;
use futures::StreamExt;

#[async_std::main]
async fn main() -> io::Result<()> {
    let mut connector = Psql::default();
    connector.endpoint = "postgres://admin:admin@localhost".into();
    connector.database = "postgres".into();
    connector.collection = "public.send_with_key".into();
    connector.query =
        Some("SELECT * FROM {{ collection }} WHERE \"number\" = {{ number }} AND \"string\" = {{ string }} AND \"boolean\" = {{ boolean }} AND \"null\" = {{ null }} AND \"array\" = {{ array }} AND \"object\" = {{ object }} AND \"date\" = {{ date }} AND \"round\" = {{ round }};".to_string());
    let data: Value = serde_json::from_str(
        r#"{"number":1,"group":1,"string":"value to test 5416","boolean":false,"null":null,"array":[1,2],"object":{"field":"value"},"date":"2019-12-31T00:00:00.000Z","round":10.156}"#,
    )
    .unwrap();
    connector.set_parameters(data);
    let datastream = connector.fetch().await.unwrap().unwrap();
    assert!(
        1 == datastream.count().await,
        "The datastream must contain one record"
    );

    Ok(())
}

fn send<'life0, 'life1, 'async_trait>( &'life0 mut self, dataset: &'life1 DataSet ) -> Pin<Box<dyn Future<Output = Result<Option<DataStream>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

See Connector::send for more details.

Examples

use chewdata::connector::psql::Psql;
use chewdata::document::json::Json;
use chewdata::connector::Connector;
use chewdata::DataResult;
use async_std::prelude::*;
use std::io;

#[async_std::main]
async fn main() -> io::Result<()> {
    let mut connector = Psql::default();
    connector.endpoint = "postgres://admin:admin@localhost:5432".into();
    connector.database = "postgres".into();
    connector.collection = "public.send".into();
    connector.erase().await.unwrap();

    let expected_result1 = DataResult::Ok(
        serde_json::from_str(
            r#"{"number":110,"string":"value1","boolean":true,"special_char":"€"}"#,
        )
        .unwrap(),
    );
    let dataset = vec![expected_result1.clone()];
    connector.send(&dataset).await.unwrap();

    Ok(())
}

fn erase<'life0, 'async_trait>( &'life0 mut self ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

See Connector::erase for more details.

Examples

use chewdata::connector::psql::Psql;
use chewdata::document::json::Json;
use chewdata::connector::Connector;
use chewdata::DataResult;
use async_std::prelude::*;
use std::io;

#[async_std::main]
async fn main() -> io::Result<()> {
    let mut connector = Psql::default();
    connector.endpoint = "psql://admin:admin@localhost".into();
    connector.database = "postgres".into();
    connector.collection = "public.erase".into();

    let expected_result1 =
        DataResult::Ok(serde_json::from_str(r#"{"data":"value1"}"#).unwrap());
    let dataset = vec![expected_result1];
    connector.send(&dataset).await.unwrap();
    connector.erase().await.unwrap();

    let mut connector_read = connector.clone();
    let datastream = connector_read.fetch().await.unwrap();
    assert!(datastream.is_none(), "The datastream should be empty");

    Ok(())
}

fn paginate<'life0, 'async_trait>( &'life0 self ) -> Pin<Box<dyn Future<Output = Result<Pin<Box<dyn Stream<Item = Result<Box<dyn Connector>>> + Send>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

See Connector::paginate for more details.

fn set_document(&mut self, _document: Box<dyn Document>) -> Result<()>

fn document(&self) -> Result<&Box<dyn Document>>

fn set_metadata(&mut self, _metadata: Metadata)

Set the connector metadata that can change with the document metadata.

fn metadata(&self) -> Metadata

Get the connector metadata

fn is_empty<'life0, 'async_trait>( &'life0 self ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Check if the resource is empty.

impl Debug for Psql

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for Psql

fn default() -> Self

Returns the “default value” for a type.

impl<'de> Deserialize<'de> for Psql

where Psql: Default,

fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>

where __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.

impl Serialize for Psql

fn serialize<__S>(&self, __serializer: __S) -> Result<__S::Ok, __S::Error>

where __S: Serializer,

Serialize this value into the given Serde serializer.