API Reference

This section provides a basic API reference generated from the docstrings in the AIREloom library using mkdocs-python-extractor.

AireloomSession

The main session class for interacting with AIREloom.

High-level session manager for interacting with OpenAIRE APIs.

This class acts as the primary entry point for users of the aireloom library. It provides convenient access to various OpenAIRE resource clients (e.g., for research products, projects) through an underlying AireloomClient instance.

The session handles the lifecycle of the AireloomClient, including its creation with appropriate settings (like timeouts and authentication) and its proper closure when the session is no longer needed. It supports asynchronous context management (async with).

Example:

async with AireloomSession(timeout=60) as session:
    product = await session.research_products.get("some_id")
    # ... further API calls

Attributes:

Name	Type	Description
research_products	ResearchProductsClient	Client for research product APIs.
organizations	OrganizationsClient	Client for organization APIs.
projects	ProjectsClient	Client for project APIs.
data_sources	DataSourcesClient	Client for data source APIs.
scholix	ScholixClient	Client for Scholix (scholarly link) APIs.
_api_client	AireloomClient	The underlying client instance.

Source code in src/aireloom/session.py

class AireloomSession:
    """High-level session manager for interacting with OpenAIRE APIs.

    This class acts as the primary entry point for users of the `aireloom` library.
    It provides convenient access to various OpenAIRE resource clients (e.g., for
    research products, projects) through an underlying `AireloomClient` instance.

    The session handles the lifecycle of the `AireloomClient`, including its
    creation with appropriate settings (like timeouts and authentication) and
    its proper closure when the session is no longer needed. It supports
    asynchronous context management (`async with`).

    Example:
    ```python
    async with AireloomSession(timeout=60) as session:
        product = await session.research_products.get("some_id")
        # ... further API calls
    ```

    Attributes:
        research_products (ResearchProductsClient): Client for research product APIs.
        organizations (OrganizationsClient): Client for organization APIs.
        projects (ProjectsClient): Client for project APIs.
        data_sources (DataSourcesClient): Client for data source APIs.
        scholix (ScholixClient): Client for Scholix (scholarly link) APIs.
        _api_client (AireloomClient): The underlying client instance.
    """

    def __init__(
        self,
        auth_strategy: AuthStrategy | None = None,
        timeout: int | None = None,
        api_base_url: str | None = None,
        scholix_base_url: str | None = None,
    ):
        """Initializes the Aireloom session and its underlying `AireloomClient`.

        The session allows for overriding certain configurations like request timeout
        and API base URLs. Authentication strategy can also be provided directly.
        If not provided, the `AireloomClient` will attempt to determine it based
        on its own settings (loaded from environment or .env files).

        Args:
            auth_strategy: An optional `AuthStrategy` instance to be used for
                all requests made through this session. If `None`, the
                `AireloomClient` will determine authentication based on its settings.
            timeout: An optional integer to override the default request timeout
                (in seconds) for all HTTP requests made during this session.
                If `None`, the timeout from global or client-specific settings is used.
            api_base_url: An optional string to override the default base URL for the
                OpenAIRE Graph API.
            scholix_base_url: An optional string to override the default base URL for
                the OpenAIRE Scholix API.
        """
        _api_base_url = api_base_url or OPENAIRE_GRAPH_API_BASE_URL
        _scholix_base_url = scholix_base_url or OPENAIRE_SCHOLIX_API_BASE_URL

        current_settings = get_settings()
        session_specific_settings: ApiSettings
        if timeout is not None:
            logger.debug(f"Overriding request timeout for this session to: {timeout}s")
            session_specific_settings = current_settings.model_copy(
                update={"request_timeout": timeout}
            )
        else:
            session_specific_settings = current_settings

        # Pass the original auth_strategy (which can be None) to the client.
        # The client will then decide its auth based on this and its settings.
        logger.debug(
            f"AireloomSession: Initializing AireloomClient with auth_strategy param: {type(auth_strategy)}"
        )
        self._api_client = AireloomClient(
            settings=session_specific_settings,
            auth_strategy=auth_strategy,  # Pass the original auth_strategy parameter
            base_url=_api_base_url,  # Pass Graph API base URL
            scholix_base_url=_scholix_base_url,  # Pass Scholix base URL
        )
        logger.info(f"AireloomSession initialized for API: {_api_base_url}")
        logger.info(f"Scholexplorer base URL configured for: {_scholix_base_url}")

    @property
    def research_products(self) -> ResearchProductsClient:
        """Access the ResearchProductsClient."""
        return self._api_client.research_products

    @property
    def organizations(self) -> OrganizationsClient:
        """Access the OrganizationsClient."""
        return self._api_client.organizations

    @property
    def projects(self) -> ProjectsClient:
        """Access the ProjectsClient."""
        return self._api_client.projects

    @property
    def data_sources(self) -> DataSourcesClient:
        """Access the DataSourcesClient."""
        return self._api_client.data_sources

    @property
    def scholix(self) -> ScholixClient:
        """Access the ScholixClient."""
        return self._api_client.scholix

    async def close(self) -> None:
        """Closes the underlying HTTP client session."""
        await self._api_client.aclose()

    async def __aenter__(self) -> "AireloomSession":
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        await self.close()

data_sources property

Access the DataSourcesClient.

organizations property

Access the OrganizationsClient.

projects property

Access the ProjectsClient.

research_products property

Access the ResearchProductsClient.

scholix property

Access the ScholixClient.

__init__(auth_strategy=None, timeout=None, api_base_url=None, scholix_base_url=None)

Initializes the Aireloom session and its underlying AireloomClient.

The session allows for overriding certain configurations like request timeout and API base URLs. Authentication strategy can also be provided directly. If not provided, the AireloomClient will attempt to determine it based on its own settings (loaded from environment or .env files).

Parameters:

Name	Type	Description	Default
auth_strategy	AuthStrategy | None	An optional AuthStrategy instance to be used for all requests made through this session. If None, the AireloomClient will determine authentication based on its settings.	None
timeout	int | None	An optional integer to override the default request timeout (in seconds) for all HTTP requests made during this session. If None, the timeout from global or client-specific settings is used.	None
api_base_url	str | None	An optional string to override the default base URL for the OpenAIRE Graph API.	None
scholix_base_url	str | None	An optional string to override the default base URL for the OpenAIRE Scholix API.	None

Source code in src/aireloom/session.py

