aspero_openalex/
lib.rs

1//! OpenAlex provider.
2//!
3//! Costs (help.openalex.org, 2026-11): single-work lookups are free, list and
4//! filter calls $0.0001, searches $0.001, from a free $1/day with an API key.
5//! So lookups of one ID use the free singleton endpoint, batches use one
6//! `filter` call per identifier kind, and search is the only real expense.
7//!
8//! Every call reports its cost and remaining budget in [`Usage`] (USD).
9
10mod http;
11mod wire;
12
13use std::sync::Arc;
14use std::time::Duration;
15
16use aspero_core::http::path_segment;
17use aspero_core::{
18    Capabilities, Cursor, Doi, Error, FieldSet, Filter, Hit, HttpClient, HttpRequest, HttpResponse,
19    IdKind, Lookup, Page, PageReq, Provider, ProviderId, Raw, ResiliencePolicy, Response, Result,
20    SearchMode, SearchQuery, SourceRecord, Usage, WorkId, WorkType,
21};
22use async_trait::async_trait;
23use secrecy::{ExposeSecret, SecretString};
24
25use crate::wire::{ARXIV_DOI_PREFIX, WireList};
26
27pub const ID: ProviderId = ProviderId::from_static("openalex");
28
29pub const DEFAULT_BASE_URL: &str = "https://api.openalex.org";
30
31/// OpenAlex's cap on `per-page` and on values OR-ed in one filter.
32const MAX_PAGE: u32 = 100;
33
34#[derive(Debug, Clone)]
35#[non_exhaustive]
36pub struct OpenAlexConfig {
37    /// Without a key OpenAlex allows $0.10/day; with a free key, $1/day.
38    pub api_key: Option<SecretString>,
39    pub base_url: String,
40}
41
42impl Default for OpenAlexConfig {
43    fn default() -> Self {
44        OpenAlexConfig {
45            api_key: None,
46            base_url: DEFAULT_BASE_URL.to_owned(),
47        }
48    }
49}
50
51pub struct OpenAlex {
52    config: OpenAlexConfig,
53    http: Arc<dyn HttpClient>,
54    capabilities: Capabilities,
55}
56
57impl OpenAlex {
58    pub fn new(config: OpenAlexConfig, http: Arc<dyn HttpClient>) -> Self {
59        let mut capabilities = Capabilities::default();
60        capabilities.lookup_kinds =
61            vec![IdKind::Doi, IdKind::Arxiv, IdKind::Pmid, IdKind::OpenAlex];
62        capabilities.max_batch = MAX_PAGE as usize;
63        capabilities.fields = FieldSet::ALL;
64        capabilities.search = true;
65        capabilities.search_filters = vec![
66            Filter::Years,
67            Filter::WorkType,
68            Filter::OpenAccess,
69            Filter::Place,
70        ];
71        OpenAlex {
72            config,
73            http,
74            capabilities,
75        }
76    }
77
78    /// OpenAlex allows 30 req/s with a key; aspero stays well under.
79    pub fn default_policy() -> ResiliencePolicy {
80        let mut policy = ResiliencePolicy::default();
81        policy.min_interval = Duration::from_millis(100);
82        policy
83    }
84
85    fn request(&self, url: String) -> HttpRequest {
86        let req = HttpRequest::get(url).header("accept", "application/json");
87        match &self.config.api_key {
88            Some(key) => req.header("authorization", format!("Bearer {}", key.expose_secret())),
89            None => req,
90        }
91    }
92
93    fn works_url(&self, rest: &str) -> String {
94        format!("{}/works{rest}", self.config.base_url.trim_end_matches('/'))
95    }
96
97    /// Sends and maps non-2xx statuses to errors. Usage comes back only
98    /// with a success; a spent quota's reset time comes in its error.
99    async fn send(&self, url: String) -> Result<(HttpResponse, Usage)> {
100        let resp = self.http.send(self.request(url)).await?;
101        http::check(&resp)?;
102        let usage = http::usage(&resp);
103        Ok((resp, usage))
104    }
105
106    async fn lookup_one(&self, key: &Key) -> Result<(Option<SourceRecord>, Option<Usage>)> {
107        let segment = path_segment(&key.path_segment());
108        let (resp, usage) = match self.send(self.works_url(&format!("/{segment}"))).await {
109            Ok(ok) => ok,
110            Err(Error::NotFound) => return Ok((None, None)),
111            Err(e) => return Err(e),
112        };
113        let value: serde_json::Value = serde_json::from_slice(&resp.body).map_err(Error::decode)?;
114        let raw = Raw {
115            media_type: "application/json".into(),
116            body: resp.text(),
117        };
118        let record = wire::to_record(&value, raw, resp.received_at).map_err(Error::decode)?;
119        Ok((Some(record), Some(usage)))
120    }
121
122    /// One list call; `params` are already-decoded query pairs.
123    async fn list(&self, params: &[(&str, String)]) -> Result<(Vec<Hit>, WireList, Usage)> {
124        let url = url::Url::parse_with_params(&self.works_url(""), params)
125            .map_err(|e| Error::BadRequest(format!("invalid OpenAlex base URL: {e}")))?;
126        let (resp, usage) = self.send(url.into()).await?;
127        let list: WireList = serde_json::from_slice(&resp.body).map_err(Error::decode)?;
128        let hits = list
129            .results
130            .iter()
131            .filter_map(|v| {
132                let score = v.get("relevance_score").and_then(serde_json::Value::as_f64);
133                let record = wire::to_record(v, Raw::json(v), resp.received_at);
134                or_skip(record).map(|r| Hit::new(r, score))
135            })
136            .collect();
137        Ok((hits, list, usage))
138    }
139}
140
141/// The record, or `None` with a warning when it doesn't decode: one bad
142/// record shouldn't cost the rest of the page.
143fn or_skip(record: serde_json::Result<SourceRecord>) -> Option<SourceRecord> {
144    record
145        .inspect_err(|e| {
146            tracing::warn!(
147                target: "aspero::decode",
148                provider = %ID, error = %e, "skipped a record that doesn't decode"
149            );
150        })
151        .ok()
152}
153
154/// How a requested ID is addressed at OpenAlex.
155struct Key {
156    /// Filter name, also the singleton path prefix.
157    filter: &'static str,
158    value: String,
159    /// The ID a returned record must carry to answer this request.
160    matches: WorkId,
161}
162
163impl Key {
164    fn new(id: &WorkId) -> Result<Key> {
165        Ok(match id {
166            WorkId::Doi(doi) => Key::doi(doi.clone()),
167            // arXiv works are indexed under their DataCite DOI.
168            WorkId::Arxiv(arxiv) => {
169                let doi = format!("{ARXIV_DOI_PREFIX}/arxiv.{}", arxiv.id());
170                Key::doi(doi.parse::<Doi>()?)
171            }
172            WorkId::Pmid(pmid) => Key {
173                filter: "pmid",
174                value: pmid.to_string(),
175                matches: id.clone(),
176            },
177            WorkId::OpenAlex(w) => Key {
178                filter: "openalex",
179                value: w.to_string(),
180                matches: id.clone(),
181            },
182            _ => {
183                return Err(Error::Unsupported(
184                    "OpenAlex lookup of this identifier kind",
185                ));
186            }
187        })
188    }
189
190    fn doi(doi: Doi) -> Key {
191        Key {
192            filter: "doi",
193            value: doi.as_str().to_owned(),
194            matches: WorkId::Doi(doi),
195        }
196    }
197
198    fn path_segment(&self) -> String {
199        match self.filter {
200            "openalex" => self.value.clone(),
201            filter => format!("{filter}:{}", self.value),
202        }
203    }
204
205    /// `,` and `|` are filter syntax, so such values can't be OR-ed.
206    fn batchable(&self) -> bool {
207        !self.value.contains([',', '|'])
208    }
209}
210
211fn type_filter(t: &WorkType) -> Result<String> {
212    let openalex = match t {
213        WorkType::JournalArticle => "article",
214        WorkType::ConferencePaper => "conference-paper",
215        WorkType::Preprint => "preprint",
216        WorkType::Book => "book",
217        WorkType::Chapter => "book-chapter",
218        WorkType::Thesis => "dissertation",
219        WorkType::Report => "report",
220        WorkType::Dataset => "dataset",
221        WorkType::Other(other) => other.as_str(),
222        _ => {
223            return Err(Error::BadRequest(format!(
224                "OpenAlex has no work type matching {:?}",
225                t.as_csl()
226            )));
227        }
228    };
229    Ok(format!("type:{openalex}"))
230}
231
232fn respond<T>(value: T, usage: Option<Usage>) -> Response<T> {
233    match usage {
234        Some(u) => Response::with_usage(value, u),
235        None => Response::new(value),
236    }
237}
238
239#[async_trait]
240impl Provider for OpenAlex {
241    fn id(&self) -> ProviderId {
242        ID
243    }
244
245    fn capabilities(&self) -> &Capabilities {
246        &self.capabilities
247    }
248
249    async fn lookup(&self, ids: &[WorkId], _fields: FieldSet) -> Result<Response<Vec<Lookup>>> {
250        let keys = ids.iter().map(Key::new).collect::<Result<Vec<_>>>()?;
251        let mut usage = None;
252
253        if let [key] = keys.as_slice() {
254            let (record, u) = self.lookup_one(key).await?;
255            let lookup = Lookup {
256                id: ids[0].clone(),
257                record,
258            };
259            return Ok(respond(vec![lookup], u));
260        }
261
262        // One filter call per kind, in order of first appearance.
263        let mut groups: Vec<(&str, Vec<&Key>)> = Vec::new();
264        let mut singles = Vec::new();
265        for key in &keys {
266            if !key.batchable() {
267                singles.push(key);
268            } else if let Some((_, group)) = groups.iter_mut().find(|(f, _)| *f == key.filter) {
269                group.push(key);
270            } else {
271                groups.push((key.filter, vec![key]));
272            }
273        }
274
275        let mut found = Vec::new();
276        for (filter, group) in groups {
277            for chunk in group.chunks(MAX_PAGE as usize) {
278                let values: Vec<&str> = chunk.iter().map(|k| k.value.as_str()).collect();
279                let params = [
280                    ("filter", format!("{filter}:{}", values.join("|"))),
281                    ("per-page", MAX_PAGE.to_string()),
282                ];
283                let (hits, _, u) = self.list(&params).await?;
284                found.extend(hits.into_iter().map(|h| h.record));
285                usage = Some(usage.map_or(u, |acc: Usage| acc.then(u)));
286            }
287        }
288        for key in singles {
289            let (record, u) = self.lookup_one(key).await?;
290            found.extend(record);
291            if let Some(u) = u {
292                usage = Some(usage.map_or(u, |acc: Usage| acc.then(u)));
293            }
294        }
295
296        let lookups = ids
297            .iter()
298            .zip(&keys)
299            .map(|(id, key)| Lookup {
300                id: id.clone(),
301                record: found.iter().find(|r| r.ids.contains(&key.matches)).cloned(),
302            })
303            .collect();
304        Ok(respond(lookups, usage))
305    }
306
307    async fn search(&self, query: &SearchQuery) -> Result<Response<Page>> {
308        if query.mode != SearchMode::Keywords {
309            return Err(Error::Unsupported("OpenAlex has no citation matcher"));
310        }
311        let mut filters = Vec::new();
312        if let Some(years) = query.years {
313            if let Some(from) = years.from {
314                filters.push(format!("from_publication_date:{from:04}-01-01"));
315            }
316            if let Some(to) = years.to {
317                filters.push(format!("to_publication_date:{to:04}-12-31"));
318            }
319        }
320        if let Some(t) = &query.work_type {
321            filters.push(type_filter(t)?);
322        }
323        if let Some(oa) = query.open_access {
324            filters.push(format!("is_oa:{oa}"));
325        }
326        if let Some(place) = &query.place {
327            // Filter values can't hold commas or pipes, which separate them.
328            let clean = |s: &str| s.replace([',', '|'], "");
329            filters.push(format!("biblio.volume:{}", clean(&place.volume)));
330            filters.push(format!("biblio.first_page:{}", clean(&place.first_page)));
331        }
332
333        let mut params: Vec<(&str, String)> = Vec::new();
334        for (key, value) in &query.extra {
335            match key.as_str() {
336                "filter" => filters.push(value.clone()),
337                "sort" => params.push(("sort", value.clone())),
338                other => {
339                    return Err(Error::BadRequest(format!(
340                        "unknown OpenAlex parameter {other:?} (supported: filter, sort)"
341                    )));
342                }
343            }
344        }
345        if !query.text.trim().is_empty() {
346            params.push(("search", query.text.trim().to_owned()));
347        }
348        if !filters.is_empty() {
349            params.push(("filter", filters.join(",")));
350        }
351        params.extend(page_params(&query.page));
352
353        let (hits, list, usage) = self.list(&params).await?;
354        let page = Page::new(hits, list.meta.count, list.meta.next_cursor.map(Cursor));
355        Ok(Response::with_usage(page, usage))
356    }
357}
358
359/// Cursor paging (needs a key; unlimited depth, unlike `page=`).
360fn page_params(page: &PageReq) -> [(&'static str, String); 2] {
361    let cursor = page.cursor.as_ref().map_or("*", |c| c.0.as_str());
362    [
363        ("per-page", page.limit.clamp(1, MAX_PAGE).to_string()),
364        ("cursor", cursor.to_owned()),
365    ]
366}
aspero_openalex/lib.rs

aspero_openalex/
lib.rs
