base

zocrates.discovery.providers.base

Base interfaces and utilities for scholarly search providers.

DEFAULT_POLITE_EMAIL module-attribute

DEFAULT_POLITE_EMAIL = 'zocrates-dev@proton.me'

DEFAULT_USER_AGENT module-attribute

DEFAULT_USER_AGENT = f'Zocrates/0.2.0 (mailto:{DEFAULT_POLITE_EMAIL})'

CandidatePaper dataclass

CandidatePaper(candidate_id, title, authors=list(), year=None, venue=None, doi=None, arxiv_id=None, openalex_id='', abstract=None, tldr=None, is_oa=False, oa_pdf_url=None, oa_pdf_urls=list(), citation_count=0, primary_topic=None, keywords=list(), in_library=False)

Represents a discovered scholarly paper candidate.

from_dict classmethod

from_dict(data)

Deserialize CandidatePaper from a dictionary.

Source code in zocrates/discovery/models.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "CandidatePaper":
    """Deserialize CandidatePaper from a dictionary."""
    citation_count_val = data.get("citation_count", 0)
    try:
        citation_count = int(citation_count_val) if citation_count_val is not None else 0
    except (ValueError, TypeError):
        citation_count = 0

    year_val = data.get("year")
    year: int | None = None
    if year_val is not None:
        try:
            year = int(year_val)
        except (ValueError, TypeError):
            year = None

    raw_authors = data.get("authors")
    if isinstance(raw_authors, list):
        authors = [str(a).strip() for a in raw_authors if a and str(a).strip()]
    elif isinstance(raw_authors, str) and raw_authors.strip():
        authors = [raw_authors.strip()]
    else:
        authors = []

    raw_keywords = data.get("keywords")
    if isinstance(raw_keywords, list):
        keywords = [str(k).strip() for k in raw_keywords if k and str(k).strip()]
    elif isinstance(raw_keywords, str) and raw_keywords.strip():
        keywords = [raw_keywords.strip()]
    else:
        keywords = []

    def _clean_opt_str(val: Any) -> str | None:
        """Clean and trim an optional string or scalar value.

        Args:
            val: Input value to convert to a stripped string.

        Returns:
            Stripped string, or None if empty or invalid.
        """
        if val is not None and isinstance(val, (str, int, float)):
            s = str(val).strip()
            return s if s else None
        return None

    raw_urls = data.get("oa_pdf_urls")
    oa_pdf_urls: list[str] = []
    if isinstance(raw_urls, list):
        oa_pdf_urls = [str(u).strip() for u in raw_urls if u and str(u).strip()]
    primary_pdf = _clean_opt_str(data.get("oa_pdf_url"))
    if primary_pdf and primary_pdf not in oa_pdf_urls:
        oa_pdf_urls.insert(0, primary_pdf)

    return cls(
        candidate_id=str(data.get("candidate_id") or ""),
        title=str(data.get("title") or "").strip(),
        authors=authors,
        year=year,
        venue=_clean_opt_str(data.get("venue")),
        doi=_clean_opt_str(data.get("doi")),
        arxiv_id=_clean_opt_str(data.get("arxiv_id")),
        openalex_id=str(data.get("openalex_id") or ""),
        abstract=_clean_opt_str(data.get("abstract")),
        tldr=_clean_opt_str(data.get("tldr")),
        is_oa=bool(data.get("is_oa", False)),
        oa_pdf_url=primary_pdf,
        oa_pdf_urls=oa_pdf_urls,
        citation_count=citation_count,
        primary_topic=_clean_opt_str(data.get("primary_topic")),
        keywords=keywords,
        in_library=bool(data.get("in_library", False)),
    )

to_dict

to_dict()

Serialize CandidatePaper to a plain dictionary.

Source code in zocrates/discovery/models.py

def to_dict(self) -> dict[str, Any]:
    """Serialize CandidatePaper to a plain dictionary."""
    return {
        "candidate_id": self.candidate_id,
        "title": self.title,
        "authors": list(self.authors),
        "year": self.year,
        "venue": self.venue,
        "doi": self.doi,
        "arxiv_id": self.arxiv_id,
        "openalex_id": self.openalex_id,
        "abstract": self.abstract,
        "tldr": self.tldr,
        "is_oa": self.is_oa,
        "oa_pdf_url": self.oa_pdf_url,
        "oa_pdf_urls": list(self.oa_pdf_urls),
        "citation_count": self.citation_count,
        "primary_topic": self.primary_topic,
        "keywords": list(self.keywords),
        "in_library": self.in_library,
    }