def __init__(
    self,
    auth_strategy: AuthStrategy | None = None,
    timeout: int | None = None,
    api_base_url: str | None = None,
    scholix_base_url: str | None = None,
):
    """Initializes the Aireloom session and its underlying `AireloomClient`.

    The session allows for overriding certain configurations like request timeout
    and API base URLs. Authentication strategy can also be provided directly.
    If not provided, the `AireloomClient` will attempt to determine it based
    on its own settings (loaded from environment or .env files).

    Args:
        auth_strategy: An optional `AuthStrategy` instance to be used for
            all requests made through this session. If `None`, the
            `AireloomClient` will determine authentication based on its settings.
        timeout: An optional integer to override the default request timeout
            (in seconds) for all HTTP requests made during this session.
            If `None`, the timeout from global or client-specific settings is used.
        api_base_url: An optional string to override the default base URL for the
            OpenAIRE Graph API.
        scholix_base_url: An optional string to override the default base URL for
            the OpenAIRE Scholix API.
    """
    _api_base_url = api_base_url or OPENAIRE_GRAPH_API_BASE_URL
    _scholix_base_url = scholix_base_url or OPENAIRE_SCHOLIX_API_BASE_URL

    current_settings = get_settings()
    session_specific_settings: ApiSettings
    if timeout is not None:
        logger.debug(f"Overriding request timeout for this session to: {timeout}s")
        session_specific_settings = current_settings.model_copy(
            update={"request_timeout": timeout}
        )
    else:
        session_specific_settings = current_settings

    # Pass the original auth_strategy (which can be None) to the client.
    # The client will then decide its auth based on this and its settings.
    logger.debug(
        f"AireloomSession: Initializing AireloomClient with auth_strategy param: {type(auth_strategy)}"
    )
    self._api_client = AireloomClient(
        settings=session_specific_settings,
        auth_strategy=auth_strategy,  # Pass the original auth_strategy parameter
        base_url=_api_base_url,  # Pass Graph API base URL
        scholix_base_url=_scholix_base_url,  # Pass Scholix base URL
    )
    logger.info(f"AireloomSession initialized for API: {_api_base_url}")
    logger.info(f"Scholexplorer base URL configured for: {_scholix_base_url}")

close() async

Closes the underlying HTTP client session.

Source code in src/aireloom/session.py

async def close(self) -> None:
    """Closes the underlying HTTP client session."""
    await self._api_client.aclose()

Resource Clients

Clients for specific OpenAIRE API endpoints.

ResearchProductsClient

For accessing research products (publications, datasets, software, etc.).

Bases: GettableMixin, SearchableMixin, CursorIterableMixin, BaseResourceClient

Client for the OpenAIRE Research Products API endpoint.

This client provides standardized methods (get, search, iterate) for accessing research product data, by inheriting from bibliofabric mixins. It is configured with the specific API path and Pydantic models relevant to OpenAIRE research products.

Attributes:

Name	Type	Description
_entity_path	str	The API path for research products.
_entity_model	type[ResearchProduct]	Pydantic model for a single research product.
_search_response_model	type[ResearchProductResponse]	Pydantic model for the search response envelope.
_valid_sort_fields	set[str]	A set of field names that are valid for sorting results from this endpoint.

Source code in src/aireloom/resources/research_products_client.py

class ResearchProductsClient(
    GettableMixin, SearchableMixin, CursorIterableMixin, BaseResourceClient
):
    """Client for the OpenAIRE Research Products API endpoint.

    This client provides standardized methods (`get`, `search`, `iterate`) for
    accessing research product data, by inheriting from `bibliofabric` mixins.
    It is configured with the specific API path and Pydantic models relevant
    to OpenAIRE research products.

    Attributes:
        _entity_path (str): The API path for research products.
        _entity_model (type[ResearchProduct]): Pydantic model for a single research product.
        _search_response_model (type[ResearchProductResponse]): Pydantic model for the
                                                                search response envelope.
        _valid_sort_fields (set[str]): A set of field names that are valid for sorting
                                       results from this endpoint.
    """

    _entity_path: str = RESEARCH_PRODUCTS
    _entity_model: type[ResearchProduct] = ResearchProduct
    _search_response_model: type[ResearchProductResponse] = ResearchProductResponse
    _valid_sort_fields = {
        "bestaccessright",
        "publicationdate",
        "relevance",
        "title",
    }

    def __init__(self, api_client: "AireloomClient"):
        """Initializes the ResearchProductsClient.

        Args:
            api_client: An instance of AireloomClient.
        """
        super().__init__(api_client)
        logger.debug(
            f"ResearchProductsClient initialized for path: {self._entity_path}"
        )

__init__(api_client)

Initializes the ResearchProductsClient.

Parameters:

Name	Type	Description	Default
api_client	AireloomClient	An instance of AireloomClient.	required

Source code in src/aireloom/resources/research_products_client.py

def __init__(self, api_client: "AireloomClient"):
    """Initializes the ResearchProductsClient.

    Args:
        api_client: An instance of AireloomClient.
    """
    super().__init__(api_client)
    logger.debug(
        f"ResearchProductsClient initialized for path: {self._entity_path}"
    )

OrganizationsClient

For accessing organization data.

Bases: BaseResourceClient

Client for the OpenAIRE Organizations API endpoint.

This client provides methods to retrieve individual organizations (get), search for organizations based on filters (search), and iterate through all organizations (iterate). It currently uses custom implementations for these methods rather than directly using the generic mixins from bibliofabric.resources.

Attributes:

Name	Type	Description
_entity_path	str	The API path for organizations.
_entity_model	type[Organization]	Pydantic model for a single organization.
_response_model	type[OrganizationResponse]	Pydantic model for the search response envelope.
_endpoint_def	dict	Configuration for this endpoint from ENDPOINT_DEFINITIONS.
_valid_sort_fields	set[str]	Valid sort fields for this endpoint.

Source code in src/aireloom/resources/organizations_client.py

