openalexpy

A modern, async-first Python library for the OpenAlex API.

Acknowledgements

This library was inspired by PyAlex by Jonathan de Bruin. PyAlex pioneered the Python interface to OpenAlex and proved the value of a fluent, pipe-able query builder for scholarly data. openalexpy builds on those foundations with native async support, full type safety via Pydantic, and cost-aware API usage.

Features

• Async-first with sync wrappers — httpx-based, asyncio-native
• Fully typed — Pydantic v2 models for all entities with IDE autocomplete
• Cost-aware — Parses X-RateLimit-* headers, distinguishes credit exhaustion from temporary rate limits, exposes cost_usd on every response
• Correct API key handling — Uses api_key query parameter as documented by OpenAlex (not undocumented headers)
• Immutable query builder — Each method returns a new instance, no shared mutable state bugs
• Two-step content download — Properly handles PDF/TEI redirects preserving rate-limit headers
• Semantic search — First-class support for search.semantic with automatic 1 req/s rate limiting

Installation

uv add openalex-py

or

pip install openalex-py

Requires Python 3.10+.

Quick Start

Configure your API key

import openalexpy

openalexpy.config.api_key = "YOUR_API_KEY"

Or set the OPENALEX_API_KEY environment variable.

Async usage (recommended)

import asyncio
import openalexpy

async def main():
    # Get a single work
    work = await openalexpy.Works().get_by_id("W2741809807")
    print(work.title)
    print(work.abstract)

    # Filter works
    results = await openalexpy.Works().filter(
        publication_year=2024, is_oa=True
    ).sort(cited_by_count="desc").get(per_page=10)

    for w in results:
        print(f"{w.title} ({w.cited_by_count} citations)")

    # Semantic search
    similar = await openalexpy.Works().similar(
        "machine learning for drug discovery"
    ).filter(publication_year=">2022").get(per_page=50)

    # Paginate through all results
    pager = openalexpy.Works().filter(
        author={"id": "A5023888391"}
    ).paginate(per_page=100)

    async for page in pager:
        for work in page:
            print(work.id)

asyncio.run(main())

Sync usage

import openalexpy

openalexpy.config.api_key = "YOUR_API_KEY"

# Get a single work
work = openalexpy.WorksSync().get_by_id("W2741809807")

# Filter and get
results = openalexpy.WorksSync().filter(
    publication_year=2024
).sort(cited_by_count="desc").get(per_page=10)

# Paginate
pager = openalexpy.WorksSync().filter(
    author={"id": "A5023888391"}
).paginate(per_page=100)

for page in pager:
    for work in page:
        print(work.id)

Supported Entities

Query Building

All query methods return a new instance — the original query is never mutated.

base = openalexpy.Works()

# Chain filters
q = base.filter(publication_year=2024).filter(is_oa=True)

# base is unchanged
assert "filter" not in base.params

Filter operators

# AND (default)
Works().filter(institutions={"country_code": ["fr", "gb"]})

# OR
Works().filter_or(doi=["10.1234/a", "10.1234/b"])

# NOT
Institutions().filter_not(country_code="us")

# Greater than / Less than
Works().filter_gt(cited_by_count=100)
Works().filter_lt(publication_year=2020)

Semantic search

# Basic semantic search (capped at 50 results per request)
results = await Works().similar("climate change impacts").get()

# Combined with filters
results = await (
    Works()
    .similar("quantum computing applications")
    .filter(publication_year=">2022", is_oa=True)
    .get(per_page=50)
)

Pagination

# Cursor pagination (default, for deep pagination)
pager = Works().filter(publication_year=2024).paginate(per_page=100, n_max=5000)

async for page in pager:
    print(len(page))

# Page-based pagination (limited to 10,000 results)
pager = Works().search("dna").paginate(method="page", per_page=100)

Content download (PDF/TEI)

work = await Works().get_by_id("W4412002745")

# Get PDF bytes
pdf_bytes = await work._pdf.get()

# Download to file
await work._pdf.download("paper.pdf")

# TEI XML
tei_bytes = await work._tei.get()

Cost Monitoring

Every API response includes cost information:

response = await Works().filter(publication_year=2024).get(
    per_page=100, return_meta=True
)
print(f"Cost: ${response.meta.cost_usd}")
print(f"Total results: {response.meta.count}")

# Check rate limit status
client = openalexpy.client.AsyncOpenAlexClient()
status = await client.get_rate_limit_status()
print(status)

Error Handling

from openalexpy import QueryError, RateLimitError, CreditsExhaustedError

try:
    results = await Works().filter(bad_filter=True).get()
except QueryError as e:
    print(f"Bad query: {e}")
except CreditsExhaustedError as e:
    print(f"Daily credits exhausted. Resets at: {e.reset_at}")
except RateLimitError as e:
    print(f"Temporarily rate limited. Retry after: {e.retry_after}s")

License

MIT