Documentation

Author: Empty2k12

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "influxdb"
-version = "0.1.0"
+version = "0.0.1"
 authors = ["Gero Gerke <11deutron11@gmail.com>"]
 edition = "2018"
 
@@ -10,7 +10,7 @@ futures = "0.1.27"
 tokio = "0.1.20"
 itertools = "0.8"
 failure = "0.1.5"
-serde = { version = "1.0.92", optional = true, features = ["derive"] }
+serde = { version = "1.0.92", optional = true }
 serde_json = { version = "1.0", optional = true }
 
 [features]

Cargo.toml

@@ -1,3 +1,20 @@
+//! Client which can read and write data from InfluxDB.
+//!
+//! # Arguments
+//!
+//!  * `url`: The URL where InfluxDB is running (ex. `http://localhost:8086`).
+//!  * `database`: The Database against which queries and writes will be run.
+//!
+//! # Examples
+//!
+//! ```rust
+//! use influxdb::client::InfluxDbClient;
+//!
+//! let client = InfluxDbClient::new("http://localhost:8086", "test");
+//!
+//! assert_eq!(client.database_name(), "test");
+//! ```
+
 use futures::{Future, Stream};
 use reqwest::r#async::{Client, Decoder};
 
@@ -6,30 +23,15 @@ use std::mem;
 use crate::error::InfluxDbError;
 use crate::query::{InfluxDbQuery, QueryType};
 
-/// Client which can read and write data from InfluxDB.
-///
-/// # Arguments
-///
-///  * `url`: The URL where InfluxDB is running (ex. `http://localhost:8086`).
-///  * `database`: The Database against which queries and writes will be run.
-///
-/// # Examples
-///
-/// ```rust
-/// use influxdb::client::InfluxDbClient;
-///
-/// let client = InfluxDbClient::new("http://localhost:8086", "test");
-///
-/// assert_eq!(client.database_name(), "test");
-/// ```
+/// Internal Representation of a Client
 pub struct InfluxDbClient {
     url: String,
     database: String,
     // auth: Option<InfluxDbAuthentication>
 }
 
 impl InfluxDbClient {
-    /// Instantiates a new [`InfluxDbClient`]
+    /// Instantiates a new [`InfluxDbClient`](crate::client::InfluxDbClient)
     ///
     /// # Arguments
     ///
@@ -45,23 +47,28 @@ impl InfluxDbClient {
     /// ```
     pub fn new<S1, S2>(url: S1, database: S2) -> Self
     where
-        S1: Into<String>,
-        S2: Into<String>,
+        S1: ToString,
+        S2: ToString,
     {
         InfluxDbClient {
-            url: url.into(),
-            database: database.into(),
+            url: url.to_string(),
+            database: database.to_string(),
         }
     }
 
-    pub fn database_name<'a>(&'a self) -> &'a str {
+    /// Returns the name of the database the client is using
+    pub fn database_name(&self) -> &str {
         &self.database
     }
 
-    pub fn database_url<'a>(&'a self) -> &'a str {
+    /// Returns the URL of the InfluxDB installation the client is using
+    pub fn database_url(&self) -> &str {
         &self.url
     }
 
+    /// Pings the InfluxDB Server
+    ///
+    /// Returns a tuple of build type and version number
     pub fn ping(&self) -> impl Future<Item = (String, String), Error = InfluxDbError> {
         Client::new()
             .get(format!("{}/ping", self.url).as_str())
@@ -82,11 +89,31 @@ impl InfluxDbClient {
 
                 (String::from(build), String::from(version))
             })