class OrganizationsClient(BaseResourceClient):
    """Client for the OpenAIRE Organizations API endpoint.

    This client provides methods to retrieve individual organizations (`get`),
    search for organizations based on filters (`search`), and iterate through
    all organizations (`iterate`). It currently uses custom implementations for these
    methods rather than directly using the generic mixins from `bibliofabric.resources`.

    Attributes:
        _entity_path (str): The API path for organizations.
        _entity_model (type[Organization]): Pydantic model for a single organization.
        _response_model (type[OrganizationResponse]): Pydantic model for the
                                                       search response envelope.
        _endpoint_def (dict): Configuration for this endpoint from `ENDPOINT_DEFINITIONS`.
        _valid_sort_fields (set[str]): Valid sort fields for this endpoint.
    """

    _entity_path: str = ORGANIZATIONS
    _entity_model: type[Organization] = Organization
    _response_model: type[OrganizationResponse] = OrganizationResponse

    def __init__(self, api_client: "AireloomClient"):
        """Initializes the OrganizationsClient.

        Args:
            api_client: An instance of AireloomClient.
        """
        super().__init__(api_client)
        if self._entity_path not in ENDPOINT_DEFINITIONS:
            raise ValueError(
                f"Missing endpoint definition for entity path: {self._entity_path}"
            )
        self._endpoint_def = ENDPOINT_DEFINITIONS[self._entity_path]
        self._valid_sort_fields = self._endpoint_def.get(
            "sort", {}
        ).keys()  # Get sort fields
        logger.debug(f"OrganizationsClient initialized for path: {self._entity_path}")

    # _validate_filters and _validate_and_convert_filter_value are removed as Pydantic handles this.

    def _validate_sort(self, sort_by: str | None) -> None:
        """Validates the sort field against endpoint definitions."""
        if not sort_by:
            return

        if not self._valid_sort_fields:
            logger.warning(
                f"Sort field '{sort_by}' provided for {self._entity_path}, "
                "but no sort fields are defined. Ignoring sort."
            )
            return
        sort_field_name = sort_by.split()[0]
        if sort_field_name not in self._valid_sort_fields:
            raise ValidationError(
                f"Invalid sort field for {self._entity_path}: '{sort_field_name}'. "
                f"Valid fields: {list(self._valid_sort_fields)}"
            )

    def _build_params(
        self,
        page: int | None,
        page_size: int,
        sort_by: str | None,
        filters: dict[str, Any] | None,  # Changed to Optional[dict]
        *,
        is_iteration: bool = False,
    ) -> dict[str, Any]:
        """Builds the query parameter dictionary."""
        params: dict[str, Any] = {"pageSize": page_size}
        if is_iteration:
            params["cursor"] = "*"
        elif page is not None:
            params["page"] = page
        if sort_by:
            params["sortBy"] = sort_by
        if filters:
            params.update(filters)
        return {k: v for k, v in params.items() if v is not None}

    async def _fetch_single_entity_impl(self, entity_id: str) -> Organization:
        """Generic method to fetch a single entity by ID using search-by-ID."""
        try:
            # Use search with ID parameter instead of direct GET
            params = {"id": entity_id, "pageSize": 1}
            response = await self._api_client.request(
                "GET", self._entity_path, params=params, data=None, json_data=None
            )
            data = response.json()

            # Parse the search response
            search_response = self._response_model.model_validate(data)

            if not search_response.results:
                raise BibliofabricError(
                    f"{self._entity_model.__name__} with ID '{entity_id}' not found."
                )

            # Return the first (and should be only) result
            return search_response.results[0]

        except httpx.HTTPStatusError as e:
            logger.error(
                f"HTTPStatusError for {self._entity_model.__name__} ID '{entity_id}': {e.response.status_code}"
            )
            raise BibliofabricError(
                f"API error fetching {self._entity_model.__name__} {entity_id}: "
                f"Status {e.response.status_code}"
            ) from e
        except Exception as e:
            if isinstance(e, BibliofabricError):
                raise
            logger.exception(
                f"Failed to fetch {self._entity_model.__name__} {entity_id} from {self._entity_path}"
            )
            raise BibliofabricError(
                f"Unexpected error fetching {self._entity_model.__name__} {entity_id}: {e}"
            ) from e

    async def _search_entities_impl(
        self, params: dict[str, Any]
    ) -> OrganizationResponse:
        """Generic method to search for entities."""
        try:
            response = await self._api_client.request(
                "GET", self._entity_path, params=params, data=None, json_data=None
            )
            return self._response_model.model_validate(response.json())
        except Exception as e:
            if isinstance(e, BibliofabricError | ValidationError):
                raise
            logger.exception(
                f"Failed to search {self._entity_path} with params {params}"
            )
            raise BibliofabricError(
                f"Unexpected error searching {self._entity_path}: {e}"
            ) from e

    async def _iterate_entities_impl(
        self, params: dict[str, Any]
    ) -> AsyncIterator[Organization]:
        """Generic method to iterate through all results using cursor pagination."""
        current_params = params.copy()
        while True:
            try:
                logger.debug(
                    f"Iterating {self._entity_path} with params: {current_params}"
                )
                response = await self._api_client.request(
                    "GET",
                    self._entity_path,
                    params=current_params,
                    data=None,
                    json_data=None,
                )
                data = response.json()
                api_response = ApiResponse[self._entity_model].model_validate(data)
                if not api_response.results:
                    logger.debug(
                        f"No more results for {self._entity_path}, stopping iteration."
                    )
                    break
                for result in api_response.results:
                    yield result
                next_cursor = api_response.header.nextCursor
                if not next_cursor:
                    logger.debug(
                        f"No nextCursor for {self._entity_path}, stopping iteration."
                    )
                    break
                current_params["cursor"] = next_cursor
                current_params.pop("page", None)
            except Exception as e:
                if isinstance(e, BibliofabricError | ValidationError):
                    raise
                logger.exception(
                    f"Failed during iteration of {self._entity_path} with params {current_params}"
                )
                raise BibliofabricError(
                    f"Unexpected error during iteration of {self._entity_path}: {e}"
                ) from e

    async def get(self, org_id: str) -> Organization:
        """Retrieves a single Organization by its ID.

        Args:
            org_id: The ID of the organization.

        Returns:
            An Organization object.
        """
        logger.info(f"Fetching organization with ID: {org_id}")
        return await self._fetch_single_entity_impl(org_id)

    async def search(
        self,
        page: int = 1,
        page_size: int = DEFAULT_PAGE_SIZE,
        sort_by: str | None = None,
        filters: OrganizationsFilters | None = None,  # Changed to Pydantic model
    ) -> OrganizationResponse:
        """Searches for Organizations.

        Args:
            page: Page number (1-indexed).
            page_size: Number of results per page.
            sort_by: Field to sort by.
            filters: An instance of OrganizationsFilters with filter criteria.

        Returns:
            An OrganizationResponse object.
        """
        filter_dict = (
            filters.model_dump(exclude_none=True, by_alias=True) if filters else {}
        )
        logger.info(
            f"Searching organizations: page={page}, size={page_size}, sort='{sort_by}', "
            f"filters={filter_dict}"
        )
        # self._validate_filters is removed
        self._validate_sort(sort_by)
        params = self._build_params(
            page=page, page_size=page_size, sort_by=sort_by, filters=filter_dict
        )
        return await self._search_entities_impl(params)

    async def iterate(
        self,
        page_size: int = 100,
        sort_by: str | None = None,
        filters: OrganizationsFilters | None = None,  # Changed to Pydantic model
    ) -> AsyncIterator[Organization]:
        """Iterates through all Organization results.

        Args:
            page_size: Number of results per page during iteration.
            sort_by: Field to sort by.
            filters: An instance of OrganizationsFilters with filter criteria.

        Yields:
            Organization objects.
        """
        filter_dict = (
            filters.model_dump(exclude_none=True, by_alias=True) if filters else {}
        )
        logger.info(
            f"Iterating organizations: size={page_size}, sort='{sort_by}', "
            f"filters={filter_dict}"
        )
        # self._validate_filters is removed
        self._validate_sort(sort_by)
        params = self._build_params(
            page=None,
            page_size=page_size,
            sort_by=sort_by,
            filters=filter_dict,
            is_iteration=True,
        )
        async for item in self._iterate_entities_impl(params):
            yield item

__init__(api_client)

Initializes the OrganizationsClient.

Parameters:

Name	Type	Description	Default
api_client	AireloomClient	An instance of AireloomClient.	required

Source code in src/aireloom/resources/organizations_client.py

