Source code for globus_sdk.services.search.client

import logging
from typing import Any, Dict, Optional, Union

from globus_sdk import client, paging, response, utils
from globus_sdk.scopes import SearchScopes
from globus_sdk.types import UUIDLike

from .data import SearchQuery
from .errors import SearchAPIError

log = logging.getLogger(__name__)


[docs]class SearchClient(client.BaseClient): r""" Client for the Globus Search API This class provides helper methods for most common resources in the API, and basic ``get``, ``put``, ``post``, and ``delete`` methods from the base client that can be used to access any API resource. :param authorizer: An authorizer instance used for all calls to Globus Search :type authorizer: :class:`GlobusAuthorizer \ <globus_sdk.authorizers.base.GlobusAuthorizer>` **Methods** .. automethodlist:: globus_sdk.SearchClient """ error_class = SearchAPIError service_name = "search" scopes = SearchScopes # # Index Management #
[docs] @utils.doc_api_method("Get Index Metadata", "search/reference/index_show/") def get_index( self, index_id: UUIDLike, *, query_params: Optional[Dict[str, Any]] = None ) -> response.GlobusHTTPResponse: """ ``GET /v1/index/<index_id>`` **Examples** >>> sc = globus_sdk.SearchClient(...) >>> index = sc.get_index(index_id) >>> assert index['index_id'] == index_id >>> print(index["display_name"], >>> "(" + index_id + "):", >>> index["description"]) """ log.info(f"SearchClient.get_index({index_id})") return self.get(f"/v1/index/{index_id}", query_params=query_params)
# # Search queries #
[docs] @utils.doc_api_method("GET Search Query", "search/reference/get_query/") @paging.has_paginator( paging.HasNextPaginator, items_key="gmeta", get_page_size=lambda x: x["count"], max_total_results=10000, page_size=100, ) def search( self, index_id: UUIDLike, q: str, *, offset: int = 0, limit: int = 10, advanced: bool = False, query_params: Optional[Dict[str, Any]] = None, ) -> response.GlobusHTTPResponse: """ ``GET /v1/index/<index_id>/search`` **Examples** >>> sc = globus_sdk.SearchClient(...) >>> result = sc.search(index_id, 'query string') >>> advanced_result = sc.search(index_id, 'author: "Ada Lovelace"', >>> advanced=True) """ if query_params is None: query_params = {} query_params.update( { "q": q, "offset": offset, "limit": limit, "advanced": advanced, } ) log.info(f"SearchClient.search({index_id}, ...)") return self.get(f"/v1/index/{index_id}/search", query_params=query_params)
# # Bulk data indexing #
[docs] @utils.doc_api_method("Ingest", "search/reference/ingest") def ingest( self, index_id: UUIDLike, data: Dict[str, Any] ) -> response.GlobusHTTPResponse: """ ``POST /v1/index/<index_id>/ingest`` **Examples** >>> sc = globus_sdk.SearchClient(...) >>> ingest_data = { >>> "ingest_type": "GMetaEntry", >>> "ingest_data": { >>> "subject": "https://example.com/foo/bar", >>> "visible_to": ["public"], >>> "content": { >>> "foo/bar": "some val" >>> } >>> } >>> } >>> sc.ingest(index_id, ingest_data) or with multiple entries at once via a GMetaList: >>> sc = globus_sdk.SearchClient(...) >>> ingest_data = { >>> "ingest_type": "GMetaList", >>> "ingest_data": { >>> "gmeta": [ >>> { >>> "subject": "https://example.com/foo/bar", >>> "visible_to": ["public"], >>> "content": { >>> "foo/bar": "some val" >>> } >>> }, >>> { >>> "subject": "https://example.com/foo/bar", >>> "id": "otherentry", >>> "visible_to": ["public"], >>> "content": { >>> "foo/bar": "some otherval" >>> } >>> } >>> ] >>> } >>> } >>> sc.ingest(index_id, ingest_data) """ log.info(f"SearchClient.ingest({index_id}, ...)") return self.post(f"/v1/index/{index_id}/ingest", data=data)
# # Bulk delete #
[docs] @utils.doc_api_method("Delete By Query", "search/reference/delete_by_query") def delete_by_query( self, index_id: UUIDLike, data: Dict[str, Any] ) -> response.GlobusHTTPResponse: """ ``POST /v1/index/<index_id>/delete_by_query`` **Examples** >>> sc = globus_sdk.SearchClient(...) >>> query_data = { >>> "q": "user query", >>> "filters": [ >>> { >>> "type": "range", >>> "field_name": "path.to.date", >>> "values": [ >>> {"from": "*", >>> "to": "2014-11-07"} >>> ] >>> } >>> ] >>> } >>> sc.delete_by_query(index_id, query_data) """ log.info(f"SearchClient.delete_by_query({index_id}, ...)") return self.post(f"/v1/index/{index_id}/delete_by_query", data=data)
# # Subject Operations #
[docs] @utils.doc_api_method("Get Subject", "search/reference/get_subject") def get_subject( self, index_id: UUIDLike, subject: str, *, query_params: Optional[Dict[str, Any]] = None, ) -> response.GlobusHTTPResponse: """ ``GET /v1/index/<index_id>/subject`` **Examples** Fetch the data for subject ``http://example.com/abc`` from index ``index_id``: >>> sc = globus_sdk.SearchClient(...) >>> subject_data = sc.get_subject(index_id, 'http://example.com/abc') """ if query_params is None: query_params = {} query_params["subject"] = subject log.info(f"SearchClient.get_subject({index_id}, {subject}, ...)") return self.get(f"/v1/index/{index_id}/subject", query_params=query_params)
[docs] @utils.doc_api_method("Delete Subject", "search/reference/delete_subject") def delete_subject( self, index_id: UUIDLike, subject: str, *, query_params: Optional[Dict[str, Any]] = None, ) -> response.GlobusHTTPResponse: """ ``DELETE /v1/index/<index_id>/subject`` **Examples** Delete all data for subject ``http://example.com/abc`` from index ``index_id``, even data which is not visible to the current user: >>> sc = globus_sdk.SearchClient(...) >>> subject_data = sc.get_subject(index_id, 'http://example.com/abc') """ if query_params is None: query_params = {} query_params["subject"] = subject log.info(f"SearchClient.delete_subject({index_id}, {subject}, ...)") return self.delete(f"/v1/index/{index_id}/subject", query_params=query_params)
# # Entry Operations #
[docs] @utils.doc_api_method("Get Entry", "search/reference/get_entry") def get_entry( self, index_id: UUIDLike, subject: str, *, entry_id: Optional[str] = None, query_params: Optional[Dict[str, Any]] = None, ) -> response.GlobusHTTPResponse: """ ``GET /v1/index/<index_id>/entry`` **Examples** Lookup the entry with a subject of ``https://example.com/foo/bar`` and a null entry_id: >>> sc = globus_sdk.SearchClient(...) >>> entry_data = sc.get_entry(index_id, 'http://example.com/foo/bar') Lookup the entry with a subject of ``https://example.com/foo/bar`` and an entry_id of ``foo/bar``: >>> sc = globus_sdk.SearchClient(...) >>> entry_data = sc.get_entry(index_id, 'http://example.com/foo/bar', >>> entry_id='foo/bar') """ if query_params is None: query_params = {} query_params["subject"] = subject if entry_id is not None: query_params["entry_id"] = entry_id log.info( "SearchClient.get_entry({}, {}, {}, ...)".format( index_id, subject, entry_id ) ) return self.get(f"/v1/index/{index_id}/entry", query_params=query_params)
[docs] @utils.doc_api_method("Create Entry", "search/reference/create_or_update_entry") def create_entry( self, index_id: UUIDLike, data: Dict[str, Any] ) -> response.GlobusHTTPResponse: """ ``POST /v1/index/<index_id>/entry`` **Examples** Create an entry with a subject of ``https://example.com/foo/bar`` and a null entry_id: >>> sc = globus_sdk.SearchClient(...) >>> sc.create_entry(index_id, { >>> "subject": "https://example.com/foo/bar", >>> "visible_to": ["public"], >>> "content": { >>> "foo/bar": "some val" >>> } >>> }) Create an entry with a subject of ``https://example.com/foo/bar`` and an entry_id of ``foo/bar``: >>> sc = globus_sdk.SearchClient(...) >>> sc.create_entry(index_id, { >>> "subject": "https://example.com/foo/bar", >>> "visible_to": ["public"], >>> "id": "foo/bar", >>> "content": { >>> "foo/bar": "some val" >>> } >>> }) """ log.info(f"SearchClient.create_entry({index_id}, ...)") return self.post(f"/v1/index/{index_id}/entry", data=data)
[docs] @utils.doc_api_method("Update Entry", "search/reference/create_or_update_entry") def update_entry( self, index_id: UUIDLike, data: Dict[str, Any] ) -> response.GlobusHTTPResponse: """ ``PUT /v1/index/<index_id>/entry`` **Examples** Update an entry with a subject of ``https://example.com/foo/bar`` and a null entry_id: >>> sc = globus_sdk.SearchClient(...) >>> sc.update_entry(index_id, { >>> "subject": "https://example.com/foo/bar", >>> "visible_to": ["public"], >>> "content": { >>> "foo/bar": "some val" >>> } >>> }) """ log.info(f"SearchClient.update_entry({index_id}, ...)") return self.put(f"/v1/index/{index_id}/entry", data=data)
[docs] @utils.doc_api_method("Delete Entry", "search/reference/delete_entry") def delete_entry( self, index_id: UUIDLike, subject: str, *, entry_id: Optional[str] = None, query_params: Optional[Dict[str, Any]] = None, ) -> response.GlobusHTTPResponse: """ ``DELETE /v1/index/<index_id>/entry`` **Examples** Delete an entry with a subject of ``https://example.com/foo/bar`` and a null entry_id: >>> sc = globus_sdk.SearchClient(...) >>> sc.delete_entry(index_id, "https://example.com/foo/bar") Delete an entry with a subject of ``https://example.com/foo/bar`` and an entry_id of "foo/bar": >>> sc = globus_sdk.SearchClient(...) >>> sc.delete_entry(index_id, "https://example.com/foo/bar", >>> entry_id="foo/bar") """ if query_params is None: query_params = {} query_params["subject"] = subject if entry_id is not None: query_params["entry_id"] = entry_id log.info( "SearchClient.delete_entry({}, {}, {}, ...)".format( index_id, subject, entry_id ) ) return self.delete(f"/v1/index/{index_id}/entry", query_params=query_params)
# # Task Management #
[docs] @utils.doc_api_method("Get Task", "search/reference/get_task") def get_task( self, task_id: UUIDLike, *, query_params: Optional[Dict[str, Any]] = None ) -> response.GlobusHTTPResponse: """ ``GET /v1/task/<task_id>`` **Examples** >>> sc = globus_sdk.SearchClient(...) >>> task = sc.get_task(task_id) >>> assert task['index_id'] == known_index_id >>> print(task["task_id"] + " | " + task['state']) """ log.info(f"SearchClient.get_task({task_id})") return self.get(f"/v1/task/{task_id}", query_params=query_params)
[docs] @utils.doc_api_method("Task List", "search/reference/task_list") def get_task_list( self, index_id: UUIDLike, *, query_params: Optional[Dict[str, Any]] = None ) -> response.GlobusHTTPResponse: """ ``GET /v1/task_list/<index_id>`` **Examples** >>> sc = globus_sdk.SearchClient(...) >>> task_list = sc.get_task_list(index_id) >>> for task in task_list['tasks']: >>> print(task["task_id"] + " | " + task['state']) """ log.info(f"SearchClient.get_task_list({index_id})") return self.get(f"/v1/task_list/{index_id}", query_params=query_params)