diff --git a/bookops_worldcat/metadata_api.py b/bookops_worldcat/metadata_api.py
index 031d7a4..8f83278 100644
--- a/bookops_worldcat/metadata_api.py
+++ b/bookops_worldcat/metadata_api.py
@@ -27,7 +27,7 @@ def __init__(
) -> None:
"""
Args:
- authorization: WorlcatAccessToken object
+ authorization: WorldcatAccessToken object
agent: "User-agent" parameter to be passed in the request
header; usage strongly encouraged
timeout: how long to wait for server to send data before
@@ -82,6 +82,60 @@ def _url_manage_ih_unset(self, oclcNumber: str) -> str:
def _url_manage_ih_current(self) -> str:
return f"{self.BASE_URL}/manage/institution/holdings/current"
+ def _url_manage_bibs_validate(self, validationLevel: str) -> str:
+ return f"{self.BASE_URL}/manage/bibs/validate/{validationLevel}"
+
+ def _url_manage_bibs_create(self) -> str:
+ return f"{self.BASE_URL}/manage/bibs"
+
+ def _url_manage_bibs_match(self) -> str:
+ return f"{self.BASE_URL}/manage/bibs/match"
+
+ def _url_manage_ih_set_on_record(self) -> str:
+ return f"{self.BASE_URL}/manage/institution/holdings/set"
+
+ def _url_manage_ih_unset_on_record(self) -> str:
+ return f"{self.BASE_URL}/manage/institution/holdings/unset"
+
+ def _url_manage_ih_codes(self) -> str:
+ return f"{self.BASE_URL}/manage/institution/holding-codes"
+
+ def _url_manage_lbd_create(self) -> str:
+ return f"{self.BASE_URL}/manage/lbds"
+
+ def _url_manage_lbd(self, controlNumber: Union[str, int]) -> str:
+ return f"{self.BASE_URL}/manage/lbds/{controlNumber}"
+
+ def _url_manage_lhr_create(self) -> str:
+ return f"{self.BASE_URL}/manage/lhrs"
+
+ def _url_manage_lhr(self, controlNumber: Union[str, int]) -> str:
+ return f"{self.BASE_URL}/manage/lhrs/{controlNumber}"
+
+ def _url_search_general_holdings_summary(self) -> str:
+ return f"{self.BASE_URL}/search/summary-holdings"
+
+ def _url_search_classification_bibs(self, oclcNumber: str) -> str:
+ return f"{self.BASE_URL}/search/classification-bibs/{oclcNumber}"
+
+ def _url_search_lhr_shared_print(self) -> str:
+ return f"{self.BASE_URL}/search/retained-holdings"
+
+ def _url_search_lhr_control_number(self, controlNumber: Union[str, int]) -> str:
+ return f"{self.BASE_URL}/search/my-holdings/{controlNumber}"
+
+ def _url_search_lhr(self) -> str:
+ return f"{self.BASE_URL}/search/my-holdings"
+
+ def _url_browse_lhr(self) -> str:
+ return f"{self.BASE_URL}/browse/my-holdings"
+
+ def _url_search_lbd_control_number(self, controlNumber: Union[str, int]) -> str:
+ return f"{self.BASE_URL}/search/my-local-bib-data/{controlNumber}"
+
+ def _url_search_lbd(self) -> str:
+ return f"{self.BASE_URL}/search/my-local-bib-data"
+
def get_brief_bib(
self, oclcNumber: Union[int, str], hooks: Optional[Dict[str, Callable]] = None
) -> Optional[Response]:
@@ -311,7 +365,7 @@ def search_brief_bibs_other_editions(
heldByGroup: restricts to holdings held by group symbol
heldBySymbol: restricts to holdings with specified intitution
symbol
- heldByInstitutionID: restrict to specified institution regisgtryId
+ heldByInstitutionID: restrict to specified institution registryId
inLanguage: restrics the response to the single
specified language, example: 'fre'
inCataloglanguage: restrics the response to specified
@@ -351,7 +405,7 @@ def search_brief_bibs_other_editions(
preferredLanguage: language of metadata description,
offset: start position of bibliographic records to
return; default 1
- limit: maximum nuber of records to return;
+ limit: maximum number of records to return;
maximum 50, default 10
orderBy: sort of restuls;
available values:
@@ -526,7 +580,7 @@ def search_brief_bibs(
'title'
offset: start position of bibliographic records to
return; default 1
- limit: maximum nuber of records to return;
+ limit: maximum number of records to return;
maximum 50, default 10
hooks: Requests library hook system that can be
used for signal event handling, see more at:
@@ -787,3 +841,1026 @@ def search_shared_print_holdings(
query = Query(self, prepared_request, timeout=self.timeout)
return query.response
+
+ def browse_my_holdings(
+ self,
+ callNumber: Optional[str] = None,
+ oclcNumber: Optional[Union[int, str]] = None,
+ holdingLocation: str = "",
+ shelvingLocation: str = "",
+ browsePosition: Optional[str] = None,
+ limit: Optional[int] = None,
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Browse local holdings
+ Uses /browse/my-holdings endpoint.
+
+ Args:
+ callNumber: call number for item
+ oclcNumber: OCLC bibliographic record number; can be
+ an integer or string with or without OCLC #
+ prefix
+ holdingLocation: holding location for item
+ shelvingLocation: shelving location for item
+ browsePosition: position within browse list where the matching
+ record should be, default is 10
+ limit: maximum number of records to return;
+ maximum 50, default 10
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` instance
+ """
+ if not holdingLocation:
+ raise TypeError("Argument 'holdingLocation' is missing.")
+ if not shelvingLocation:
+ raise TypeError("Argument 'shelvingLocation' is missing.")
+
+ if oclcNumber is not None:
+ oclcNumber = verify_oclc_number(oclcNumber)
+
+ url = self._url_browse_lhr()
+ header = {"Accept": "application/json"}
+ payload = {
+ "callNumber": callNumber,
+ "oclcNumber": oclcNumber,
+ "holdingLocation": holdingLocation,
+ "shelvingLocation": shelvingLocation,
+ "browsePosition": browsePosition,
+ "limit": limit,
+ }
+
+ # prep request
+ req = Request("GET", url, params=payload, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def create_bib(
+ self,
+ record: Optional[str] = None,
+ recordFormat: Optional[str] = None,
+ responseFormat: Optional[str] = "application/marcxml+xml",
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Create a bib record in OCLC if it does not already exist
+ Uses /manage/bibs endpoint.
+
+ Args:
+ record: MARC record to be created
+ recordFormat: format of MARC record, options:
+ 'application/marcxml+xml', 'application/marc'
+ responseFormat: format of returned record; options:
+ 'application/marcxml+xml', 'application/marc'
+ default is 'application/marcxml+xml'
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` instance
+ """
+ if not record:
+ raise TypeError("Argument 'record' is missing.")
+ if not recordFormat:
+ raise TypeError("Argument 'recordFormat' is missing.")
+
+ url = self._url_manage_bibs_create()
+ header = {
+ "Accept": responseFormat,
+ "content-type": recordFormat,
+ }
+
+ # prep request
+ req = Request("POST", url, data=record, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def create_lbd(
+ self,
+ record: Optional[str] = None,
+ recordFormat: Optional[str] = None,
+ responseFormat: Optional[str] = "application/marcxml+xml",
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Given a local bibliographic data record, create it in WorldCat
+ Uses /manage/lbds endpoint.
+
+ Args:
+ record: MARC record to be created
+ recordFormat: format of MARC record, options:
+ 'application/marcxml+xml', 'application/marc'
+ responseFormat: format of returned record; options:
+ 'application/marcxml+xml', 'application/marc'
+ default is 'application/marcxml+xml'
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` instance
+ """
+ if not record:
+ raise TypeError("Argument 'record' is missing.")
+ if not recordFormat:
+ raise TypeError("Argument 'recordFormat' is missing.")
+
+ url = self._url_manage_lbd_create()
+ header = {
+ "Accept": responseFormat,
+ "content-type": recordFormat,
+ }
+
+ # prep request
+ req = Request("POST", url, data=record, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def create_lhr(
+ self,
+ record: Optional[str] = None,
+ recordFormat: Optional[str] = None,
+ responseFormat: Optional[str] = "application/marcxml+xml",
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Given a local holdings record, create it in WorldCat
+ Uses /manage/lhrs endpoint.
+
+ Args:
+ record: Holdings record to be created
+ recordFormat: format of MARC record, options:
+ 'application/marcxml+xml', 'application/marc'
+ responseFormat: format of returned record; options:
+ 'application/marcxml+xml', 'application/marc'
+ default is 'application/marcxml+xml'
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` instance
+ """
+ if not record:
+ raise TypeError("Argument 'record' is missing.")
+ if not recordFormat:
+ raise TypeError("Argument 'recordFormat' is missing.")
+
+ url = self._url_manage_lhr_create()
+ header = {
+ "Accept": responseFormat,
+ "content-type": recordFormat,
+ }
+
+ # prep request
+ req = Request("POST", url, data=record, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def delete_lbd(
+ self,
+ controlNumber: Union[int, str],
+ responseFormat: Optional[str] = "application/marcxml+xml",
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Given a control number, delete the associated Local Bibliographic Data record
+ Uses /manage/lbds/{controlNumber} endpoint.
+
+ Args:
+ controlNumber: control number associated with Local Bibliographic
+ Data record; can be an integer, or string
+ response_format: format of returned record, options:
+ 'application/marcxml+xml', 'application/marc',
+ default is 'application/marcxml+xml'
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ url = self._url_manage_lbd(controlNumber)
+ header = {"Accept": responseFormat}
+
+ # prep request
+ req = Request("DELETE", url, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def delete_lhr(
+ self,
+ controlNumber: Union[int, str],
+ responseFormat: Optional[str] = "application/marcxml+xml",
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Given a control number, delete a Local Holdings record.
+ Uses /manage/lhrs/{controlNumber} endpoint.
+
+ Args:
+ controlNumber: control number associated with Local Holdings
+ record; can be an integer, or string
+ response_format: format of returned record, options:
+ 'application/marcxml+xml', 'application/marc',
+ default is 'application/marcxml+xml'
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ url = self._url_manage_lhr(controlNumber)
+ header = {"Accept": responseFormat}
+
+ # prep request
+ req = Request("DELETE", url, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def get_bib_classification(
+ self,
+ oclcNumber: Union[int, str],
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Given an OCLC number, retrieve classification recommendations for the bib record
+ Uses /search/classification-bibs/{oclcNumber} endpoint.
+
+ Args:
+ oclcNumber: OCLC bibliographic record number; can be
+ an integer or string with or without OCLC # prefix
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ oclcNumber = verify_oclc_number(oclcNumber)
+
+ url = self._url_search_classification_bibs(oclcNumber)
+ header = {"Accept": "application/json"}
+
+ # prep request
+ req = Request("GET", url, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def get_bib_holdings(
+ self,
+ oclcNumber: Union[int, str],
+ holdingsAllEditions: Optional[bool] = None,
+ holdingsAllVariantRecords: Optional[bool] = None,
+ holdingsFilterFormat: Optional[List[str]] = None,
+ heldInCountry: Optional[str] = None,
+ heldInState: Optional[str] = None,
+ heldByGroup: Optional[str] = None,
+ heldBySymbol: Optional[List[str]] = None,
+ heldByInstitutionID: Optional[List[int]] = None,
+ heldByLibraryType: Optional[List[str]] = None,
+ lat: Optional[float] = None,
+ lon: Optional[float] = None,
+ distance: Optional[int] = None,
+ unit: Optional[str] = None,
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Given an OCLC number get summary of holdings
+ Uses /search/summary-holdings endpoint.
+
+ Args:
+ oclcNumber: OCLC bibliographic record number; can be
+ an integer or string with or without OCLC #
+ prefix
+ holdingsAllEditions: get holdings for all editions;
+ options: True, False, default is False
+ holdingsAllVariantRecords: get holdings for specific edition across
+ all variant records; options: True, False,
+ default is False
+ holdingsFilterFormat: get holdings for specific itemSubType,
+ example: book-digital
+ heldInCountry: limits to holdings held by institutions
+ in requested country
+ heldInState: limits to holdings held by institutions
+ in requested state, example: 'US-NY'
+ heldByGroup: limits to holdings held by institutions
+ indicated by group symbol
+ heldBySymbol: limits to holdings held by institutions
+ indicated by institution symbol
+ heldByInstitutionID: limits to holdings held by institutions
+ indicated by institution registryID
+ heldByLibraryType: limits to holdings held by library type,
+ options: 'PUBLIC', 'ALL'
+ lat: limit to latitude, example: 37.502508
+ lon: limit to longitute, example: -122.22702
+ distance: distance from latitude and longitude
+ unit: unit of distance param; options:
+ 'M' (miles) or 'K' (kilometers)
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ oclcNumber = verify_oclc_number(oclcNumber)
+
+ url = self._url_search_general_holdings_summary()
+ header = {"Accept": "application/json"}
+ payload = {
+ "oclcNumber": oclcNumber,
+ "holdingsAllEditions": holdingsAllEditions,
+ "holdingsAllVariantRecords": holdingsAllVariantRecords,
+ "holdingsFilterFormat": holdingsFilterFormat,
+ "heldInCountry": heldInCountry,
+ "heldInState": heldInState,
+ "heldByGroup": heldByGroup,
+ "heldBySymbol": heldBySymbol,
+ "heldByInstitutionID": heldByInstitutionID,
+ "heldByLibraryType": heldByLibraryType,
+ "lat": lat,
+ "lon": lon,
+ "distance": distance,
+ "unit": unit,
+ }
+
+ # prep request
+ req = Request("GET", url, params=payload, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def get_institution_holding_codes(
+ self,
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Retrieve the all holding codes for the authenticated institution.
+ Uses /manage/institution/holding-codes endpoint.
+
+ Args:
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ url = self._url_manage_ih_codes()
+ header = {"Accept": "application/json"}
+
+ # prep request
+ req = Request("GET", url, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def get_lbd_data(
+ self,
+ controlNumber: Union[int, str],
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Retrieve LBD Resource
+ Uses /search/my-local-bib-data/{controlNumber} endpoint.
+
+ Args:
+ controlNumber: control number associated with Local Bibliographic
+ Data record; can be an integer, or string
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ url = self._url_search_lbd_control_number(controlNumber)
+ header = {"Accept": "application/json"}
+
+ # prep request
+ req = Request("GET", url, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def get_lhr_data(
+ self,
+ controlNumber: Union[int, str],
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Retrieve LHR Resource
+ Uses /search/my-holdings/{controlNumber} endpoint.
+
+ Args:
+ controlNumber: control number associated with Local Holdings
+ record; can be an integer, or string
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ url = self._url_search_lhr_control_number(controlNumber)
+ header = {"Accept": "application/json"}
+
+ # prep request
+ req = Request("GET", url, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def get_lbd_record(
+ self,
+ controlNumber: Union[int, str],
+ response_format: Optional[str] = "application/marcxml+xml",
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Given a Control Number, retrieve a Local Bibliographic Data record.
+ Uses /manage/lbds/{controlNumber} endpoint.
+
+ Args:
+ controlNumber: control number associated with Local Bibliographic
+ Data record; can be an integer, or string
+ response_format: format of returned record, options:
+ 'application/marcxml+xml', 'application/marc',
+ default is 'application/marcxml+xml'
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ url = self._url_manage_lbd(controlNumber)
+ header = {"Accept": response_format}
+
+ # prep request
+ req = Request("GET", url, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def get_lhr_record(
+ self,
+ controlNumber: Union[int, str],
+ response_format: Optional[str] = "application/marcxml+xml",
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Send a GET request for a local holdings record
+ Uses /manage/lhrs/{controlNumber} endpoint.
+
+ Args:
+ controlNumber: control number associated with Local Holdings
+ record; can be an integer, or string
+ response_format: format of returned record, options:
+ 'application/marcxml+xml', 'application/marc',
+ default is 'application/marcxml+xml'
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ url = self._url_manage_lhr(controlNumber)
+ header = {"Accept": response_format}
+
+ # prep request
+ req = Request("GET", url, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def holding_set_on_record(
+ self,
+ record: Optional[str] = None,
+ recordFormat: Optional[str] = None,
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Given a MARC record in MARC XML or MARC21, set a holding on the record.
+ MARC record must contain OCLC number in 001 or 035 subfield a.
+ Only one MARC record is allowed in the request body.
+ Uses /manage/institution/holdings/set endpoint.
+
+ Args:
+ record: MARC record on which to set holdings
+ recordFormat: format of MARC record, options:
+ 'application/marcxml+xml', 'application/marc'
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ if not record:
+ raise TypeError("Argument 'record' is missing.")
+ if not recordFormat:
+ raise TypeError("Argument 'recordFormat' is missing.")
+
+ url = self._url_manage_ih_set_on_record()
+ header = {
+ "Accept": "application/json",
+ "content-type": recordFormat,
+ }
+
+ # prep request
+ req = Request("POST", url, data=record, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def holding_unset_on_record(
+ self,
+ record: Optional[str] = None,
+ recordFormat: Optional[str] = None,
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Given a MARC record in MARC XML or MARC21, unset a holding on the record.
+ MARC record must contain OCLC number in 001 or 035 subfield a.
+ Only one MARC record is allowed in the request body.
+ Uses /manage/institution/holdings/unset endpoint.
+
+ Args:
+ record: MARC record on which to unset holdings
+ recordFormat: format of MARC record, options:
+ 'application/marcxml+xml', 'application/marc'
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ if not record:
+ raise TypeError("Argument 'record' is missing.")
+ if not recordFormat:
+ raise TypeError("Argument 'recordFormat' is missing.")
+
+ url = self._url_manage_ih_unset_on_record()
+ header = {
+ "Accept": "application/json",
+ "content-type": recordFormat,
+ }
+
+ # prep request
+ req = Request("POST", url, data=record, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+ return query.response
+
+ def match_bib(
+ self,
+ record: Optional[str] = None,
+ recordFormat: Optional[str] = None,
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Given a bib record in MARC21 or MARCXML identify the best match in WorldCat.
+ Record must contain at minimum an 008 and 245. Response contains number of
+ potential matches in numberOfRecords and best match in briefRecords
+ Uses /manage/bibs/match endpoint.
+
+ Args:
+ record: MARC record to be matched
+ recordFormat: format of MARC record, options:
+ 'application/marcxml+xml', 'application/marc'
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` instance
+ """
+ if not record:
+ raise TypeError("Argument 'record' is missing.")
+ if not recordFormat:
+ raise TypeError("Argument 'recordFormat' is missing.")
+
+ url = self._url_manage_bibs_match()
+ header = {
+ "Accept": "application/json",
+ "content-type": recordFormat,
+ }
+
+ # prep request
+ req = Request("POST", url, data=record, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def replace_bib(
+ self,
+ oclcNumber: Union[int, str],
+ record: Optional[str] = None,
+ recordFormat: Optional[str] = None,
+ responseFormat: Optional[str] = "application/marcxml+xml",
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Given an OCLC number and MARC record, find record in WorldCat and replace it.
+ If the record does not exist in WorldCat, a new bib record will be created.
+ Uses /manage/bibs/{oclcNumber} endpoint.
+
+ Args:
+ oclcNumber: OCLC bibliographic record number for record to be
+ replaced; can be an integer or string with or
+ without OCLC # prefix
+ record: MARC record to replace existing WorldCat record
+ recordFormat: format of MARC record, options:
+ 'application/marcxml+xml', 'application/marc'
+ responseFormat: format of returned record; options:
+ 'application/marcxml+xml', 'application/marc'
+ default is 'application/marcxml+xml'
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` instance
+ """
+ oclcNumber = verify_oclc_number(oclcNumber)
+
+ if not record:
+ raise TypeError("Argument 'record' is missing.")
+ if not recordFormat:
+ raise TypeError("Argument 'recordFormat' is missing.")
+
+ url = self._url_manage_bibs(oclcNumber)
+ header = {
+ "Accept": responseFormat,
+ "content-type": recordFormat,
+ }
+
+ # prep request
+ req = Request("PUT", url, data=record, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def replace_lbd(
+ self,
+ controlNumber: Union[int, str],
+ record: Optional[str] = None,
+ recordFormat: Optional[str] = None,
+ responseFormat: Optional[str] = "application/marcxml+xml",
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Given a Control Number, find the associated Local Bibliographic Data
+ Record and replace it. If the Control Number is not found in
+ WorldCat, then the provided Local Bibliographic Data Record will be created.
+ Uses /manage/lbds/{controlNumber} endpoint.
+
+ Args:
+ controlNumber: control number associated with Local Bibliographic
+ Data record; can be an integer, or string
+ response_format: format of returned record, options:
+ 'application/marcxml+xml', 'application/marc',
+ default is 'application/marcxml+xml'
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ if not record:
+ raise TypeError("Argument 'record' is missing.")
+ if not recordFormat:
+ raise TypeError("Argument 'recordFormat' is missing.")
+
+ url = self._url_manage_lbd(controlNumber)
+ header = {
+ "Accept": responseFormat,
+ "content-type": recordFormat,
+ }
+
+ # prep request
+ req = Request("PUT", url, data=record, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def replace_lhr(
+ self,
+ controlNumber: Union[int, str],
+ record: Optional[str] = None,
+ recordFormat: Optional[str] = None,
+ responseFormat: Optional[str] = "application/marcxml+xml",
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Given a Control Number, find the associated Local Holdings
+ Record and replace it. If the Control Number is not found in
+ WorldCat, then the provided Local Holdings Record will be created.
+ Uses /manage/lhrs/{controlNumber} endpoint.
+
+ Args:
+ controlNumber: control number associated with Local Holdings
+ record; can be an integer, or string
+ response_format: format of returned record, options:
+ 'application/marcxml+xml', 'application/marc',
+ default is 'application/marcxml+xml'
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ if not record:
+ raise TypeError("Argument 'record' is missing.")
+ if not recordFormat:
+ raise TypeError("Argument 'recordFormat' is missing.")
+
+ url = self._url_manage_lhr(controlNumber)
+ header = {
+ "Accept": responseFormat,
+ "content-type": recordFormat,
+ }
+
+ # prep request
+ req = Request("PUT", url, data=record, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def search_my_holdings(
+ self,
+ oclcNumber: Optional[Union[int, str]] = None,
+ barcode: Optional[str] = None,
+ orderBy: Optional[str] = None,
+ offset: Optional[int] = None,
+ limit: Optional[int] = None,
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Search LHR Resources
+ Uses /search/my-holdings endpoint.
+
+ Args:
+ oclcNumber: OCLC bibliographic record number; can be
+ an integer, or string that can include
+ OCLC # prefix
+ barcode: barcode as a string,
+ orderBy: results sort key;
+ options:
+ 'commitmentExpirationDate'
+ 'location'
+ 'oclcSymbol'
+ default is 'oclcSymbol'
+ offset: start position of bibliographic records to
+ return; default is 1
+ limit: maximum number of records to return;
+ maximum is 50, default is 10
+ hooks: Requests library hook system that can be used for
+ signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ if oclcNumber is not None:
+ oclcNumber = verify_oclc_number(oclcNumber)
+
+ url = self._url_search_lhr()
+ header = {"Accept": "application/json"}
+ payload = {
+ "oclcNumber": oclcNumber,
+ "barcode": barcode,
+ "orderBy": orderBy,
+ "offset": offset,
+ "limit": limit,
+ }
+
+ # prep request
+ req = Request("GET", url, params=payload, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def search_my_local_bibs(
+ self,
+ q: str,
+ offset: Optional[int] = None,
+ limit: Optional[int] = None,
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Search LBD Resources
+ Uses /search/my-local-bib-data endpoint.
+
+ Args:
+ q: query in the form of a keyword search or
+ fielded search;
+ examples:
+ ti:Zendegi
+ ti:"Czarne oceany"
+ bn:9781680502404
+ kw:python databases
+ ti:Zendegi AND au:greg egan
+ (au:Okken OR au:Myers) AND su:python
+ offset: start position of bibliographic records to
+ return; default is 1
+ limit: maximum number of records to return;
+ maximum is 50, default is 10
+ hooks: Requests library hook system that can be used for
+ signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ if not q:
+ raise TypeError("Argument 'q' is requried to construct query.")
+
+ url = self._url_search_lbd()
+ header = {"Accept": "application/json"}
+ payload = {"q": q, "offset": offset, "limit": limit}
+
+ # prep request
+ req = Request("GET", url, params=payload, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def search_shared_print_lhr(
+ self,
+ oclcNumber: Optional[Union[int, str]] = None,
+ barcode: Optional[str] = None,
+ heldBySymbol: Optional[List[str]] = None,
+ heldByInstitutionID: Optional[List[int]] = None,
+ spProgram: Optional[List[str]] = None,
+ orderBy: Optional[str] = None,
+ offset: Optional[int] = None,
+ limit: Optional[int] = None,
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Search for shared print LHR Resources
+ Uses /search/retained-holdings endpoint.
+
+ Args:
+ oclcNumber: OCLC bibliographic record number; can be
+ an integer, or string that can include
+ OCLC # prefix
+ barcode: barcode as a string,
+ heldBySymbol: restricts to holdings with specified intitution
+ symbol
+ heldByInstitutionID: restrict to specified institution registryId
+ spProgram: restricts responses to bibliographic records
+ associated with particular shared print
+ program
+ orderBy: results sort key;
+ options:
+ 'commitmentExpirationDate'
+ 'location'
+ 'oclcSymbol'
+ default is 'oclcSymbol'
+ offset: start position of bibliographic records to
+ return; default is 1
+ limit: maximum number of records to return;
+ maximum is 50, default is 10
+ hooks: Requests library hook system that can be used for
+ signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` object
+ """
+ if oclcNumber is not None:
+ oclcNumber = verify_oclc_number(oclcNumber)
+
+ url = self._url_search_lhr_shared_print()
+ header = {"Accept": "application/json"}
+ payload = {
+ "oclcNumber": oclcNumber,
+ "barcode": barcode,
+ "heldBySymbol": heldBySymbol,
+ "heldByInstitutionID": heldByInstitutionID,
+ "spProgram": spProgram,
+ "orderBy": orderBy,
+ "offset": offset,
+ "limit": limit,
+ }
+
+ # prep request
+ req = Request("GET", url, params=payload, headers=header, hooks=hooks)
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
+
+ def validate_bib(
+ self,
+ record: Optional[str] = None,
+ recordFormat: Optional[str] = None,
+ validationLevel: str = "validateFull",
+ hooks: Optional[Dict[str, Callable]] = None,
+ ) -> Optional[Response]:
+ """
+ Given a bib record, validate that record conforms to MARC standards
+ Uses /manage/bibs/validate/{validationLevel} endpoint.
+
+ Args:
+ record: MARC record to be validated
+ recordFormat: format of MARC record, options:
+ 'application/marcxml+xml', 'application/marc'
+ validationLevel: Level at which to validate records
+ available values: 'validateFull', 'validateAdd',
+ 'validateReplace'
+ default is 'validateFull'
+ hooks: Requests library hook system that can be
+ used for signal event handling, see more at:
+ https://requests.readthedocs.io/en/master/user/advanced/#event-hooks
+ Returns:
+ `requests.Response` instance
+ """
+ if not record:
+ raise TypeError("Argument 'record' is missing.")
+
+ if not recordFormat:
+ raise TypeError("Argument 'recordFormat' is missing.")
+
+ url = self._url_manage_bibs_validate(validationLevel)
+ header = {
+ "Accept": "application/json",
+ "content-type": recordFormat,
+ }
+
+ # prep request
+ req = Request(
+ "POST",
+ url,
+ data=record,
+ headers=header,
+ hooks=hooks,
+ )
+ prepared_request = self.prepare_request(req)
+
+ # send request
+ query = Query(self, prepared_request, timeout=self.timeout)
+
+ return query.response
diff --git a/tests/conftest.py b/tests/conftest.py
index 72522d3..9225ace 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -23,6 +23,28 @@ def live_keys():
os.environ["WCScopes"] = data["scopes"]
+@pytest.fixture
+def stub_marc_xml():
+ stub_marc_xml = "00000nam a2200000 a 4500120827s2012 nyua 000 0 eng d 63011276 ocn850940548OCWMSengOCWMSOCLC Developer NetworkTest RecordFOR OCLC DEVELOPER NETWORK DOCUMENTATION"
+ return stub_marc_xml
+
+
+@pytest.fixture
+def stub_holding_xml():
+ stub_holding_xml = "00000nx a2200000zi 4500312010zu1103280p 0 4001uueng0210908OCWMSEASTEAST-STACKS879456"
+ return stub_holding_xml
+
+
+@pytest.fixture
+def stub_marc21():
+ fh = os.path.join(
+ os.environ["USERPROFILE"], "github/bookops-worldcat/temp/test.mrc"
+ )
+ with open(fh, "rb") as stub:
+ stub_marc21 = stub.read()
+ return stub_marc21
+
+
class FakeUtcNow(datetime.datetime):
@classmethod
def now(cls, tzinfo=datetime.timezone.utc):
diff --git a/tests/test_metadata_api.py b/tests/test_metadata_api.py
index a3d2236..3dd52e4 100644
--- a/tests/test_metadata_api.py
+++ b/tests/test_metadata_api.py
@@ -149,6 +149,150 @@ def test_url_manage_ih_current(self, stub_session):
== "https://metadata.api.oclc.org/worldcat/manage/institution/holdings/current"
)
+ @pytest.mark.parametrize(
+ "validationLevel",
+ ["vaidateFull", "validateAdd", "validateReplace"],
+ )
+ def test_url_manage_bibs_validate(self, validationLevel, stub_session):
+ assert (
+ stub_session._url_manage_bibs_validate(validationLevel)
+ == f"https://metadata.api.oclc.org/worldcat/manage/bibs/validate/{validationLevel}"
+ )
+
+ def test_url_manage_bibs_current_oclc_number(self, stub_session):
+ assert (
+ stub_session._url_manage_bibs_current_oclc_number()
+ == "https://metadata.api.oclc.org/worldcat/manage/bibs/current"
+ )
+
+ def test_url_manage_bibs_create(self, stub_session):
+ assert (
+ stub_session._url_manage_bibs_create()
+ == "https://metadata.api.oclc.org/worldcat/manage/bibs"
+ )
+
+ def test_url_manage_bibs(self, stub_session):
+ assert (
+ stub_session._url_manage_bibs(oclcNumber="12345")
+ == "https://metadata.api.oclc.org/worldcat/manage/bibs/12345"
+ )
+
+ def test_url_manage_bibs_match(self, stub_session):
+ assert (
+ stub_session._url_manage_bibs_match()
+ == "https://metadata.api.oclc.org/worldcat/manage/bibs/match"
+ )
+
+ def test_url_manage_ih_set_on_record(self, stub_session):
+ assert (
+ stub_session._url_manage_ih_set_on_record()
+ == "https://metadata.api.oclc.org/worldcat/manage/institution/holdings/set"
+ )
+
+ def test_url_manage_ih_unset_on_record(self, stub_session):
+ assert (
+ stub_session._url_manage_ih_unset_on_record()
+ == "https://metadata.api.oclc.org/worldcat/manage/institution/holdings/unset"
+ )
+
+ def test_url_manage_ih_codes(self, stub_session):
+ assert (
+ stub_session._url_manage_ih_codes()
+ == "https://metadata.api.oclc.org/worldcat/manage/institution/holding-codes"
+ )
+
+ def test_url_manage_lbd_create(self, stub_session):
+ assert (
+ stub_session._url_manage_lbd_create()
+ == "https://metadata.api.oclc.org/worldcat/manage/lbds"
+ )
+
+ @pytest.mark.parametrize(
+ "controlNumber",
+ ["12345", 12345],
+ )
+ def test_url_manage_lbd(self, controlNumber, stub_session):
+ assert (
+ stub_session._url_manage_lbd(controlNumber)
+ == f"https://metadata.api.oclc.org/worldcat/manage/lbds/{controlNumber}"
+ )
+
+ def test_url_manage_lhr_create(self, stub_session):
+ assert (
+ stub_session._url_manage_lhr_create()
+ == "https://metadata.api.oclc.org/worldcat/manage/lhrs"
+ )
+
+ @pytest.mark.parametrize(
+ "controlNumber",
+ ["12345", 12345],
+ )
+ def test_url_manage_lhr(self, controlNumber, stub_session):
+ assert (
+ stub_session._url_manage_lhr(controlNumber)
+ == f"https://metadata.api.oclc.org/worldcat/manage/lhrs/{controlNumber}"
+ )
+
+ def test_url_search_general_holdings_summary(self, stub_session):
+ assert (
+ stub_session._url_search_general_holdings_summary()
+ == "https://metadata.api.oclc.org/worldcat/search/summary-holdings"
+ )
+
+ @pytest.mark.parametrize(
+ "oclcNumber",
+ ["850940461", "850940463", 850940467],
+ )
+ def test_url_search_classification_bibs(self, oclcNumber, stub_session):
+ assert (
+ stub_session._url_search_classification_bibs(oclcNumber)
+ == f"https://metadata.api.oclc.org/worldcat/search/classification-bibs/{oclcNumber}"
+ )
+
+ def test_url_search_lhr_shared_print(self, stub_session):
+ assert (
+ stub_session._url_search_lhr_shared_print()
+ == "https://metadata.api.oclc.org/worldcat/search/retained-holdings"
+ )
+
+ @pytest.mark.parametrize(
+ "controlNumber",
+ ["12345", 12345],
+ )
+ def test_url_search_lhr_control_number(self, controlNumber, stub_session):
+ assert (
+ stub_session._url_search_lhr_control_number(controlNumber)
+ == f"https://metadata.api.oclc.org/worldcat/search/my-holdings/{controlNumber}"
+ )
+
+ def test_url_search_lhr(self, stub_session):
+ assert (
+ stub_session._url_search_lhr()
+ == "https://metadata.api.oclc.org/worldcat/search/my-holdings"
+ )
+
+ def test_url_browse_lhr(self, stub_session):
+ assert (
+ stub_session._url_browse_lhr()
+ == "https://metadata.api.oclc.org/worldcat/browse/my-holdings"
+ )
+
+ @pytest.mark.parametrize(
+ "controlNumber",
+ ["12345", 12345],
+ )
+ def test_url_search_lbd_control_number(self, controlNumber, stub_session):
+ assert (
+ stub_session._url_search_lbd_control_number(controlNumber)
+ == f"https://metadata.api.oclc.org/worldcat/search/my-local-bib-data/{controlNumber}"
+ )
+
+ def test_url_search_lbd(self, stub_session):
+ assert (
+ stub_session._url_search_lbd()
+ == "https://metadata.api.oclc.org/worldcat/search/my-local-bib-data"
+ )
+
@pytest.mark.http_code(200)
def test_get_brief_bib(self, stub_session, mock_session_response):
assert stub_session.get_brief_bib(12345).status_code == 200
@@ -191,6 +335,10 @@ def test_get_full_bib_None_oclcNumber_passed(self, stub_session):
with pytest.raises(InvalidOclcNumber):
stub_session.get_full_bib(oclcNumber=None)
+ @pytest.mark.http_code(200)
+ def test_get_institution_holding_codes(self, stub_session, mock_session_response):
+ assert stub_session.get_institution_holding_codes().status_code == 200
+
@pytest.mark.http_code(200)
def test_get_institution_holdings(self, stub_session, mock_session_response):
assert stub_session.get_institution_holdings("12345")[0].status_code == 200
@@ -311,6 +459,368 @@ def test_search_shared_print_holdings_with_invalid_oclc_number_passsed(
stub_session.search_shared_print_holdings(oclcNumber="odn12345")
assert msg in str(exc.value)
+ @pytest.mark.http_code(200)
+ def test_browse_my_holdings(self, stub_session, mock_session_response):
+ assert (
+ stub_session.browse_my_holdings(
+ oclcNumber="12345", holdingLocation="foo", shelvingLocation="bar"
+ ).status_code
+ == 200
+ )
+
+ @pytest.mark.http_code(200)
+ def test_browse_my_holdings_no_oclc_number(
+ self, stub_session, mock_session_response
+ ):
+ assert (
+ stub_session.browse_my_holdings(
+ holdingLocation="foo", shelvingLocation="bar"
+ ).status_code
+ == 200
+ )
+
+ def test_browse_my_holdings_no_holdingLocation_passed(self, stub_session):
+ with pytest.raises(TypeError) as exc:
+ stub_session.browse_my_holdings(shelvingLocation="bar")
+ assert "Argument 'holdingLocation' is missing." in str(exc.value)
+
+ def test_browse_my_holdings_no_shelvingLocation_passed(self, stub_session):
+ with pytest.raises(TypeError) as exc:
+ stub_session.browse_my_holdings(holdingLocation="foo")
+ assert "Argument 'shelvingLocation' is missing." in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_create_bib(self, stub_session, mock_session_response, stub_marc_xml):
+ assert (
+ stub_session.create_bib(
+ stub_marc_xml, recordFormat="application/marcxml+xml"
+ ).status_code
+ == 200
+ )
+
+ def test_create_bib_no_record_passed(self, stub_session):
+ with pytest.raises(TypeError) as exc:
+ stub_session.create_bib(recordFormat="application/marcxml+xml")
+ assert "Argument 'record' is missing." in str(exc.value)
+
+ def test_create_bib_no_recordFormat_passed(
+ self, stub_session, mock_session_response, stub_marc_xml
+ ):
+ with pytest.raises(TypeError) as exc:
+ stub_session.create_bib(stub_marc_xml)
+ assert "Argument 'recordFormat' is missing." in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_create_lbd(self, stub_session, mock_session_response, stub_marc_xml):
+ assert (
+ stub_session.create_lbd(
+ stub_marc_xml, recordFormat="application/marcxml+xml"
+ ).status_code
+ == 200
+ )
+
+ def test_create_lbd_no_record_passed(self, stub_session):
+ with pytest.raises(TypeError) as exc:
+ stub_session.create_lbd(recordFormat="application/marcxml+xml")
+ assert "Argument 'record' is missing." in str(exc.value)
+
+ def test_create_lbd_no_recordFormat_passed(
+ self, stub_session, mock_session_response, stub_marc_xml
+ ):
+ with pytest.raises(TypeError) as exc:
+ stub_session.create_lbd(stub_marc_xml)
+ assert "Argument 'recordFormat' is missing." in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_create_lhr(self, stub_session, mock_session_response, stub_holding_xml):
+ assert (
+ stub_session.create_lhr(
+ stub_holding_xml, recordFormat="application/marcxml+xml"
+ ).status_code
+ == 200
+ )
+
+ def test_create_lhr_no_record_passed(self, stub_session):
+ with pytest.raises(TypeError) as exc:
+ stub_session.create_lhr(recordFormat="application/marcxml+xml")
+ assert "Argument 'record' is missing." in str(exc.value)
+
+ def test_create_lhr_no_recordFormat_passed(
+ self, stub_session, mock_session_response, stub_holding_xml
+ ):
+ with pytest.raises(TypeError) as exc:
+ stub_session.create_lhr(stub_holding_xml)
+ assert "Argument 'recordFormat' is missing." in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_delete_lbd(self, stub_session, mock_session_response):
+ assert stub_session.delete_lbd("12345").status_code == 200
+
+ @pytest.mark.http_code(200)
+ def test_delete_lhr(self, stub_session, mock_session_response):
+ assert stub_session.delete_lhr("12345").status_code == 200
+
+ @pytest.mark.http_code(200)
+ def test_get_bib_classification(self, stub_session, mock_session_response):
+ assert stub_session.get_bib_classification(12345).status_code == 200
+
+ def test_get_bib_classification_no_oclcNumber_passed(self, stub_session):
+ with pytest.raises(TypeError):
+ stub_session.get_bib_classification()
+
+ @pytest.mark.http_code(200)
+ def test_get_bib_holdings(self, stub_session, mock_session_response):
+ assert stub_session.get_bib_holdings(oclcNumber=12345).status_code == 200
+
+ def test_get_bib_holdings_no_oclcNumber_passed(self, stub_session):
+ with pytest.raises(TypeError):
+ stub_session.get_bib_holdings(holdingsAllVariantRecords=True)
+
+ @pytest.mark.http_code(200)
+ def test_get_lbd_data(self, stub_session, mock_session_response):
+ assert stub_session.get_lbd_data(12345).status_code == 200
+
+ def test_get_lbd_data_no_controlNumber_passed(self, stub_session):
+ with pytest.raises(TypeError):
+ stub_session.get_lbd_data()
+
+ @pytest.mark.http_code(200)
+ def test_get_lhr_data(self, stub_session, mock_session_response):
+ assert stub_session.get_lhr_data(12345).status_code == 200
+
+ def test_get_lhr_data_no_controlNumber_passed(self, stub_session):
+ with pytest.raises(TypeError):
+ stub_session.get_lhr_data()
+
+ @pytest.mark.http_code(200)
+ def test_get_lbd_record(self, stub_session, mock_session_response):
+ assert stub_session.get_lbd_record(12345).status_code == 200
+
+ def test_get_lbd_record_no_controlNumber_passed(self, stub_session):
+ with pytest.raises(TypeError):
+ stub_session.get_lbd_record()
+
+ @pytest.mark.http_code(200)
+ def test_get_lhr_record(self, stub_session, mock_session_response):
+ assert stub_session.get_lhr_record(12345).status_code == 200
+
+ def test_get_lhr_record_no_controlNumber_passed(self, stub_session):
+ with pytest.raises(TypeError):
+ stub_session.get_lhr_record()
+
+ @pytest.mark.http_code(200)
+ def test_holding_set_on_record(
+ self, stub_session, mock_session_response, stub_marc_xml
+ ):
+ assert (
+ stub_session.holding_set_on_record(
+ stub_marc_xml, recordFormat="application/marcxml+xml"
+ ).status_code
+ == 200
+ )
+
+ def test_holding_set_on_record_no_record_passed(self, stub_session):
+ with pytest.raises(TypeError) as exc:
+ stub_session.holding_set_on_record(recordFormat="application/marcxml+xml")
+ assert "Argument 'record' is missing." in str(exc.value)
+
+ def test_holding_set_on_record_no_recordFormat_passed(
+ self, stub_session, mock_session_response, stub_marc_xml
+ ):
+ with pytest.raises(TypeError) as exc:
+ stub_session.holding_set_on_record(stub_marc_xml)
+ assert "Argument 'recordFormat' is missing." in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_holding_unset_on_record(
+ self, stub_session, mock_session_response, stub_marc_xml
+ ):
+ assert (
+ stub_session.holding_unset_on_record(
+ record=stub_marc_xml, recordFormat="application/marcxml+xml"
+ ).status_code
+ == 200
+ )
+
+ def test_holding_unset_on_record_no_record_passed(self, stub_session):
+ with pytest.raises(TypeError) as exc:
+ stub_session.holding_unset_on_record(recordFormat="application/marcxml+xml")
+ assert "Argument 'record' is missing." in str(exc.value)
+
+ def test_holding_unset_on_record_no_recordFormat_passed(
+ self, stub_session, mock_session_response, stub_marc_xml
+ ):
+ with pytest.raises(TypeError) as exc:
+ stub_session.holding_unset_on_record(stub_marc_xml)
+ assert "Argument 'recordFormat' is missing." in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_match_bib(self, stub_session, mock_session_response, stub_marc_xml):
+ assert (
+ stub_session.match_bib(
+ stub_marc_xml, recordFormat="application/marcxml+xml"
+ ).status_code
+ == 200
+ )
+
+ def test_match_bib_no_record_passed(self, stub_session):
+ with pytest.raises(TypeError) as exc:
+ stub_session.match_bib(recordFormat="application/marcxml+xml")
+ assert "Argument 'record' is missing." in str(exc.value)
+
+ def test_match_bib_no_recordFormat_passed(
+ self, stub_session, mock_session_response, stub_marc_xml
+ ):
+ with pytest.raises(TypeError) as exc:
+ stub_session.match_bib(stub_marc_xml)
+ assert "Argument 'recordFormat' is missing." in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_replace_bib(self, stub_session, mock_session_response, stub_marc_xml):
+ assert (
+ stub_session.replace_bib(
+ "12345", stub_marc_xml, recordFormat="application/marcxml+xml"
+ ).status_code
+ == 200
+ )
+
+ @pytest.mark.http_code(200)
+ def test_replace_bib_no_record_passed(self, stub_session):
+ with pytest.raises(TypeError) as exc:
+ stub_session.replace_bib("12345", recordFormat="application/marcxml+xml")
+ assert "Argument 'record' is missing." in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_replace_bib_no_recordFormat_passed(
+ self, stub_session, mock_session_response, stub_marc_xml
+ ):
+ with pytest.raises(TypeError) as exc:
+ stub_session.replace_bib("12345", stub_marc_xml)
+ assert "Argument 'recordFormat' is missing." in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_replace_lbd(self, stub_session, mock_session_response, stub_marc_xml):
+ assert (
+ stub_session.replace_lbd(
+ "12345", stub_marc_xml, recordFormat="application/marcxml+xml"
+ ).status_code
+ == 200
+ )
+
+ @pytest.mark.http_code(200)
+ def test_replace_lbd_no_record_passed(self, stub_session):
+ with pytest.raises(TypeError) as exc:
+ stub_session.replace_lbd("12345", recordFormat="application/marcxml+xml")
+ assert "Argument 'record' is missing." in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_replace_lbd_no_recordFormat_passed(
+ self, stub_session, mock_session_response, stub_marc_xml
+ ):
+ with pytest.raises(TypeError) as exc:
+ stub_session.replace_lbd("12345", stub_marc_xml)
+ assert "Argument 'recordFormat' is missing." in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_replace_lhr(self, stub_session, mock_session_response, stub_holding_xml):
+ assert (
+ stub_session.replace_lhr(
+ "12345", stub_holding_xml, recordFormat="application/marcxml+xml"
+ ).status_code
+ == 200
+ )
+
+ @pytest.mark.http_code(200)
+ def test_replace_lhr_no_record_passed(self, stub_session):
+ with pytest.raises(TypeError) as exc:
+ stub_session.replace_lhr("12345", recordFormat="application/marcxml+xml")
+ assert "Argument 'record' is missing." in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_replace_lhr_no_recordFormat_passed(
+ self, stub_session, mock_session_response, stub_holding_xml
+ ):
+ with pytest.raises(TypeError) as exc:
+ stub_session.replace_lhr("12345", stub_holding_xml)
+ assert "Argument 'recordFormat' is missing." in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_search_my_holdings(self, stub_session, mock_session_response):
+ assert stub_session.search_my_holdings(oclcNumber=12345).status_code == 200
+
+ def test_search_my_holdings_no_oclcNumber_passed(
+ self, stub_session, mock_session_response
+ ):
+ assert stub_session.search_my_holdings(barcode=12345).status_code == 200
+
+ def test_search_my_holdings_invalid_oclc_number(self, stub_session):
+ msg = "Argument 'oclcNumber' does not look like real OCLC #."
+ with pytest.raises(InvalidOclcNumber) as exc:
+ stub_session.search_my_holdings(oclcNumber="odn12345")
+ assert msg in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_search_my_local_bibs(self, stub_session, mock_session_response):
+ assert stub_session.search_my_local_bibs(q="ti:foo").status_code == 200
+
+ @pytest.mark.parametrize("argm", [(None), ("")])
+ def test_search_my_local_bibs_missing_query(self, stub_session, argm):
+ with pytest.raises(TypeError) as exc:
+ stub_session.search_my_local_bibs(argm)
+ assert "Argument 'q' is requried to construct query." in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_search_shared_print_lhr(self, stub_session, mock_session_response):
+ assert stub_session.search_shared_print_lhr(oclcNumber=12345).status_code == 200
+
+ @pytest.mark.http_code(200)
+ def test_search_shared_print_lhr_no_oclcNumber_passed(
+ self, stub_session, mock_session_response
+ ):
+ assert stub_session.search_shared_print_lhr(barcode=12345).status_code == 200
+
+ def test_search_shared_print_lhr_with_invalid_oclc_number_passsed(
+ self, stub_session
+ ):
+ msg = "Argument 'oclcNumber' does not look like real OCLC #."
+ with pytest.raises(InvalidOclcNumber) as exc:
+ stub_session.search_shared_print_lhr(oclcNumber="odn12345")
+ assert msg in str(exc.value)
+
+ @pytest.mark.http_code(200)
+ def test_validate_bib(self, stub_session, mock_session_response, stub_marc_xml):
+ assert (
+ stub_session.validate_bib(
+ stub_marc_xml,
+ recordFormat="application/marcxml+xml",
+ validationLevel="validateFull",
+ ).status_code
+ == 200
+ )
+
+ @pytest.mark.http_code(200)
+ def test_validate_bib_default(
+ self, stub_session, mock_session_response, stub_marc_xml
+ ):
+ assert (
+ stub_session.validate_bib(
+ stub_marc_xml, recordFormat="application/marcxml+xml"
+ ).status_code
+ == 200
+ )
+
+ def test_validate_bib_no_record_passed(self, stub_session):
+ with pytest.raises(TypeError) as exc:
+ stub_session.validate_bib(recordFormat="application/marcxml+xml")
+ assert "Argument 'record' is missing." in str(exc.value)
+
+ def test_validate_bib_no_recordFormat_passed(
+ self, stub_session, mock_session_response, stub_marc_xml
+ ):
+ with pytest.raises(TypeError) as exc:
+ stub_session.validate_bib(stub_marc_xml)
+ assert "Argument 'recordFormat' is missing." in str(exc.value)
+
@pytest.mark.webtest
class TestLiveMetadataSession:
@@ -509,3 +1019,178 @@ def test_get_current_oclc_number(self, live_keys):
jres = response.json()
assert sorted(jres.keys()) == ["controlNumbers"]
assert sorted(jres["controlNumbers"][0].keys()) == ["current", "requested"]
+
+ def test_get_bib_classification(self, live_keys):
+ token = WorldcatAccessToken(
+ key=os.getenv("WCKey"),
+ secret=os.getenv("WCSecret"),
+ scopes=os.getenv("WCScopes"),
+ )
+
+ with MetadataSession(authorization=token) as session:
+ response = session.get_bib_classification(41266045)
+
+ assert (
+ response.url
+ == "https://metadata.api.oclc.org/worldcat/search/classification-bibs/41266045"
+ )
+ assert response.status_code == 200
+ assert sorted(response.json().keys()) == [
+ "dewey",
+ "lc",
+ ]
+
+ def test_get_current_oclc_number_str(self, live_keys):
+ token = WorldcatAccessToken(
+ key=os.getenv("WCKey"),
+ secret=os.getenv("WCSecret"),
+ scopes=os.getenv("WCScopes"),
+ )
+
+ with MetadataSession(authorization=token) as session:
+ response = session.get_current_oclc_number("41266045")
+
+ assert response.status_code == 200
+ assert (
+ response.request.url
+ == "https://metadata.api.oclc.org/worldcat/manage/bibs/current?oclcNumbers=41266045"
+ )
+ jres = response.json()
+ assert sorted(jres.keys()) == ["controlNumbers"]
+ assert sorted(jres["controlNumbers"][0].keys()) == ["current", "requested"]
+
+ @pytest.mark.holdings
+ def test_get_institution_holding_codes(self, live_keys):
+ token = WorldcatAccessToken(
+ key=os.getenv("WCKey"),
+ secret=os.getenv("WCSecret"),
+ scopes=os.getenv("WCScopes"),
+ )
+
+ with MetadataSession(authorization=token) as session:
+ response = session.get_institution_holding_codes()
+
+ assert (
+ response.url
+ == "https://metadata.api.oclc.org/worldcat/manage/institution/holding-codes"
+ )
+ assert response.status_code == 200
+ assert sorted(response.json().keys()) == ["holdingLibraryCodes"]
+ assert {"code": "Print Collection", "name": "NYPC"} in response.json()[
+ "holdingLibraryCodes"
+ ]
+
+ @pytest.mark.holdings
+ def test_holding_set_unset_marcxml(self, live_keys, stub_marc_xml):
+ token = WorldcatAccessToken(
+ key=os.getenv("WCKey"),
+ secret=os.getenv("WCSecret"),
+ scopes=os.getenv("WCScopes"),
+ )
+
+ with MetadataSession(authorization=token) as session:
+ response = session.get_institution_holdings("850940548")[0]
+ holdings = response.json()["holdings"]
+
+ # make sure no holdings are set initially
+ if len(holdings) > 0:
+ response = session.holding_unset_on_record(
+ stub_marc_xml, recordFormat="application/marcxml+xml"
+ )
+
+ response = session.holding_set_on_record(
+ stub_marc_xml, recordFormat="application/marcxml+xml"
+ )
+ assert (
+ response.url
+ == "https://metadata.api.oclc.org/worldcat/manage/institution/holdings/set"
+ )
+ assert response.status_code == 200
+ assert response.json()["action"] == "Set Holdings"
+
+ response = session.holding_unset_on_record(
+ stub_marc_xml, recordFormat="application/marcxml+xml"
+ )
+ assert response.status_code == 200
+ assert (
+ response.request.url
+ == "https://metadata.api.oclc.org/worldcat/manage/institution/holdings/unset"
+ )
+ assert response.json()["action"] == "Unset Holdings"
+
+ def test_match_bib_marcxml(self, live_keys, stub_marc_xml):
+ token = WorldcatAccessToken(
+ key=os.getenv("WCKey"),
+ secret=os.getenv("WCSecret"),
+ scopes=os.getenv("WCScopes"),
+ )
+
+ with MetadataSession(authorization=token) as session:
+ response = session.match_bib(
+ stub_marc_xml, recordFormat="application/marcxml+xml"
+ )
+ assert response.status_code == 200
+ assert sorted(response.json().keys()) == sorted(
+ ["numberOfRecords", "briefRecords"]
+ )
+
+ @pytest.mark.holdings
+ def test_search_bibs_holdings_oclc(self, live_keys):
+ fields = sorted(["briefRecords", "numberOfRecords"])
+ token = WorldcatAccessToken(
+ key=os.getenv("WCKey"),
+ secret=os.getenv("WCSecret"),
+ scopes=os.getenv("WCScopes"),
+ )
+
+ with MetadataSession(authorization=token) as session:
+ response = session.search_bibs_holdings(oclcNumber="41266045")
+
+ assert response.status_code == 200
+ assert sorted(response.json().keys()) == fields
+
+ def test_search_bibs_holdings_isbn(self, live_keys):
+ fields = sorted(["briefRecords", "numberOfRecords"])
+ token = WorldcatAccessToken(
+ key=os.getenv("WCKey"),
+ secret=os.getenv("WCSecret"),
+ scopes=os.getenv("WCScopes"),
+ )
+
+ with MetadataSession(authorization=token) as session:
+ response = session.search_bibs_holdings(isbn="9781597801744")
+
+ assert response.status_code == 200
+ assert sorted(response.json().keys()) == fields
+
+ def test_search_brief_bibs_other_editions(self, live_keys):
+ fields = sorted(["briefRecords", "numberOfRecords"])
+ token = WorldcatAccessToken(
+ key=os.getenv("WCKey"),
+ secret=os.getenv("WCSecret"),
+ scopes=os.getenv("WCScopes"),
+ )
+
+ with MetadataSession(authorization=token) as session:
+ response = session.search_brief_bibs_other_editions(41266045)
+
+ assert response.status_code == 200
+ assert sorted(response.json().keys()) == fields
+
+ def test_validate_bib(self, live_keys, stub_marc21):
+ token = WorldcatAccessToken(
+ key=os.getenv("WCKey"),
+ secret=os.getenv("WCSecret"),
+ scopes=os.getenv("WCScopes"),
+ )
+
+ with MetadataSession(authorization=token) as session:
+ response = session.validate_bib(
+ stub_marc21, recordFormat="application/marc"
+ )
+ assert response.status_code == 200
+ assert (
+ response.url
+ == "https://metadata.api.oclc.org/worldcat/manage/bibs/validate/validateFull"
+ )
+ assert sorted(response.json().keys()) == sorted(["httpStatus", "status"])