def __init__(self, api_client: "AireloomClient"):
    """Initializes the OrganizationsClient.

    Args:
        api_client: An instance of AireloomClient.
    """
    super().__init__(api_client)
    if self._entity_path not in ENDPOINT_DEFINITIONS:
        raise ValueError(
            f"Missing endpoint definition for entity path: {self._entity_path}"
        )
    self._endpoint_def = ENDPOINT_DEFINITIONS[self._entity_path]
    self._valid_sort_fields = self._endpoint_def.get(
        "sort", {}
    ).keys()  # Get sort fields
    logger.debug(f"OrganizationsClient initialized for path: {self._entity_path}")

get(org_id) async

Retrieves a single Organization by its ID.

Parameters:

Name	Type	Description	Default
org_id	str	The ID of the organization.	required

Returns:

Type	Description
Organization	An Organization object.

Source code in src/aireloom/resources/organizations_client.py

async def get(self, org_id: str) -> Organization:
    """Retrieves a single Organization by its ID.

    Args:
        org_id: The ID of the organization.

    Returns:
        An Organization object.
    """
    logger.info(f"Fetching organization with ID: {org_id}")
    return await self._fetch_single_entity_impl(org_id)

iterate(page_size=100, sort_by=None, filters=None) async

Iterates through all Organization results.

Parameters:

Name	Type	Description	Default
page_size	int	Number of results per page during iteration.	100
sort_by	str | None	Field to sort by.	None
filters	OrganizationsFilters | None	An instance of OrganizationsFilters with filter criteria.	None

Yields:

Type	Description
AsyncIterator[Organization]	Organization objects.

Source code in src/aireloom/resources/organizations_client.py

async def iterate(
    self,
    page_size: int = 100,
    sort_by: str | None = None,
    filters: OrganizationsFilters | None = None,  # Changed to Pydantic model
) -> AsyncIterator[Organization]:
    """Iterates through all Organization results.

    Args:
        page_size: Number of results per page during iteration.
        sort_by: Field to sort by.
        filters: An instance of OrganizationsFilters with filter criteria.

    Yields:
        Organization objects.
    """
    filter_dict = (
        filters.model_dump(exclude_none=True, by_alias=True) if filters else {}
    )
    logger.info(
        f"Iterating organizations: size={page_size}, sort='{sort_by}', "
        f"filters={filter_dict}"
    )
    # self._validate_filters is removed
    self._validate_sort(sort_by)
    params = self._build_params(
        page=None,
        page_size=page_size,
        sort_by=sort_by,
        filters=filter_dict,
        is_iteration=True,
    )
    async for item in self._iterate_entities_impl(params):
        yield item

search(page=1, page_size=DEFAULT_PAGE_SIZE, sort_by=None, filters=None) async

Searches for Organizations.

Parameters:

Name	Type	Description	Default
page	int	Page number (1-indexed).	1
page_size	int	Number of results per page.	DEFAULT_PAGE_SIZE
sort_by	str | None	Field to sort by.	None
filters	OrganizationsFilters | None	An instance of OrganizationsFilters with filter criteria.	None

Returns:

Type	Description
OrganizationResponse	An OrganizationResponse object.

Source code in src/aireloom/resources/organizations_client.py

async def search(
    self,
    page: int = 1,
    page_size: int = DEFAULT_PAGE_SIZE,
    sort_by: str | None = None,
    filters: OrganizationsFilters | None = None,  # Changed to Pydantic model
) -> OrganizationResponse:
    """Searches for Organizations.

    Args:
        page: Page number (1-indexed).
        page_size: Number of results per page.
        sort_by: Field to sort by.
        filters: An instance of OrganizationsFilters with filter criteria.

    Returns:
        An OrganizationResponse object.
    """
    filter_dict = (
        filters.model_dump(exclude_none=True, by_alias=True) if filters else {}
    )
    logger.info(
        f"Searching organizations: page={page}, size={page_size}, sort='{sort_by}', "
        f"filters={filter_dict}"
    )
    # self._validate_filters is removed
    self._validate_sort(sort_by)
    params = self._build_params(
        page=page, page_size=page_size, sort_by=sort_by, filters=filter_dict
    )
    return await self._search_entities_impl(params)

ProjectsClient

For accessing research project data.

Bases: GettableMixin, SearchableMixin, CursorIterableMixin, BaseResourceClient

Client for the OpenAIRE Projects API endpoint.

This client offers standardized methods (get, search, iterate) for accessing project data by inheriting from bibliofabric mixins. It is configured with the API path and Pydantic models specific to OpenAIRE project entities.

Attributes:

Name	Type	Description
_entity_path	str	The API path for projects.
_entity_model	type[Project]	Pydantic model for a single project.
_search_response_model	type[ProjectResponse]	Pydantic model for the search response envelope.
_valid_sort_fields	set[str]	Valid sort fields for this endpoint.

Source code in src/aireloom/resources/projects_client.py

class ProjectsClient(
    GettableMixin, SearchableMixin, CursorIterableMixin, BaseResourceClient
):
    """Client for the OpenAIRE Projects API endpoint.

    This client offers standardized methods (`get`, `search`, `iterate`) for
    accessing project data by inheriting from `bibliofabric` mixins.
    It is configured with the API path and Pydantic models specific to
    OpenAIRE project entities.

    Attributes:
        _entity_path (str): The API path for projects.
        _entity_model (type[Project]): Pydantic model for a single project.
        _search_response_model (type[ProjectResponse]): Pydantic model for the
                                                        search response envelope.
        _valid_sort_fields (set[str]): Valid sort fields for this endpoint.
    """

    _entity_path: str = PROJECTS
    _entity_model: type[Project] = Project
    _search_response_model: type[ProjectResponse] = ProjectResponse
    _valid_sort_fields = {
        "acronym",
        "code",
        "enddate",
        "fundinglevel",
        "fundingtree",
        "id",
        "startdate",
        "title",
    }

    def __init__(self, api_client: "AireloomClient"):
        """Initializes the ProjectsClient.

        Args:
            api_client: An instance of AireloomClient.
        """
        super().__init__(api_client)
        logger.debug(f"ProjectsClient initialized for path: {self._entity_path}")

__init__(api_client)

Initializes the ProjectsClient.

Parameters:

Name	Type	Description	Default
api_client	AireloomClient	An instance of AireloomClient.	required

Source code in src/aireloom/resources/projects_client.py

def __init__(self, api_client: "AireloomClient"):
    """Initializes the ProjectsClient.

    Args:
        api_client: An instance of AireloomClient.
    """
    super().__init__(api_client)
    logger.debug(f"ProjectsClient initialized for path: {self._entity_path}")

DataSourcesClient

For accessing data source information.

Bases: BaseResourceClient

Client for the OpenAIRE Data Sources API endpoint.

This client allows interaction with OpenAIRE's data source entities, offering methods for retrieval (get), searching (search), and iteration (iterate). It currently employs custom logic for these operations.

Attributes:

