Skip to content

Scholix

ScholixFilters

Bases: BaseModel

Filter model for Scholix API endpoint.

Attributes:

Name Type Description
sourcePid str | None

Persistent identifier of the source entity.
targetPid str | None

Persistent identifier of the target entity.
sourcePublisher str | None

Publisher of the source entity.
targetPublisher str | None

Publisher of the target entity.
sourceType Literal['Publication', 'Dataset', 'Software', 'Other'] | None

Type of the source entity.
targetType Literal['Publication', 'Dataset', 'Software', 'Other'] | None

Type of the target entity.
relation str | None

Type of relation between the source and target entities.
from_date date | None

Start date of the relation (API calls use "from").
to_date date | None

End date of the relation (API calls use "to").
Source code in src/aireloom/endpoints.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
class ScholixFilters(BaseModel):
    """Filter model for Scholix API endpoint.

    Attributes:
        sourcePid (str | None): Persistent identifier of the source entity.
        targetPid (str | None): Persistent identifier of the target entity.
        sourcePublisher (str | None): Publisher of the source entity.
        targetPublisher (str | None): Publisher of the target entity.
        sourceType (Literal["Publication", "Dataset", "Software", "Other"] | None): Type of the source entity.
        targetType (Literal["Publication", "Dataset", "Software", "Other"] | None): Type of the target entity.
        relation (str | None): Type of relation between the source and target entities.
        from_date (date | None): Start date of the relation (API calls use "from").
        to_date (date | None): End date of the relation (API calls use "to").
    """

    sourcePid: str | None = None
    targetPid: str | None = None
    sourcePublisher: str | None = None
    targetPublisher: str | None = None
    sourceType: Literal["Publication", "Dataset", "Software", "Other"] | None = None
    targetType: Literal["Publication", "Dataset", "Software", "Other"] | None = None
    relation: str | None = None
    from_date: date | None = Field(default=None, alias="from")  # API uses "from"
    to_date: date | None = Field(default=None, alias="to")  # API uses "to"

    model_config = ConfigDict(extra="forbid", populate_by_name=True)

from_date = Field(default=None, alias='from') class-attribute instance-attribute

model_config = ConfigDict(extra='forbid', populate_by_name=True) class-attribute instance-attribute

relation = None class-attribute instance-attribute

sourcePid = None class-attribute instance-attribute

sourcePublisher = None class-attribute instance-attribute

sourceType = None class-attribute instance-attribute

targetPid = None class-attribute instance-attribute

targetPublisher = None class-attribute instance-attribute

targetType = None class-attribute instance-attribute

to_date = Field(default=None, alias='to') class-attribute instance-attribute

ScholixClient

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 (_base_url_override) 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").
_base_url_override 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 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
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 (``_base_url_override``) 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").
        _base_url_override (str): The base URL for the Scholexplorer API.
        _endpoint_def (dict): Configuration for this endpoint from ``ENDPOINT_DEFINITIONS``.
    """

    _base_url_override: str | None = None
    _entity_model = None
    _search_response_model = ScholixResponse
    _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._base_url_override = 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._base_url_override}"
        )

    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 v3 API uses 'size' 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 'size' parameter).
            filters: A dictionary of filter criteria to include in the parameters.

        Returns:
            A dictionary of query parameters suitable for the Scholix API.
        """
        # Scholix v3 uses 'size' for page_size and 0-indexed 'page'
        params: dict[str, Any] = {"page": page, "size": 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 (max 99; values >=100 are
                silently truncated to 10 by V3).
            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."
            )
        if page_size > MAX_LINK_PAGE_SIZE:
            logger.warning(
                "Scholix page_size=%d exceeds the V3 maximum of %d "
                "(values >=100 are silently truncated to 10); clamping.",
                page_size,
                MAX_LINK_PAGE_SIZE,
            )
            page_size = MAX_LINK_PAGE_SIZE

        # Pydantic model validation happens at instantiation or via .model_validate()

        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._base_url_override,
                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._base_url_override}"
            )
            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.")

    # ── Standard-name aliases for BaseResourceClient.collect/count/first ──

    async def search(
        self,
        page: int = 0,
        page_size: int = DEFAULT_PAGE_SIZE,
        filters: ScholixFilters | None = None,
        sort_by: str | None = None,
        search: str | None = None,  # noqa: ARG002 — Scholix doesn't support free-text search
    ) -> ScholixResponse:
        """Alias for ``search_links`` so ``collect``/``count`` can find it."""
        return await self.search_links(page=page, page_size=page_size, filters=filters)

    async def iterate(
        self,
        page_size: int = DEFAULT_PAGE_SIZE,
        filters: ScholixFilters | None = None,
        sort_by: str | None = None,
        search: str | None = None,  # noqa: ARG002
    ) -> AsyncIterator[ScholixRelationship]:
        """Alias for ``iterate_links`` so ``collect``/``count`` can find it."""
        async for link in self.iterate_links(page_size=page_size, filters=filters):
            yield link

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

Alias for search_links so collect/count can find it.

Source code in src/aireloom/resources/scholix_client.py 
249
250
251
252
253
254
255
256
257
258
async def search(
    self,
    page: int = 0,
    page_size: int = DEFAULT_PAGE_SIZE,
    filters: ScholixFilters | None = None,
    sort_by: str | None = None,
    search: str | None = None,  # noqa: ARG002 — Scholix doesn't support free-text search
) -> ScholixResponse:
    """Alias for ``search_links`` so ``collect``/``count`` can find it."""
    return await self.search_links(page=page, page_size=page_size, filters=filters)

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 (max 99; values >=100 are silently truncated to 10 by V3).

 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 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
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 (max 99; values >=100 are
            silently truncated to 10 by V3).
        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."
        )
    if page_size > MAX_LINK_PAGE_SIZE:
        logger.warning(
            "Scholix page_size=%d exceeds the V3 maximum of %d "
            "(values >=100 are silently truncated to 10); clamping.",
            page_size,
            MAX_LINK_PAGE_SIZE,
        )
        page_size = MAX_LINK_PAGE_SIZE

    # Pydantic model validation happens at instantiation or via .model_validate()

    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._base_url_override,
            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._base_url_override}"
        )
        raise BibliofabricError(
            f"Unexpected error searching {self._entity_path}: {e}"
        ) from e

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 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
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.")

iterate(page_size=DEFAULT_PAGE_SIZE, filters=None, sort_by=None, search=None) async

Alias for iterate_links so collect/count can find it.

Source code in src/aireloom/resources/scholix_client.py 
260
261
262
263
264
265
266
267
268
269
async def iterate(
    self,
    page_size: int = DEFAULT_PAGE_SIZE,
    filters: ScholixFilters | None = None,
    sort_by: str | None = None,
    search: str | None = None,  # noqa: ARG002
) -> AsyncIterator[ScholixRelationship]:
    """Alias for ``iterate_links`` so ``collect``/``count`` can find it."""
    async for link in self.iterate_links(page_size=page_size, filters=filters):
        yield link