odata-simple-client 0.2.6

Simplified OpenData API Client
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

use crate::{
    deserialize_as, path::Format, Connector, DataSource, Error, GetRequest, ListRequest, Page,
    PathBuilder,
};
use hyper::{Body, Response};
use serde::de::DeserializeOwned;
use std::{num::NonZeroU32, sync::Arc};

pub type RateLimiter = governor::RateLimiter<
    governor::state::NotKeyed,
    governor::state::InMemoryState,
    governor::clock::QuantaClock,
>;
pub use governor::Quota;

/// Rate-limited wrapper around a DataSource. Requires the 'rate-limiter' feature to be enabled.
/// Cloning the RateLimitedDataSource shares the rate-limiting mechanism between the two copies,
/// preserving the rate-limiting guarantees across all of them.
#[derive(Debug, Clone)]
pub struct RateLimitedDataSource<C>
where
    C: Connector,
{
    datasource: DataSource<C>,
    rate_limiter: Arc<RateLimiter>,
}

impl<C> RateLimitedDataSource<C>
where
    C: Connector,
{
    /// Construct a RateLimitedDataSource from an existing [`DataSource`], and a [`Quota`]
    pub fn new(datasource: DataSource<C>, quota: Quota) -> Self {
        Self {
            datasource,
            rate_limiter: Arc::new(RateLimiter::direct(quota)),
        }
    }

    /// Construct a RateLimitedResource from an existing [`DataSource`],
    /// and a non-zero integer indicating the maximum number of requests
    /// the DataSource should serve per second.
    pub fn per_second(datasource: DataSource<C>, per_second: NonZeroU32) -> Self {
        Self::new(datasource, Quota::per_second(per_second))
    }

    async fn execute<R>(&self, request: R) -> Result<Response<Body>, Error>
    where
        R: Into<PathBuilder>,
    {
        self.rate_limiter.until_ready().await;
        self.datasource.execute(request).await
    }

    /// Fetch two resources on a datasource rate-limited to one per second,
    /// and assert more than a second passed in total.
    /// ```rust
    /// # use hyper::{Client, client::HttpConnector};
    /// # use hyper_openssl::{HttpsConnector};
    /// # use odata_simple_client::{RateLimitedDataSource, DataSource, GetRequest};
    /// # use serde::Deserialize;
    /// #
    /// # let client: Client<HttpsConnector<HttpConnector>> =
    /// #   Client::builder().build(HttpsConnector::<HttpConnector>::new().unwrap());
    /// #
    /// # let datasource = DataSource::new(client, "oda.ft.dk", Some(String::from("/api"))).unwrap();
    /// #
    /// #[derive(Deserialize)]
    /// struct Dokument {
    ///     titel: String,
    /// }
    ///
    /// let datasource = RateLimitedDataSource::per_second(
    ///     datasource,
    ///     std::num::NonZeroU32::new(1u32).unwrap()
    /// );
    ///
    /// let start = std::time::Instant::now();
    ///
    /// # tokio_test::block_on(async {
    /// let first: Dokument = datasource.fetch(
    ///         GetRequest::new("Dokument", 24)
    ///     ).await.unwrap();
    ///
    /// let second: Dokument = datasource.fetch(
    ///         GetRequest::new("Dokument", 26)
    ///     ).await.unwrap();
    ///
    /// assert!(start.elapsed().as_millis() >= 1000);
    ///
    /// # assert_eq!(first.titel, "Grund- og nærhedsnotat vedr. sanktioner på toldområdet");
    /// # assert_eq!(second.titel, "Revideret grund- og nærhedsnotat om sanktioner på toldområdet\n");
    /// # });
    /// ```
    pub async fn fetch<T>(&self, request: GetRequest) -> Result<T, Error>
    where
        T: DeserializeOwned,
    {
        let response = self
            .execute(Into::<PathBuilder>::into(request).format(Format::Json))
            .await?;
        deserialize_as::<T>(response).await
    }

    /// Fetch a [`Page`]d list of resources using a [`ListRequest`]
    /// ```rust
    /// # use hyper::{Client, client::HttpConnector};
    /// # use hyper_openssl::{HttpsConnector};
    /// # use odata_simple_client::{DataSource, ListRequest, Page, InlineCount, RateLimitedDataSource};
    /// # use serde::Deserialize;
    /// #
    /// # let client: Client<HttpsConnector<HttpConnector>> =
    /// #   Client::builder().build(HttpsConnector::<HttpConnector>::new().unwrap());
    /// #
    /// # let datasource = DataSource::new(client, "oda.ft.dk", Some(String::from("/api"))).unwrap();
    ///
    /// // Rate-limit the data source.
    /// let datasource = RateLimitedDataSource::per_second(
    ///     datasource,
    ///     std::num::NonZeroU32::new(1u32).unwrap()
    /// );
    /// #
    /// #[derive(Deserialize)]
    /// struct Dokument {
    ///     titel: String,
    /// }
    ///
    /// # tokio_test::block_on(async {
    /// let page: Page<Dokument> = datasource
    ///     .fetch_paged(ListRequest::new("Dokument")
    ///         .inline_count(InlineCount::AllPages)
    ///     ).await.unwrap();
    /// assert!(page.count.unwrap().parse::<u32>().unwrap() > 0)
    /// # });
    /// ```
    pub async fn fetch_paged<T>(&self, request: ListRequest) -> Result<Page<T>, Error>
    where
        T: DeserializeOwned,
    {
        let response = self
            .execute(Into::<PathBuilder>::into(request).format(Format::Json))
            .await?;
        deserialize_as::<Page<T>>(response).await
    }
}