Name	Type	Description
_entity_path	str	The API path for data sources.
_entity_model	type[DataSource]	Pydantic model for a single data source.
_response_model	type[DataSourceResponse]	Pydantic model for the search response envelope.
_endpoint_def	dict	Configuration for this endpoint from ENDPOINT_DEFINITIONS.
_valid_sort_fields	set[str]	Valid sort fields for this endpoint.

Source code in src/aireloom/resources/data_sources_client.py

class DataSourcesClient(BaseResourceClient):
    """Client for the OpenAIRE Data Sources API endpoint.

    This client allows interaction with OpenAIRE's data source entities,
    offering methods for retrieval (`get`), searching (`search`), and iteration
    (`iterate`). It currently employs custom logic for these operations.

    Attributes:
        _entity_path (str): The API path for data sources.
        _entity_model (type[DataSource]): Pydantic model for a single data source.
        _response_model (type[DataSourceResponse]): Pydantic model for the
                                                     search response envelope.
        _endpoint_def (dict): Configuration for this endpoint from `ENDPOINT_DEFINITIONS`.
        _valid_sort_fields (set[str]): Valid sort fields for this endpoint.
    """

    _entity_path: str = DATA_SOURCES
    _entity_model: type[DataSource] = DataSource
    _response_model: type[DataSourceResponse] = DataSourceResponse

    def __init__(self, api_client: "AireloomClient"):
        """Initializes the DataSourcesClient.

        Args:
            api_client: An instance of AireloomClient.
        """
        super().__init__(api_client)
        if self._entity_path not in ENDPOINT_DEFINITIONS:
            raise ValueError(
                f"Missing endpoint definition for entity path: {self._entity_path}"
            )
        self._endpoint_def = ENDPOINT_DEFINITIONS[self._entity_path]
        self._valid_sort_fields = self._endpoint_def.get(
            "sort", {}
        ).keys()  # Get sort fields
        logger.debug(f"DataSourcesClient initialized for path: {self._entity_path}")

    # _validate_filters and _validate_and_convert_filter_value are removed as Pydantic handles this.

    def _validate_sort(self, sort_by: str | None) -> None:
        """Validates the sort field against endpoint definitions."""
        if not sort_by:
            return

        if not self._valid_sort_fields:
            logger.warning(
                f"Sort field '{sort_by}' provided for {self._entity_path}, "
                "but no sort fields are defined. Ignoring sort."
            )
            return
        sort_field_name = sort_by.split()[0]
        if sort_field_name not in self._valid_sort_fields:
            raise ValidationError(
                f"Invalid sort field for {self._entity_path}: '{sort_field_name}'. "
                f"Valid fields: {list(self._valid_sort_fields)}"
            )

    def _build_params(
        self,
        page: int | None,
        page_size: int,
        sort_by: str | None,
        filters: dict[str, Any] | None,  # Changed to Optional[dict]
        *,
        is_iteration: bool = False,
    ) -> dict[str, Any]:
        """Builds the query parameter dictionary."""
        params: dict[str, Any] = {"pageSize": page_size}
        if is_iteration:
            params["cursor"] = "*"
        elif page is not None:
            params["page"] = page
        if sort_by:
            params["sortBy"] = sort_by
        if filters:
            params.update(filters)
        return {k: v for k, v in params.items() if v is not None}

    async def _fetch_single_entity_impl(self, entity_id: str) -> DataSource:
        """Generic method to fetch a single entity by ID using search-by-ID."""
        try:
            # Use search with ID parameter instead of direct GET
            params = {"id": entity_id, "pageSize": 1}
            response = await self._api_client.request(
                "GET", self._entity_path, params=params, data=None, json_data=None
            )
            data = response.json()

            # Parse the search response
            search_response = self._response_model.model_validate(data)

            if not search_response.results:
                raise BibliofabricError(
                    f"{self._entity_model.__name__} with ID '{entity_id}' not found."
                )

            # Return the first (and should be only) result
            return search_response.results[0]

        except httpx.HTTPStatusError as e:
            logger.error(
                f"HTTPStatusError for {self._entity_model.__name__} ID '{entity_id}': {e.response.status_code}"
            )
            raise BibliofabricError(
                f"API error fetching {self._entity_model.__name__} {entity_id}: "
                f"Status {e.response.status_code}"
            ) from e
        except Exception as e:
            if isinstance(e, BibliofabricError):
                raise
            logger.exception(
                f"Failed to fetch {self._entity_model.__name__} {entity_id} from {self._entity_path}"
            )
            raise BibliofabricError(
                f"Unexpected error fetching {self._entity_model.__name__} {entity_id}: {e}"
            ) from e

    async def _search_entities_impl(self, params: dict[str, Any]) -> DataSourceResponse:
        """Generic method to search for entities."""
        try:
            response = await self._api_client.request(
                "GET", self._entity_path, params=params, data=None, json_data=None
            )
            return self._response_model.model_validate(response.json())
        except Exception as e:
            if isinstance(e, BibliofabricError | ValidationError):
                raise
            logger.exception(
                f"Failed to search {self._entity_path} with params {params}"
            )
            raise BibliofabricError(
                f"Unexpected error searching {self._entity_path}: {e}"
            ) from e

    async def _iterate_entities_impl(
        self, params: dict[str, Any]
    ) -> AsyncIterator[DataSource]:
        """Generic method to iterate through all results using cursor pagination."""
        current_params = params.copy()
        while True:
            try:
                logger.debug(
                    f"Iterating {self._entity_path} with params: {current_params}"
                )
                response = await self._api_client.request(
                    "GET",
                    self._entity_path,
                    params=current_params,
                    data=None,
                    json_data=None,
                )
                data = response.json()
                api_response = ApiResponse[self._entity_model].model_validate(data)
                if not api_response.results:
                    logger.debug(
                        f"No more results for {self._entity_path}, stopping iteration."
                    )
                    break
                for result in api_response.results:
                    yield result
                next_cursor = api_response.header.nextCursor
                if not next_cursor:
                    logger.debug(
                        f"No nextCursor for {self._entity_path}, stopping iteration."
                    )
                    break
                current_params["cursor"] = next_cursor
                current_params.pop("page", None)
            except Exception as e:
                if isinstance(e, BibliofabricError | ValidationError):
                    raise
                logger.exception(
                    f"Failed during iteration of {self._entity_path} with params {current_params}"
                )
                raise BibliofabricError(
                    f"Unexpected error during iteration of {self._entity_path}: {e}"
                ) from e

    async def get(self, source_id: str) -> DataSource:
        """Retrieves a single Data Source by its ID.

        Args:
            source_id: The ID of the data source.

        Returns:
            A DataSource object.
        """
        logger.info(f"Fetching data source with ID: {source_id}")
        return await self._fetch_single_entity_impl(source_id)

    async def search(
        self,
        page: int = 1,
        page_size: int = DEFAULT_PAGE_SIZE,
        sort_by: str | None = None,
        filters: DataSourcesFilters | None = None,  # Changed to Pydantic model
    ) -> DataSourceResponse:
        """Searches for Data Sources.

        Args:
            page: Page number (1-indexed).
            page_size: Number of results per page.
            sort_by: Field to sort by.
            filters: An instance of DataSourcesFilters with filter criteria.

        Returns:
            A DataSourceResponse object.
        """
        filter_dict = (
            filters.model_dump(exclude_none=True, by_alias=True) if filters else {}
        )
        logger.info(
            f"Searching data sources: page={page}, size={page_size}, sort='{sort_by}', "
            f"filters={filter_dict}"
        )
        # self._validate_filters is removed
        self._validate_sort(sort_by)
        params = self._build_params(
            page=page, page_size=page_size, sort_by=sort_by, filters=filter_dict
        )
        return await self._search_entities_impl(params)

    async def iterate(
        self,
        page_size: int = 100,
        sort_by: str | None = None,
        filters: DataSourcesFilters | None = None,  # Changed to Pydantic model
    ) -> AsyncIterator[DataSource]:
        """Iterates through all Data Source results.

        Args:
            page_size: Number of results per page during iteration.
            sort_by: Field to sort by.
            filters: An instance of DataSourcesFilters with filter criteria.

        Yields:
            DataSource objects.
        """
        filter_dict = (
            filters.model_dump(exclude_none=True, by_alias=True) if filters else {}
        )
        logger.info(
            f"Iterating data sources: size={page_size}, sort='{sort_by}', "
            f"filters={filter_dict}"
        )
        # self._validate_filters is removed
        self._validate_sort(sort_by)
        params = self._build_params(
            page=None,
            page_size=page_size,
            sort_by=sort_by,
            filters=filter_dict,
            is_iteration=True,
        )
        async for item in self._iterate_entities_impl(params):
            yield item

