Skip to content

Organizations

Models

Pydantic models for representing OpenAIRE Organization entities.

This module defines the Pydantic model for an OpenAIRE Organization, including nested models for country and persistent identifiers (PIDs), based on the OpenAIRE data model documentation. Reference: https://graph.openaire.eu/docs/data-model/entities/organization

OrganizationResponse = ApiResponse[Organization] module-attribute

Type alias for an API response containing a list of Organization entities.

Country

Bases: BaseModel

Represents the country associated with an organization.

Attributes:

Name Type Description
code str | None

The ISO 3166-1 alpha-2 country code (e.g., "GR", "US").
label str | None

The human-readable name of the country (e.g., "Greece").
Source code in src/aireloom/models/organization.py 
16
17
18
19
20
21
22
23
24
25
26
27
class Country(BaseModel):
    """Represents the country associated with an organization.

    Attributes:
        code: The ISO 3166-1 alpha-2 country code (e.g., "GR", "US").
        label: The human-readable name of the country (e.g., "Greece").
    """

    code: str | None = None
    label: str | None = None

    model_config = ConfigDict(extra="allow")

Organization

Bases: BaseEntity

Model representing an OpenAIRE Organization entity.

Captures details about an organization, including its names, website, country, and various persistent identifiers. Inherits the id field from BaseEntity.

Attributes:

Name Type Description
legalShortName str | None

The official short name or acronym of the organization.
legalName str | None

The full official legal name of the organization.
alternativeNames list[str] | None

A list of other known names for the organization.
websiteUrl str | None

The URL of the organization's official website.
country Country | None

A Country object representing the organization's country.
pids list[OrganizationPid] | None

A list of OrganizationPid objects representing various PIDs associated with the organization.
Source code in src/aireloom/models/organization.py 
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
class Organization(BaseEntity):
    """Model representing an OpenAIRE Organization entity.

    Captures details about an organization, including its names, website,
    country, and various persistent identifiers. Inherits the `id` field
    from `BaseEntity`.

    Attributes:
        legalShortName: The official short name or acronym of the organization.
        legalName: The full official legal name of the organization.
        alternativeNames: A list of other known names for the organization.
        websiteUrl: The URL of the organization's official website.
        country: A `Country` object representing the organization's country.
        pids: A list of `OrganizationPid` objects representing various PIDs
              associated with the organization.
    """

    # id is inherited from BaseEntity
    legalShortName: str | None = None
    legalName: str | None = None
    alternativeNames: list[str] | None = Field(default_factory=list)
    websiteUrl: str | None = None
    country: Country | None = None
    pids: list[OrganizationPid] | None = Field(default_factory=list)

    model_config = ConfigDict(extra="allow")

OrganizationPid

Bases: BaseModel

Represents a persistent identifier (PID) for an organization.

Attributes:

Name Type Description
scheme str | None

The scheme of the PID (e.g., "ror", "grid", "isni").
value str | None

The value of the PID.
Source code in src/aireloom/models/organization.py 
30
31
32
33
34
35
36
37
38
39
40
41
class OrganizationPid(BaseModel):
    """Represents a persistent identifier (PID) for an organization.

    Attributes:
        scheme: The scheme of the PID (e.g., "ror", "grid", "isni").
        value: The value of the PID.
    """

    scheme: str | None = None
    value: str | None = None

    model_config = ConfigDict(extra="allow")

Filters

Bases: BaseModel

Filter model for Organizations API endpoint.

Attributes:

Name Type Description
search str | None

Search term for the organization.
legalName str | None

Legal name of the organization.
legalShortName str | None

Legal short name of the organization.
id str | None

OpenAIRE id for the organization.
pid str | None

Persistent identifier for the organization.
countryCode str | None

Country code of the organization.
relCommunityId str | None

Related community ID.
relCollectedFromDatasourceId str | None

ID of the datasource from which this was collected.
Source code in src/aireloom/endpoints.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
class OrganizationsFilters(BaseModel):
    """Filter model for Organizations API endpoint.

    Attributes:
        search (str | None): Search term for the organization.
        legalName (str | None): Legal name of the organization.
        legalShortName (str | None): Legal short name of the organization.
        id (str | None): OpenAIRE id for the organization.
        pid (str | None): Persistent identifier for the organization.
        countryCode (str | None): Country code of the organization.
        relCommunityId (str | None): Related community ID.
        relCollectedFromDatasourceId (str | None): ID of the datasource from which this was collected.
    """

    search: str | None = None
    legalName: str | None = None
    legalShortName: str | None = None
    id: str | None = None
    pid: str | None = None
    countryCode: str | None = None
    relCommunityId: str | None = None
    relCollectedFromDatasourceId: str | None = None

    model_config = ConfigDict(extra="forbid")

Client

Client for interacting with the OpenAIRE Organizations API endpoint.

This module provides the OrganizationsClient for accessing OpenAIRE's organization data. Unlike some other resource clients in aireloom that fully leverage bibliofabric mixins for CRUD operations, this client currently maintains its own implementations for get, search, and iterate, though they are based on similar patterns.

OrganizationsClient

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 
 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
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
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 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
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 
216
217
218
219
220
221
222
223
224
225
226
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 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
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 
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
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)