alpaca_data/
client.rs

1use std::fmt;
2use std::sync::Arc;
3use std::time::Duration;
4
5use alpaca_core::{BaseUrl, Credentials, env};
6use alpaca_http::{
7    ConcurrencyLimit, HttpClient, HttpResponse, NoContent, RequestParts, RetryConfig,
8    StaticHeaderAuthenticator, TransportObserver,
9};
10use serde::de::DeserializeOwned;
11
12use crate::{
13    Error, corporate_actions::CorporateActionsClient, news::NewsClient, options::OptionsClient,
14    stocks::StocksClient,
15};
16
17pub const DATA_API_KEY_ENV: &str = "ALPACA_DATA_API_KEY";
18pub const DATA_SECRET_KEY_ENV: &str = "ALPACA_DATA_SECRET_KEY";
19const DEFAULT_DATA_BASE_URL: &str = "https://data.alpaca.markets";
20const APCA_API_KEY_HEADER: &str = "APCA-API-KEY-ID";
21const APCA_API_SECRET_HEADER: &str = "APCA-API-SECRET-KEY";
22const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
23const DEFAULT_MAX_IN_FLIGHT: usize = 50;
24const DEFAULT_POOL_MAX_IDLE_PER_HOST: usize = 50;
25const DEFAULT_POOL_IDLE_TIMEOUT: Duration = Duration::from_secs(90);
26const DEFAULT_TCP_KEEPALIVE: Duration = Duration::from_secs(60);
27
28#[derive(Clone)]
29pub struct Client {
30    pub(crate) inner: Arc<ClientInner>,
31}
32
33#[allow(dead_code)]
34pub(crate) struct ClientInner {
35    http: HttpClient,
36    auth: StaticHeaderAuthenticator,
37    base_url: BaseUrl,
38}
39
40#[derive(Clone, Default)]
41pub struct ClientBuilder {
42    api_key: Option<String>,
43    secret_key: Option<String>,
44    base_url: Option<BaseUrl>,
45    timeout: Option<Duration>,
46    observer: Option<Arc<dyn TransportObserver>>,
47    retry_config: RetryConfig,
48    max_in_flight: Option<usize>,
49}
50
51impl Client {
52    #[must_use]
53    pub fn builder() -> ClientBuilder {
54        ClientBuilder::default()
55    }
56
57    pub fn new(credentials: Credentials) -> Result<Self, Error> {
58        Self::builder().credentials(credentials).build()
59    }
60
61    pub fn from_env() -> Result<Self, Error> {
62        Self::builder().credentials_from_env()?.build()
63    }
64
65    #[must_use]
66    pub fn base_url(&self) -> &BaseUrl {
67        &self.inner.base_url
68    }
69
70    #[must_use]
71    pub fn stocks(&self) -> StocksClient {
72        StocksClient::new(self.inner.clone())
73    }
74
75    #[must_use]
76    pub fn options(&self) -> OptionsClient {
77        OptionsClient::new(self.inner.clone())
78    }
79
80    #[must_use]
81    pub fn news(&self) -> NewsClient {
82        NewsClient::new(self.inner.clone())
83    }
84
85    #[must_use]
86    pub fn corporate_actions(&self) -> CorporateActionsClient {
87        CorporateActionsClient::new(self.inner.clone())
88    }
89}
90
91impl fmt::Debug for Client {
92    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
93        f.debug_struct("Client")
94            .field("base_url", self.base_url())
95            .field("http", &"HttpClient")
96            .field("auth", &"[REDACTED]")
97            .finish()
98    }
99}
100
101impl fmt::Debug for ClientInner {
102    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
103        f.debug_struct("ClientInner")
104            .field("base_url", &self.base_url)
105            .field("http", &"HttpClient")
106            .field("auth", &"[REDACTED]")
107            .finish()
108    }
109}
110
111impl fmt::Debug for ClientBuilder {
112    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
113        f.debug_struct("ClientBuilder")
114            .field("api_key", &redacted_option(&self.api_key))
115            .field("secret_key", &redacted_option(&self.secret_key))
116            .field("base_url", &self.base_url)
117            .field("timeout", &self.timeout)
118            .field(
119                "observer",
120                &self.observer.as_ref().map(|_| "TransportObserver"),
121            )
122            .field("retry_config", &self.retry_config)
123            .field("max_in_flight", &self.max_in_flight)
124            .finish()
125    }
126}
127
128impl ClientBuilder {
129    #[must_use]
130    pub fn credentials(mut self, credentials: Credentials) -> Self {
131        self.api_key = Some(credentials.api_key().to_owned());
132        self.secret_key = Some(credentials.secret_key().to_owned());
133        self
134    }
135
136    #[must_use]
137    pub fn api_key(mut self, api_key: impl Into<String>) -> Self {
138        self.api_key = Some(api_key.into());
139        self
140    }
141
142    #[must_use]
143    pub fn secret_key(mut self, secret_key: impl Into<String>) -> Self {
144        self.secret_key = Some(secret_key.into());
145        self
146    }
147
148    #[must_use]
149    pub fn base_url(mut self, base_url: BaseUrl) -> Self {
150        self.base_url = Some(base_url);
151        self
152    }
153
154    pub fn base_url_str(mut self, base_url: impl AsRef<str>) -> Result<Self, Error> {
155        self.base_url = Some(BaseUrl::new(base_url.as_ref())?);
156        Ok(self)
157    }
158
159    pub fn credentials_from_env(self) -> Result<Self, Error> {
160        self.credentials_from_env_names(DATA_API_KEY_ENV, DATA_SECRET_KEY_ENV)
161    }
162
163    pub fn credentials_from_env_names(
164        mut self,
165        api_key_var: &str,
166        secret_key_var: &str,
167    ) -> Result<Self, Error> {
168        if let Some(credentials) = env::credentials_from_env_names(api_key_var, secret_key_var)? {
169            return Ok(self.credentials(credentials));
170        }
171
172        if let Some(credentials) = env::credentials_from_env()? {
173            self = self.credentials(credentials);
174        }
175
176        Ok(self)
177    }
178
179    #[must_use]
180    pub fn timeout(mut self, timeout: Duration) -> Self {
181        self.timeout = Some(timeout);
182        self
183    }
184
185    #[must_use]
186    pub fn observer(mut self, observer: Arc<dyn TransportObserver>) -> Self {
187        self.observer = Some(observer);
188        self
189    }
190
191    #[must_use]
192    pub fn retry_config(mut self, retry_config: RetryConfig) -> Self {
193        self.retry_config = retry_config;
194        self
195    }
196
197    #[must_use]
198    pub fn max_in_flight(mut self, max_in_flight: usize) -> Self {
199        self.max_in_flight = Some(max_in_flight);
200        self
201    }
202
203    pub fn build(self) -> Result<Client, Error> {
204        let credentials = match (self.api_key, self.secret_key) {
205            (Some(api_key), Some(secret_key)) => Credentials::new(api_key, secret_key)?,
206            (None, None) => return Err(Error::MissingCredentials),
207            _ => {
208                return Err(Error::InvalidConfiguration(
209                    "api_key and secret_key must be paired".to_owned(),
210                ));
211            }
212        };
213
214        let base_url = match self.base_url {
215            Some(base_url) => base_url,
216            None => BaseUrl::new(DEFAULT_DATA_BASE_URL)?,
217        };
218        let auth = StaticHeaderAuthenticator::from_pairs([
219            (APCA_API_KEY_HEADER, credentials.api_key()),
220            (APCA_API_SECRET_HEADER, credentials.secret_key()),
221        ])?;
222
223        let timeout = self.timeout.unwrap_or(DEFAULT_TIMEOUT);
224        let reqwest_client = Self::build_reqwest_client(timeout)?;
225
226        let mut http_builder = HttpClient::builder()
227            .retry_config(self.retry_config)
228            .reqwest_client(reqwest_client);
229        if let Some(observer) = self.observer {
230            http_builder = http_builder.observer(observer);
231        }
232        http_builder = http_builder.concurrency_limit(ConcurrencyLimit::new(Some(
233            self.max_in_flight.unwrap_or(DEFAULT_MAX_IN_FLIGHT),
234        )));
235
236        let http = http_builder.build()?;
237
238        Ok(Client {
239            inner: Arc::new(ClientInner {
240                http,
241                auth,
242                base_url,
243            }),
244        })
245    }
246
247    fn build_reqwest_client(timeout: Duration) -> Result<reqwest::Client, Error> {
248        reqwest::Client::builder()
249            .no_proxy()
250            .pool_max_idle_per_host(DEFAULT_POOL_MAX_IDLE_PER_HOST)
251            .pool_idle_timeout(DEFAULT_POOL_IDLE_TIMEOUT)
252            .tcp_keepalive(DEFAULT_TCP_KEEPALIVE)
253            .timeout(timeout)
254            .http1_only()
255            .build()
256            .map_err(|error| alpaca_http::Error::from_reqwest(error, None).into())
257    }
258}
259
260impl ClientInner {
261    #[allow(dead_code)]
262    pub(crate) async fn send_json<T>(&self, request: RequestParts) -> Result<HttpResponse<T>, Error>
263    where
264        T: DeserializeOwned,
265    {
266        self.http
267            .send_json(&self.base_url, request, Some(&self.auth))
268            .await
269            .map_err(Error::from)
270    }
271
272    #[allow(dead_code)]
273    pub(crate) async fn send_text(
274        &self,
275        request: RequestParts,
276    ) -> Result<HttpResponse<String>, Error> {
277        self.http
278            .send_text(&self.base_url, request, Some(&self.auth))
279            .await
280            .map_err(Error::from)
281    }
282
283    #[allow(dead_code)]
284    pub(crate) async fn send_no_content(
285        &self,
286        request: RequestParts,
287    ) -> Result<HttpResponse<NoContent>, Error> {
288        self.http
289            .send_no_content(&self.base_url, request, Some(&self.auth))
290            .await
291            .map_err(Error::from)
292    }
293}
294
295fn redacted_option(value: &Option<String>) -> &'static str {
296    match value {
297        Some(_) => "[REDACTED]",
298        None => "None",
299    }
300}
alpaca_data/client.rs

alpaca_data/
client.rs