__init__(api_client)

Initializes the DataSourcesClient.

Parameters:

Name	Type	Description	Default
api_client	AireloomClient	An instance of AireloomClient.	required

Source code in src/aireloom/resources/data_sources_client.py

def __init__(self, api_client: "AireloomClient"):
    """Initializes the DataSourcesClient.

    Args:
        api_client: An instance of AireloomClient.
    """
    super().__init__(api_client)
    if self._entity_path not in ENDPOINT_DEFINITIONS:
        raise ValueError(
            f"Missing endpoint definition for entity path: {self._entity_path}"
        )
    self._endpoint_def = ENDPOINT_DEFINITIONS[self._entity_path]
    self._valid_sort_fields = self._endpoint_def.get(
        "sort", {}
    ).keys()  # Get sort fields
    logger.debug(f"DataSourcesClient initialized for path: {self._entity_path}")

get(source_id) async

Retrieves a single Data Source by its ID.

Parameters:

Name	Type	Description	Default
source_id	str	The ID of the data source.	required

Returns:

Type	Description
DataSource	A DataSource object.

Source code in src/aireloom/resources/data_sources_client.py

async def get(self, source_id: str) -> DataSource:
    """Retrieves a single Data Source by its ID.

    Args:
        source_id: The ID of the data source.

    Returns:
        A DataSource object.
    """
    logger.info(f"Fetching data source with ID: {source_id}")
    return await self._fetch_single_entity_impl(source_id)

iterate(page_size=100, sort_by=None, filters=None) async

Iterates through all Data Source results.

Parameters:

Name	Type	Description	Default
page_size	int	Number of results per page during iteration.	100
sort_by	str | None	Field to sort by.	None
filters	DataSourcesFilters | None	An instance of DataSourcesFilters with filter criteria.	None

Yields:

Type	Description
AsyncIterator[DataSource]	DataSource objects.

Source code in src/aireloom/resources/data_sources_client.py

async def iterate(
    self,
    page_size: int = 100,
    sort_by: str | None = None,
    filters: DataSourcesFilters | None = None,  # Changed to Pydantic model
) -> AsyncIterator[DataSource]:
    """Iterates through all Data Source results.

    Args:
        page_size: Number of results per page during iteration.
        sort_by: Field to sort by.
        filters: An instance of DataSourcesFilters with filter criteria.

    Yields:
        DataSource objects.
    """
    filter_dict = (
        filters.model_dump(exclude_none=True, by_alias=True) if filters else {}
    )
    logger.info(
        f"Iterating data sources: size={page_size}, sort='{sort_by}', "
        f"filters={filter_dict}"
    )
    # self._validate_filters is removed
    self._validate_sort(sort_by)
    params = self._build_params(
        page=None,
        page_size=page_size,
        sort_by=sort_by,
        filters=filter_dict,
        is_iteration=True,
    )
    async for item in self._iterate_entities_impl(params):
        yield item

search(page=1, page_size=DEFAULT_PAGE_SIZE, sort_by=None, filters=None) async

Searches for Data Sources.

Parameters:

Name	Type	Description	Default
page	int	Page number (1-indexed).	1
page_size	int	Number of results per page.	DEFAULT_PAGE_SIZE
sort_by	str | None	Field to sort by.	None
filters	DataSourcesFilters | None	An instance of DataSourcesFilters with filter criteria.	None

Returns:

Type	Description
DataSourceResponse	A DataSourceResponse object.

Source code in src/aireloom/resources/data_sources_client.py

async def search(
    self,
    page: int = 1,
    page_size: int = DEFAULT_PAGE_SIZE,
    sort_by: str | None = None,
    filters: DataSourcesFilters | None = None,  # Changed to Pydantic model
) -> DataSourceResponse:
    """Searches for Data Sources.

    Args:
        page: Page number (1-indexed).
        page_size: Number of results per page.
        sort_by: Field to sort by.
        filters: An instance of DataSourcesFilters with filter criteria.

    Returns:
        A DataSourceResponse object.
    """
    filter_dict = (
        filters.model_dump(exclude_none=True, by_alias=True) if filters else {}
    )
    logger.info(
        f"Searching data sources: page={page}, size={page_size}, sort='{sort_by}', "
        f"filters={filter_dict}"
    )
    # self._validate_filters is removed
    self._validate_sort(sort_by)
    params = self._build_params(
        page=page, page_size=page_size, sort_by=sort_by, filters=filter_dict
    )
    return await self._search_entities_impl(params)

ScholixClient

For accessing Scholix link data via the Scholexplorer API.

Bases: BaseResourceClient

Client for the OpenAIRE Scholexplorer API (Scholix links).

This client handles requests to the Scholix API, which provides data on relationships between research artifacts (e.g., citations, supplements). It uses a specific base URL (_scholix_base_url) and custom methods (search_links, iterate_links) tailored to the Scholix API's structure, including its 0-indexed pagination and specific request parameters.

Attributes:

Name	Type	Description
_entity_path	str	The API path for Scholix links (typically "Links").
_scholix_base_url	str	The base URL for the Scholexplorer API.
_endpoint_def	dict	Configuration for this endpoint from ENDPOINT_DEFINITIONS.

Source code in src/aireloom/resources/scholix_client.py