-            .map_err(|err| InfluxDbError::UnspecifiedError {
+            .map_err(|err| InfluxDbError::ProtocolError {
                 error: format!("{}", err),
             })
     }
 
+    /// Sends a [`InfluxDbReadQuery`](crate::query::read_query::InfluxDbReadQuery) or [`InfluxDbWriteQuery`](crate::query::write_query::InfluxDbWriteQuery) to the InfluxDB Server.InfluxDbError
+    ///
+    /// A version capable of parsing the returned string is available under the [serde_integration](crate::integrations::serde_integration)
+    ///
+    /// # Arguments
+    ///
+    ///  * `q`: Query of type [`InfluxDbReadQuery`](crate::query::read_query::InfluxDbReadQuery) or [`InfluxDbWriteQuery`](crate::query::write_query::InfluxDbWriteQuery)
+    ///
+    /// # Examples
+    ///
+    /// ```rust
+    /// use influxdb::client::InfluxDbClient;
+    /// use influxdb::query::InfluxDbQuery;
+    ///
+    /// let client = InfluxDbClient::new("http://localhost:8086", "test");
+    /// let _future = client.query(
+    ///     InfluxDbQuery::write_query("weather")
+    ///         .add_field("temperature", 82)
+    /// );
+    /// ```
     pub fn query<Q>(&self, q: Q) -> Box<dyn Future<Item = String, Error = InfluxDbError>>
     where
         Q: InfluxDbQuery,
@@ -101,7 +128,7 @@ impl InfluxDbClient {
 
         let query = match q.build() {
             Err(err) => {
-                let error = InfluxDbError::UnspecifiedError {
+                let error = InfluxDbError::InvalidQueryError {
                     error: format!("{}", err),
                 };
                 return Box::new(future::err::<String, InfluxDbError>(error));
@@ -146,7 +173,7 @@ impl InfluxDbClient {
                     let body = mem::replace(res.body_mut(), Decoder::empty());
                     body.concat2()
                 })
-                .map_err(|err| InfluxDbError::UnspecifiedError {
+                .map_err(|err| InfluxDbError::ProtocolError {
                     error: format!("{}", err),
                 })
                 .and_then(|body| {
@@ -155,16 +182,16 @@ impl InfluxDbClient {
 
                         // todo: improve error parsing without serde
                         if s.contains("\"error\"") {
-                            return futures::future::err(InfluxDbError::UnspecifiedError {
+                            return futures::future::err(InfluxDbError::DatabaseError {
                                 error: format!("influxdb error: \"{}\"", s),
                             });
                         }
 
                         return futures::future::ok(s);
                     }
 
-                    futures::future::err(InfluxDbError::UnspecifiedError {
-                        error: "some other error has happened here!".to_string(),
+                    futures::future::err(InfluxDbError::DeserializationError {
+                        error: "response could not be converted to UTF-8".to_string(),
                     })
                 }),
         )

src/client/mod.rs

@@ -1,16 +1,18 @@
+//! Errors that might happen in the crate
+
 #[derive(Debug, Fail)]
-/// Errors that might happen in the crate
 pub enum InfluxDbError {
-    #[fail(display = "query must contain at least one field")]
-    /// Error happens when query has zero fields
-    InvalidQueryError,
+    #[fail(display = "query is invalid: {}", error)]
+    /// Error happens when a query is invalid
+    InvalidQueryError { error: String },
+
+    #[fail(display = "http protocol error: {}", error)]
+    /// Error happens when a query is invalid
+    ProtocolError { error: String },
 
-    #[fail(
-        display = "an error happened: \"{}\". this case should be handled better, please file an issue.",
-        error
-    )]
-    /// todo: Error which is a placeholder for more meaningful errors. This should be refactored away.
-    UnspecifiedError { error: String },
+    #[fail(display = "http protocol error: {}", error)]
+    /// Error happens when Serde cannot deserialize the response
+    DeserializationError { error: String },
 
     #[fail(display = "InfluxDB encountered the following error: {}", error)]
     /// Error which has happened inside InfluxDB

src/error.rs

@@ -1,13 +1,52 @@
+//! Serde Integration for InfluxDB. Provides deserialization of query returns.
+//!
+//! When querying multiple series in the same query (e.g. with a regex query), it might be desirable to flat map
+//! the resulting series into a single `Vec` like so. The example assumes, that there are weather readings in multiple
+//! series named `weather_<city_name>` (e.g. `weather_berlin`, or `weather_london`). Since we're using a Regex query,
+//! we don't actually know which series will be returned. To assign the city name to the series, we can use the series
+//! `name`, InfluxDB provides alongside query results.
+//!
+//! ```rust,no_run
+//! use influxdb::query::InfluxDbQuery;
+//! use influxdb::client::InfluxDbClient;
+//! use serde::Deserialize;
+//!
+//! #[derive(Deserialize)]
+//! struct WeatherWithoutCityName {
+//!     temperature: i32
+//! }
+//!
+//! #[derive(Deserialize)]
+//! struct Weather {
+//!     city_name: String,
+//!     weather: WeatherWithoutCityName,
+//! }
+//!
+//! let mut rt = tokio::runtime::current_thread::Runtime::new().unwrap();
+//! let client = InfluxDbClient::new("http://localhost:8086", "test");
+//! let query = InfluxDbQuery::raw_read_query("SELECT temperature FROM /weather_[a-z]*$/ WHERE time > now() - 1m ORDER BY DESC");
+//! let _result = rt.block_on(client.json_query::<WeatherWithoutCityName, _>(query))
+//!     .map(|it| {
+//!         it.map(|series_vec| {
+//!             series_vec
+//!                 .into_iter()
+//!                 .map(|mut city_series| {
+//!                     let city_name = city_series.name.split("_").collect::<Vec<&str>>().remove(2);
+//!                     Weather { weather: city_series.values.remove(0), city_name: city_name.to_string() }
+//!                 }).collect::<Vec<Weather>>()
+//!         })
+//!     });
+
 use crate::client::InfluxDbClient;
 
 use serde::de::DeserializeOwned;
 
 use futures::{Future, Stream};
 use reqwest::r#async::{Client, Decoder};
+use std::mem;
 
-use serde_json;
 use serde::Deserialize;
-use std::mem;
+use serde_json;
 
 use crate::error::InfluxDbError;
 use crate::query::{InfluxDbQuery, QueryType};
@@ -31,14 +70,17 @@ pub struct InfluxDbReturn<T> {
 }
 
 #[derive(Deserialize, Debug)]
-#[doc(hidden)]
+/// Represents a returned series from InfluxDB
 pub struct InfluxDbSeries<T> {
     pub name: String,
     pub values: Vec<T>,
 }
 
 impl InfluxDbClient {
-    pub fn json_query<T: 'static, Q>(&self, q: Q) -> Box<dyn Future<Item = Option<Vec<InfluxDbSeries<T>>>, Error = InfluxDbError>>
+    pub fn json_query<T: 'static, Q>(
+        &self,
+        q: Q,
+    ) -> Box<dyn Future<Item = Option<Vec<InfluxDbSeries<T>>>, Error = InfluxDbError>>
     where
         Q: InfluxDbQuery,
         T: DeserializeOwned,
@@ -53,10 +95,12 @@ impl InfluxDbClient {
 
         let query = match q.build() {
             Err(err) => {
-                let error = InfluxDbError::UnspecifiedError {
+                let error = InfluxDbError::InvalidQueryError {
                     error: format!("{}", err),
                 };
-                return Box::new(future::err::<Option<Vec<InfluxDbSeries<T>>>, InfluxDbError>(error));
+                return Box::new(
+                    future::err::<Option<Vec<InfluxDbSeries<T>>>, InfluxDbError>(error),
+                );
             }
             Ok(query) => query,
         };
@@ -98,29 +142,31 @@ impl InfluxDbClient {
                     let body = mem::replace(res.body_mut(), Decoder::empty());
                     body.concat2()
                 })
-                .map_err(|err| InfluxDbError::UnspecifiedError {
-                    error: format!("{}", err)
+                .map_err(|err| InfluxDbError::ProtocolError {
+                    error: format!("{}", err),
                 })
                 .and_then(|body| {
                     // Try parsing InfluxDBs { "error": "error message here" }
                     if let Ok(error) = serde_json::from_slice::<_DatabaseError>(&body) {
                         return futures::future::err(InfluxDbError::DatabaseError {
-                            error: error.error.to_string()
-                        })
+                            error: error.error.to_string(),
+                        });
                     } else {
-                        // Json has another structure, let's try actually parsing it to the type we're deserializing 
+                        // Json has another structure, let's try actually parsing it to the type we're deserializing
                         let from_slice = serde_json::from_slice::<DatabaseQueryResult<T>>(&body);
 
                         let mut deserialized = match from_slice {
                             Ok(deserialized) => deserialized,
-                            Err(err) => return futures::future::err(InfluxDbError::UnspecifiedError {
-                                error: format!("serde error: {}", err)
-                            })
+                            Err(err) => {
+                                return futures::future::err(InfluxDbError::DeserializationError {
+                                    error: format!("serde error: {}", err),
+                                })
+                            }
                         };
 
                         return futures::future::result(Ok(deserialized.results.remove(0).series));
                     }
-                })
+                }),
         )
     }
 }