pipestreamr 0.2.0

Rust SDK and CLI for the PipeStreamr unified events API
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

use uuid::Uuid;

use crate::error::Error;
use crate::models::*;

/// Client for the PipeStreamr public API.
///
/// # Example
/// ```no_run
/// use pipestreamr::PipeStreamrClient;
///
/// # async fn run() -> Result<(), pipestreamr::Error> {
/// let client = PipeStreamrClient::new("https://api.pipestreamr.com", "ps_your_api_key");
/// let events = client.list_events(&Default::default()).await?;
/// println!("{} events", events.events.len());
/// # Ok(())
/// # }
/// ```
pub struct PipeStreamrClient {
    base_url: String,
    api_key: String,
    http: reqwest::Client,
}

impl PipeStreamrClient {
    pub fn new(base_url: &str, api_key: &str) -> Self {
        Self {
            base_url: base_url.trim_end_matches('/').to_string(),
            api_key: api_key.to_string(),
            http: reqwest::Client::new(),
        }
    }

    async fn check_response(&self, resp: reqwest::Response) -> Result<reqwest::Response, Error> {
        if resp.status().is_success() {
            return Ok(resp);
        }
        let status = resp.status().as_u16();
        let message = resp
            .json::<serde_json::Value>()
            .await
            .ok()
            .and_then(|v| v.get("error")?.as_str().map(String::from))
            .unwrap_or_else(|| "Unknown error".into());
        Err(Error::Api { status, message })
    }

    /// List events with optional filters and cursor-based pagination.
    pub async fn list_events(&self, query: &EventQuery) -> Result<EventListResponse, Error> {
        let mut params: Vec<(&str, String)> = Vec::new();
        if let Some(ref t) = query.event_type {
            params.push(("type", t.clone()));
        }
        if let Some(ref p) = query.platform {
            params.push(("platform", p.clone()));
        }
        if let Some(sev) = query.severity {
            params.push(("severity", sev.to_string()));
        }
        if let Some(ref svc) = query.service {
            params.push(("service", svc.clone()));
        }
        if let Some(ref f) = query.from {
            params.push(("from", f.clone()));
        }
        if let Some(ref s) = query.search {
            params.push(("search", s.clone()));
        }
        if let Some(s) = query.since {
            params.push(("since", s.to_rfc3339()));
        }
        if let Some(u) = query.until {
            params.push(("until", u.to_rfc3339()));
        }
        if let Some(l) = query.limit {
            params.push(("limit", l.to_string()));
        }
        if let Some(ref c) = query.cursor {
            params.push(("cursor", c.clone()));
        }

        let resp = self
            .http
            .get(format!("{}/api/v1/events", self.base_url))
            .bearer_auth(&self.api_key)
            .query(&params)
            .send()
            .await?;
        let resp = self.check_response(resp).await?;
        Ok(resp.json().await?)
    }

    /// Get a single event by ID.
    pub async fn get_event(&self, id: Uuid) -> Result<EventResponse, Error> {
        let resp = self
            .http
            .get(format!("{}/api/v1/events/{}", self.base_url, id))
            .bearer_auth(&self.api_key)
            .send()
            .await?;
        let resp = self.check_response(resp).await?;
        Ok(resp.json().await?)
    }

    /// Get message statistics (total, today, by platform).
    pub async fn stats(&self) -> Result<StatsResponse, Error> {
        let resp = self
            .http
            .get(format!("{}/api/v1/stats", self.base_url))
            .bearer_auth(&self.api_key)
            .send()
            .await?;
        let resp = self.check_response(resp).await?;
        Ok(resp.json().await?)
    }

    /// Check if the API server is healthy.
    pub async fn health(&self) -> Result<bool, Error> {
        let resp = self
            .http
            .get(format!("{}/health", self.base_url))
            .send()
            .await?;
        Ok(resp.status().is_success())
    }
}