class ScholixClient(BaseResourceClient):
    """Client for the OpenAIRE Scholexplorer API (Scholix links).

    This client handles requests to the Scholix API, which provides data on
    relationships between research artifacts (e.g., citations, supplements).
    It uses a specific base URL (`_scholix_base_url`) and custom methods
    (`search_links`, `iterate_links`) tailored to the Scholix API's structure,
    including its 0-indexed pagination and specific request parameters.

    Attributes:
        _entity_path (str): The API path for Scholix links (typically "Links").
        _scholix_base_url (str): The base URL for the Scholexplorer API.
        _endpoint_def (dict): Configuration for this endpoint from `ENDPOINT_DEFINITIONS`.
    """

    _entity_path: str = SCHOLIX  # This is the endpoint path, typically "Links"

    def __init__(
        self, api_client: "AireloomClient", scholix_base_url: str | None = None
    ):
        """Initializes the ScholixClient.

        Args:
            api_client: An instance of `AireloomClient` to be used for making requests.
            scholix_base_url: Optional base URL for the Scholexplorer API. If None,
                the default from `aireloom.constants` is used.
        """
        super().__init__(api_client)
        self._scholix_base_url: str = scholix_base_url or OPENAIRE_SCHOLIX_API_BASE_URL
        if self._entity_path not in ENDPOINT_DEFINITIONS:
            raise ValueError(
                f"Missing endpoint definition for Scholix path: {self._entity_path}"
            )
        self._endpoint_def = ENDPOINT_DEFINITIONS[self._entity_path]
        # Scholix does not have sort fields defined in ENDPOINT_DEFINITIONS
        logger.debug(
            f"ScholixClient initialized for base URL: {self._scholix_base_url}"
        )

    # _validate_filters and _validate_and_convert_filter_value are removed as Pydantic handles this.
    # Scholix API has specific PID requirements handled in search_links.

    def _build_scholix_params(
        self,
        page: int,
        page_size: int,
        filters: dict[str, Any] | None,
    ) -> dict[str, Any]:
        """Builds the query parameter dictionary specifically for the Scholix API.

        The Scholix API uses 'rows' for page size and expects 'page' to be 0-indexed.

        Args:
            page: The 0-indexed page number.
            page_size: The number of results per page (maps to 'rows' parameter).
            filters: A dictionary of filter criteria to include in the parameters.

        Returns:
            A dictionary of query parameters suitable for the Scholix API.
        """
        # Scholix uses 'rows' for page_size and 0-indexed 'page'
        params: dict[str, Any] = {"page": page, "rows": page_size}
        if filters:
            params.update(filters)
        return {k: v for k, v in params.items() if v is not None}

    async def search_links(
        self,
        page: int = 0,  # Scholix default is 0-indexed
        page_size: int = DEFAULT_PAGE_SIZE,
        filters: ScholixFilters | None = None,  # Changed to Pydantic model
    ) -> ScholixResponse:
        """Searches for Scholexplorer relationship links.

        Args:
            page: The page number to retrieve (0-indexed).
            page_size: The number of results per page.
            filters: An instance of ScholixFilters with filter criteria.
                       `sourcePid` or `targetPid` is typically required within the model.

        Returns:
            A ScholixResponse object containing the results for the requested page.

        Raises:
            ValueError: If neither sourcePid nor targetPid is provided in the filters model.
            BibliofabricError: For API communication errors or unexpected issues.
        """
        filter_dict = (
            filters.model_dump(exclude_none=True, by_alias=True) if filters else {}
        )
        logger.info(
            f"Searching Scholix links: page={page}, size={page_size}, filters={filter_dict}"
        )

        if not filter_dict.get("sourcePid") and not filter_dict.get("targetPid"):
            raise ValueError(
                "Either sourcePid or targetPid must be provided for Scholix search within the filters."
            )

        # Pydantic model validation happens at instantiation or via .model_validate()
        # No need for self._validate_filters(filter_dict) here.

        params = self._build_scholix_params(
            page=page, page_size=page_size, filters=filter_dict
        )

        try:
            response = await self._api_client.request(
                method="GET",
                path=self._entity_path,  # SCHOLIX constant
                params=params,
                base_url_override=self._scholix_base_url,
                data=None,
                json_data=None,
            )
            return ScholixResponse.model_validate(response.json())
        except Exception as e:
            if isinstance(
                e, BibliofabricError | ValidationError
            ):  # ValidationError can come from Pydantic
                raise
            logger.exception(
                f"Failed to search {self._entity_path} with params {params} at {self._scholix_base_url}"
            )
            raise BibliofabricError(
                f"Unexpected error searching {self._entity_path}: {e}"
            ) from e

    async def iterate_links(
        self,
        page_size: int = DEFAULT_PAGE_SIZE,
        filters: ScholixFilters | None = None,  # Changed to Pydantic model
    ) -> AsyncIterator[ScholixRelationship]:
        """Iterates through all Scholexplorer relationship links matching the filters.

        Handles pagination automatically based on 'total_pages'.

        Args:
            page_size: The number of results per page during iteration.
            filters: An instance of ScholixFilters with filter criteria.
                       `sourcePid` or `targetPid` is typically required.

        Yields:
            ScholixRelationship objects matching the query.

        Raises:
            ValueError: If neither sourcePid nor targetPid is provided in the filters.
            BibliofabricError: For API communication errors or unexpected issues.
        """
        # The Pydantic model (ScholixFilters) will be passed to search_links,
        # which now expects the model instance.
        logger.info(
            f"Iterating Scholix links: size={page_size}, filters provided: {filters is not None}"
        )

        current_page = 0
        total_pages = 1  # Assume at least one page initially

        while current_page < total_pages:
            logger.debug(
                f"Iterating Scholix page {current_page + 1}/{total_pages if total_pages > 1 else '?'}"
            )
            try:
                # search_links now takes the ScholixFilters model directly
                response_data = await self.search_links(
                    page=current_page,
                    page_size=page_size,
                    filters=filters,
                )

                if not response_data.result:
                    logger.debug(
                        "No results found on this Scholix page, stopping iteration."
                    )
                    break

                for link in response_data.result:
                    yield link

                if current_page == 0:  # Only update total_pages on the first call
                    total_pages = response_data.total_pages
                    logger.debug(f"Total pages reported by Scholix: {total_pages}")
                    if total_pages == 0:  # No results at all
                        logger.debug(
                            "Scholix reported 0 total pages. Stopping iteration."
                        )
                        break

                if current_page >= total_pages - 1:
                    logger.debug("Last Scholix page processed, stopping iteration.")
                    break

                current_page += 1

            except Exception as e:
                if isinstance(e, BibliofabricError | ValidationError):
                    raise
                logger.exception(
                    f"Failed during iteration of {self._entity_path} on page {current_page}"
                )
                raise BibliofabricError(
                    f"Failed during iteration of {self._entity_path} on page {current_page}: {e}"
                ) from e
        logger.debug("Scholix iteration finished.")

__init__(api_client, scholix_base_url=None)

Initializes the ScholixClient.

Parameters:

Name	Type	Description	Default
api_client	AireloomClient	An instance of AireloomClient to be used for making requests.	required
scholix_base_url	str | None	Optional base URL for the Scholexplorer API. If None, the default from aireloom.constants is used.	None

