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, crypto::CryptoClient, news::NewsClient,
14    options::OptionsClient, 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 crypto(&self) -> CryptoClient {
77        CryptoClient::new(self.inner.clone())
78    }
79
80    #[must_use]
81    pub fn options(&self) -> OptionsClient {
82        OptionsClient::new(self.inner.clone())
83    }
84
85    #[must_use]
86    pub fn news(&self) -> NewsClient {
87        NewsClient::new(self.inner.clone())
88    }
89
90    #[must_use]
91    pub fn corporate_actions(&self) -> CorporateActionsClient {
92        CorporateActionsClient::new(self.inner.clone())
93    }
94}
95
96impl fmt::Debug for Client {
97    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
98        f.debug_struct("Client")
99            .field("base_url", self.base_url())
100            .field("http", &"HttpClient")
101            .field("auth", &"[REDACTED]")
102            .finish()
103    }
104}
105
106impl fmt::Debug for ClientInner {
107    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
108        f.debug_struct("ClientInner")
109            .field("base_url", &self.base_url)
110            .field("http", &"HttpClient")
111            .field("auth", &"[REDACTED]")
112            .finish()
113    }
114}
115
116impl fmt::Debug for ClientBuilder {
117    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
118        f.debug_struct("ClientBuilder")
119            .field("api_key", &redacted_option(&self.api_key))
120            .field("secret_key", &redacted_option(&self.secret_key))
121            .field("base_url", &self.base_url)
122            .field("timeout", &self.timeout)
123            .field(
124                "observer",
125                &self.observer.as_ref().map(|_| "TransportObserver"),
126            )
127            .field("retry_config", &self.retry_config)
128            .field("max_in_flight", &self.max_in_flight)
129            .finish()
130    }
131}
132
133impl ClientBuilder {
134    #[must_use]
135    pub fn credentials(mut self, credentials: Credentials) -> Self {
136        self.api_key = Some(credentials.api_key().to_owned());
137        self.secret_key = Some(credentials.secret_key().to_owned());
138        self
139    }
140
141    #[must_use]
142    pub fn api_key(mut self, api_key: impl Into<String>) -> Self {
143        self.api_key = Some(api_key.into());
144        self
145    }
146
147    #[must_use]
148    pub fn secret_key(mut self, secret_key: impl Into<String>) -> Self {
149        self.secret_key = Some(secret_key.into());
150        self
151    }
152
153    #[must_use]
154    pub fn base_url(mut self, base_url: BaseUrl) -> Self {
155        self.base_url = Some(base_url);
156        self
157    }
158
159    pub fn base_url_str(mut self, base_url: impl AsRef<str>) -> Result<Self, Error> {
160        self.base_url = Some(BaseUrl::new(base_url.as_ref())?);
161        Ok(self)
162    }
163
164    pub fn credentials_from_env(self) -> Result<Self, Error> {
165        self.credentials_from_env_names(DATA_API_KEY_ENV, DATA_SECRET_KEY_ENV)
166    }
167
168    pub fn credentials_from_env_names(
169        mut self,
170        api_key_var: &str,
171        secret_key_var: &str,
172    ) -> Result<Self, Error> {
173        if let Some(credentials) = env::credentials_from_env_names(api_key_var, secret_key_var)? {
174            return Ok(self.credentials(credentials));
175        }
176
177        if let Some(credentials) = env::credentials_from_env()? {
178            self = self.credentials(credentials);
179        }
180
181        Ok(self)
182    }
183
184    #[must_use]
185    pub fn timeout(mut self, timeout: Duration) -> Self {
186        self.timeout = Some(timeout);
187        self
188    }
189
190    #[must_use]
191    pub fn observer(mut self, observer: Arc<dyn TransportObserver>) -> Self {
192        self.observer = Some(observer);
193        self
194    }
195
196    #[must_use]
197    pub fn retry_config(mut self, retry_config: RetryConfig) -> Self {
198        self.retry_config = retry_config;
199        self
200    }
201
202    #[must_use]
203    pub fn max_in_flight(mut self, max_in_flight: usize) -> Self {
204        self.max_in_flight = Some(max_in_flight);
205        self
206    }
207
208    pub fn build(self) -> Result<Client, Error> {
209        let credentials = match (self.api_key, self.secret_key) {
210            (Some(api_key), Some(secret_key)) => Credentials::new(api_key, secret_key)?,
211            (None, None) => return Err(Error::MissingCredentials),
212            _ => {
213                return Err(Error::InvalidConfiguration(
214                    "api_key and secret_key must be paired".to_owned(),
215                ));
216            }
217        };
218
219        let base_url = match self.base_url {
220            Some(base_url) => base_url,
221            None => BaseUrl::new(DEFAULT_DATA_BASE_URL)?,
222        };
223        let auth = StaticHeaderAuthenticator::from_pairs([
224            (APCA_API_KEY_HEADER, credentials.api_key()),
225            (APCA_API_SECRET_HEADER, credentials.secret_key()),
226        ])?;
227
228        let timeout = self.timeout.unwrap_or(DEFAULT_TIMEOUT);
229        let reqwest_client = Self::build_reqwest_client(timeout)?;
230
231        let mut http_builder = HttpClient::builder()
232            .retry_config(self.retry_config)
233            .reqwest_client(reqwest_client);
234        if let Some(observer) = self.observer {
235            http_builder = http_builder.observer(observer);
236        }
237        http_builder = http_builder.concurrency_limit(ConcurrencyLimit::new(Some(
238            self.max_in_flight.unwrap_or(DEFAULT_MAX_IN_FLIGHT),
239        )));
240
241        let http = http_builder.build()?;
242
243        Ok(Client {
244            inner: Arc::new(ClientInner {
245                http,
246                auth,
247                base_url,
248            }),
249        })
250    }
251
252    fn build_reqwest_client(timeout: Duration) -> Result<reqwest::Client, Error> {
253        reqwest::Client::builder()
254            .no_proxy()
255            .pool_max_idle_per_host(DEFAULT_POOL_MAX_IDLE_PER_HOST)
256            .pool_idle_timeout(DEFAULT_POOL_IDLE_TIMEOUT)
257            .tcp_keepalive(DEFAULT_TCP_KEEPALIVE)
258            .timeout(timeout)
259            .http1_only()
260            .build()
261            .map_err(|error| alpaca_http::Error::from_reqwest(error, None).into())
262    }
263}
264
265impl ClientInner {
266    #[allow(dead_code)]
267    pub(crate) async fn send_json<T>(&self, request: RequestParts) -> Result<HttpResponse<T>, Error>
268    where
269        T: DeserializeOwned,
270    {
271        self.http
272            .send_json(&self.base_url, request, Some(&self.auth))
273            .await
274            .map_err(Error::from)
275    }
276
277    #[allow(dead_code)]
278    pub(crate) async fn send_text(
279        &self,
280        request: RequestParts,
281    ) -> Result<HttpResponse<String>, Error> {
282        self.http
283            .send_text(&self.base_url, request, Some(&self.auth))
284            .await
285            .map_err(Error::from)
286    }
287
288    #[allow(dead_code)]
289    pub(crate) async fn send_no_content(
290        &self,
291        request: RequestParts,
292    ) -> Result<HttpResponse<NoContent>, Error> {
293        self.http
294            .send_no_content(&self.base_url, request, Some(&self.auth))
295            .await
296            .map_err(Error::from)
297    }
298}
299
300fn redacted_option(value: &Option<String>) -> &'static str {
301    match value {
302        Some(_) => "[REDACTED]",
303        None => "None",
304    }
305}
alpaca_data/client.rs

alpaca_data/
client.rs
