Source code for parasolr.solr.admin
from typing import Any, Optional
from urllib.parse import urljoin
import requests
from addict import Dict as AttrDict
from parasolr.solr.base import ClientBase, SolrConnectionNotFound
[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]
try:
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"
except SolrConnectionNotFound:
return False