Source code in src/aireloom/resources/scholix_client.py

def __init__(
    self, api_client: "AireloomClient", scholix_base_url: str | None = None
):
    """Initializes the ScholixClient.

    Args:
        api_client: An instance of `AireloomClient` to be used for making requests.
        scholix_base_url: Optional base URL for the Scholexplorer API. If None,
            the default from `aireloom.constants` is used.
    """
    super().__init__(api_client)
    self._scholix_base_url: str = scholix_base_url or OPENAIRE_SCHOLIX_API_BASE_URL
    if self._entity_path not in ENDPOINT_DEFINITIONS:
        raise ValueError(
            f"Missing endpoint definition for Scholix path: {self._entity_path}"
        )
    self._endpoint_def = ENDPOINT_DEFINITIONS[self._entity_path]
    # Scholix does not have sort fields defined in ENDPOINT_DEFINITIONS
    logger.debug(
        f"ScholixClient initialized for base URL: {self._scholix_base_url}"
    )

iterate_links(page_size=DEFAULT_PAGE_SIZE, filters=None) async

Iterates through all Scholexplorer relationship links matching the filters.

Handles pagination automatically based on 'total_pages'.

Parameters:

Name	Type	Description	Default
page_size	int	The number of results per page during iteration.	DEFAULT_PAGE_SIZE
filters	ScholixFilters | None	An instance of ScholixFilters with filter criteria. sourcePid or targetPid is typically required.	None

Yields:

Type	Description
AsyncIterator[ScholixRelationship]	ScholixRelationship objects matching the query.

Raises:

Type	Description
ValueError	If neither sourcePid nor targetPid is provided in the filters.
BibliofabricError	For API communication errors or unexpected issues.

Source code in src/aireloom/resources/scholix_client.py

async def iterate_links(
    self,
    page_size: int = DEFAULT_PAGE_SIZE,
    filters: ScholixFilters | None = None,  # Changed to Pydantic model
) -> AsyncIterator[ScholixRelationship]:
    """Iterates through all Scholexplorer relationship links matching the filters.

    Handles pagination automatically based on 'total_pages'.

    Args:
        page_size: The number of results per page during iteration.
        filters: An instance of ScholixFilters with filter criteria.
                   `sourcePid` or `targetPid` is typically required.

    Yields:
        ScholixRelationship objects matching the query.

    Raises:
        ValueError: If neither sourcePid nor targetPid is provided in the filters.
        BibliofabricError: For API communication errors or unexpected issues.
    """
    # The Pydantic model (ScholixFilters) will be passed to search_links,
    # which now expects the model instance.
    logger.info(
        f"Iterating Scholix links: size={page_size}, filters provided: {filters is not None}"
    )

    current_page = 0
    total_pages = 1  # Assume at least one page initially

    while current_page < total_pages:
        logger.debug(
            f"Iterating Scholix page {current_page + 1}/{total_pages if total_pages > 1 else '?'}"
        )
        try:
            # search_links now takes the ScholixFilters model directly
            response_data = await self.search_links(
                page=current_page,
                page_size=page_size,
                filters=filters,
            )

            if not response_data.result:
                logger.debug(
                    "No results found on this Scholix page, stopping iteration."
                )
                break

            for link in response_data.result:
                yield link

            if current_page == 0:  # Only update total_pages on the first call
                total_pages = response_data.total_pages
                logger.debug(f"Total pages reported by Scholix: {total_pages}")
                if total_pages == 0:  # No results at all
                    logger.debug(
                        "Scholix reported 0 total pages. Stopping iteration."
                    )
                    break

            if current_page >= total_pages - 1:
                logger.debug("Last Scholix page processed, stopping iteration.")
                break

            current_page += 1

        except Exception as e:
            if isinstance(e, BibliofabricError | ValidationError):
                raise
            logger.exception(
                f"Failed during iteration of {self._entity_path} on page {current_page}"
            )
            raise BibliofabricError(
                f"Failed during iteration of {self._entity_path} on page {current_page}: {e}"
            ) from e
    logger.debug("Scholix iteration finished.")

search_links(page=0, page_size=DEFAULT_PAGE_SIZE, filters=None) async

Searches for Scholexplorer relationship links.

Parameters:

Name	Type	Description	Default
page	int	The page number to retrieve (0-indexed).	0
page_size	int	The number of results per page.	DEFAULT_PAGE_SIZE
filters	ScholixFilters | None	An instance of ScholixFilters with filter criteria. sourcePid or targetPid is typically required within the model.	None

Returns:

Type	Description
ScholixResponse	A ScholixResponse object containing the results for the requested page.

Raises:

Type	Description
ValueError	If neither sourcePid nor targetPid is provided in the filters model.
BibliofabricError	For API communication errors or unexpected issues.

Source code in src/aireloom/resources/scholix_client.py

async def search_links(
    self,
    page: int = 0,  # Scholix default is 0-indexed
    page_size: int = DEFAULT_PAGE_SIZE,
    filters: ScholixFilters | None = None,  # Changed to Pydantic model
) -> ScholixResponse:
    """Searches for Scholexplorer relationship links.

    Args:
        page: The page number to retrieve (0-indexed).
        page_size: The number of results per page.
        filters: An instance of ScholixFilters with filter criteria.
                   `sourcePid` or `targetPid` is typically required within the model.

    Returns:
        A ScholixResponse object containing the results for the requested page.

    Raises:
        ValueError: If neither sourcePid nor targetPid is provided in the filters model.
        BibliofabricError: For API communication errors or unexpected issues.
    """
    filter_dict = (
        filters.model_dump(exclude_none=True, by_alias=True) if filters else {}
    )
    logger.info(
        f"Searching Scholix links: page={page}, size={page_size}, filters={filter_dict}"
    )

    if not filter_dict.get("sourcePid") and not filter_dict.get("targetPid"):
        raise ValueError(
            "Either sourcePid or targetPid must be provided for Scholix search within the filters."
        )

    # Pydantic model validation happens at instantiation or via .model_validate()
    # No need for self._validate_filters(filter_dict) here.

    params = self._build_scholix_params(
        page=page, page_size=page_size, filters=filter_dict
    )

    try:
        response = await self._api_client.request(
            method="GET",
            path=self._entity_path,  # SCHOLIX constant
            params=params,
            base_url_override=self._scholix_base_url,
            data=None,
            json_data=None,
        )
        return ScholixResponse.model_validate(response.json())
    except Exception as e:
        if isinstance(
            e, BibliofabricError | ValidationError
        ):  # ValidationError can come from Pydantic
            raise
        logger.exception(
            f"Failed to search {self._entity_path} with params {params} at {self._scholix_base_url}"
        )
        raise BibliofabricError(
            f"Unexpected error searching {self._entity_path}: {e}"
        ) from e

---

Note: For this extractor to work, the specified Python modules and classes must have docstrings. The level of detail in this API reference depends directly on the comprehensiveness of those docstrings.