Source code for parasolr.solr.admin

from typing import Any, Optional
from urllib.parse import urljoin

from attrdict import AttrDict
import requests

from parasolr.solr.base import ClientBase


[docs]class CoreAdmin(ClientBase): """API client for Solr core admin. Args: solr_url: Base url for Solr. handler: Handler for CoreAdmin API session: A python-requests :class:`requests.Session` """ def __init__(self, solr_url: str, handler: str, session: Optional[requests.Session]=None) -> None: super().__init__(session=session) self.solr_url = solr_url self.url = urljoin('%s/' % solr_url, handler)
[docs] def create(self, name: str, **kwargs: Any) -> None: """Create a new core and register it. Args: name: Name of core to create. **kwargs: Any valid parameter for core creation in Solr. """ params = {'name': name, 'action': 'CREATE'} params.update(kwargs) self.make_request('get', self.url, params=params)
[docs] def unload(self, core: str, **kwargs: Any) -> None: """Unload a core, without defaults to remove data dir or index. Args: core: Name of core to unload. **kwargs: Any valid parameter for core unload in Solr. """ params = {'core': core, 'action': 'UNLOAD'} params.update(kwargs) self.make_request('get', self.url, params=params)
[docs] def reload(self, core: str) -> None: """Reload a Solr Core. Args: core: Name of core to reload. """ params = {'core': core, 'action': 'RELOAD'} return self.make_request('get', self.url, params=params)
[docs] def status(self, core: str=None) -> Optional[AttrDict]: """Get the status of all cores or one core. Args: core: Name of core to get status. """ params = {} if core: params['core'] = core response = self.make_request('get', self.url, params=params) if response: return AttrDict( initFailures=response.initFailures, status=response.status )
[docs] def ping(self, core: str) -> bool: """Ping a core to check status. Args: core: Name of core to ping. Returns: True if core status is OK, otherwise False. """ ping_url = '/'.join([self.solr_url.rstrip('/'), core, 'admin', 'ping']) # ping returns 404 if core does not exist, but that's ok here allowed_responses = [requests.codes.ok, requests.codes.not_found] response = self.make_request('get', ping_url, allowed_responses=allowed_responses) # return True if response is valid and status is OK return response and response.status == 'OK'