SearchProvider

Bases: ABC

Abstract base class for scholarly literature search providers.

resolve_identifier abstractmethod async

resolve_identifier(identifier)

Resolve a single scholarly paper by DOI, arXiv ID, or provider-specific ID.

Parameters:

Name	Type	Description	Default
identifier	str	DOI, arXiv ID, or provider ID string.	required

Returns:

Type	Description
CandidatePaper | None	CandidatePaper if resolved, or None if not found.

Source code in zocrates/discovery/providers/base.py

@abstractmethod
async def resolve_identifier(self, identifier: str) -> CandidatePaper | None:
    """
    Resolve a single scholarly paper by DOI, arXiv ID, or provider-specific ID.

    Args:
        identifier: DOI, arXiv ID, or provider ID string.

    Returns:
        CandidatePaper if resolved, or None if not found.
    """
    ...

search abstractmethod async

search(query, limit=10, year_from=None, year_to=None, is_oa_only=False)

Search for scholarly literature matching the query and filters.

Parameters:

Name	Type	Description	Default
query	str	Free-text search terms or conceptual query.	required
limit	int	Maximum number of candidate papers to return.	10
year_from	int | None	Optional minimum publication year (inclusive).	None
year_to	int | None	Optional maximum publication year (inclusive).	None
is_oa_only	bool	When True, restrict results to open-access works.	False

Returns:

Type	Description
list[CandidatePaper]	List of CandidatePaper instances.

Source code in zocrates/discovery/providers/base.py

@abstractmethod
async def search(
    self,
    query: str,
    limit: int = 10,
    year_from: int | None = None,
    year_to: int | None = None,
    is_oa_only: bool = False,
) -> list[CandidatePaper]:
    """
    Search for scholarly literature matching the query and filters.

    Args:
        query: Free-text search terms or conceptual query.
        limit: Maximum number of candidate papers to return.
        year_from: Optional minimum publication year (inclusive).
        year_to: Optional maximum publication year (inclusive).
        is_oa_only: When True, restrict results to open-access works.

    Returns:
        List of CandidatePaper instances.
    """
    ...

TokenBucketRateLimiter

TokenBucketRateLimiter(rate=10.0, capacity=10.0)

Token-bucket rate limiter for respectful, throttled API consumption. Thread/asyncio safe.

Parameters:

Name	Type	Description	Default
rate	float	Refill rate in tokens per second.	10.0
capacity	float	Maximum burst capacity in tokens.	10.0

Source code in zocrates/discovery/providers/base.py

def __init__(self, rate: float = 10.0, capacity: float = 10.0) -> None:
    """
    Args:
        rate: Refill rate in tokens per second.
        capacity: Maximum burst capacity in tokens.
    """
    self.rate = max(0.1, float(rate))
    self.capacity = max(1.0, float(capacity))
    self.tokens = self.capacity
    self.last_update = time.monotonic()
    self._lock = asyncio.Lock()

acquire async

acquire(tokens=1.0)

Acquire tokens from the bucket, sleeping if necessary until replenished.

Source code in zocrates/discovery/providers/base.py

async def acquire(self, tokens: float = 1.0) -> None:
    """Acquire `tokens` from the bucket, sleeping if necessary until replenished."""
    async with self._lock:
        now = time.monotonic()
        elapsed = now - self.last_update
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_update = now

        if self.tokens < tokens:
            needed = tokens - self.tokens
            wait_time = needed / self.rate
            await asyncio.sleep(wait_time)
            self.tokens = 0.0
            self.last_update = time.monotonic()
        else:
            self.tokens -= tokens

make_user_agent

make_user_agent(email=DEFAULT_POLITE_EMAIL)

Build a respectful User-Agent header string adhering to API etiquette.

Source code in zocrates/discovery/providers/base.py

def make_user_agent(email: str = DEFAULT_POLITE_EMAIL) -> str:
    """Build a respectful User-Agent header string adhering to API etiquette."""
    polite_email = email.strip() if email and email.strip() else DEFAULT_POLITE_EMAIL
    return f"Zocrates/0.2.0 (mailto:{polite_email})"