# Copyright © 2011-2026 Splunk, Inc.

#

# Licensed under the Apache License, Version 2.0 (the "License"): you may

# not use this file except in compliance with the License. You may obtain

# a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT

# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the

# License for the specific language governing permissions and limitations

# under the License.

"""The **splunklib.client** module provides a Pythonic interface to the

`Splunk REST API <http://docs.splunk.com/Documentation/Splunk/latest/RESTAPI/RESTcontents>`_,

allowing you programmatically access Splunk's resources.

**splunklib.client** wraps a Pythonic layer around the wire-level

binding of the **splunklib.binding** module. The core of the library is the

:class:`Service` class, which encapsulates a connection to the server, and

provides access to the various aspects of Splunk's functionality, which are

exposed via the REST API. Typically you connect to a running Splunk instance

with the :func:`connect` function::

import splunklib.client as client

service = client.connect(host="localhost", port=8089, username="admin", password="...")

assert isinstance(service, client.Service)

:class:`Service` objects have fields for the various Splunk resources (such as apps,

jobs, saved searches, inputs, and indexes). All of these fields are

:class:`Collection` objects::

appcollection = service.apps

my_app = appcollection.create("my_app")

my_app = appcollection["my_app"]

appcollection.delete("my_app")

The individual elements of the collection, in this case *applications*,

are subclasses of :class:`Entity`. An ``Entity`` object has fields for its

attributes, and methods that are specific to each kind of entity. For example::

print(my_app["author"]) # Or: print(my_app.author)

my_app.package() # Creates a compressed package of this application

The purpose of this module is to provide a friendlier domain interface to

various Splunk endpoints. The approach here is to leverage the binding

layer to capture endpoint context and provide objects and methods that

offer simplified access their corresponding endpoints. The design avoids

caching resource state. From the perspective of this module, the 'policy'

for caching resource state belongs in the application or a higher level

framework, and its the purpose of this module to provide simplified

access to that resource state.

A side note, the objects below that provide helper methods for updating eg:

Entity state, are written so that they may be used in a fluent style.

"""

import contextlib

import json

import logging

import re

import socket

from datetime import datetime, timedelta

from time import sleep

from urllib import parse

try:

from warnings import deprecated

except ImportError:

def deprecated(message): # pyright: ignore[reportUnknownParameterType]

return lambda _msg: None

from . import data

from .binding import (

AuthenticationError,

Context,

HTTPError,

UrlEncoded,

_encode,

_make_cookie_header,

_NoAuthenticationToken,

namespace,

)

from .data import record

logger = logging.getLogger(__name__)

__all__ = [

"AuthenticationError",

"IncomparableException",

"NotSupportedError",

"OperationError",

"Service",

"connect",

"namespace",

]

PATH_APPS = "apps/local/"

PATH_CAPABILITIES = "authorization/capabilities/"

PATH_CONF = "configs/conf-%s/"

PATH_PROPERTIES = "properties/"

PATH_DEPLOYMENT_CLIENTS = "deployment/client/"

PATH_DEPLOYMENT_TENANTS = "deployment/tenants/"

PATH_DEPLOYMENT_SERVERS = "deployment/server/"

PATH_DEPLOYMENT_SERVERCLASSES = "deployment/serverclass/"

PATH_EVENT_TYPES = "saved/eventtypes/"

PATH_FIRED_ALERTS = "alerts/fired_alerts/"

PATH_INDEXES = "data/indexes/"

PATH_INPUTS = "data/inputs/"

PATH_JOBS = "search/jobs/"

PATH_JOBS_V2 = "search/v2/jobs/"

PATH_LOGGER = "/services/server/logger/"

PATH_MACROS = "configs/conf-macros/"

PATH_MESSAGES = "messages/"

PATH_MODULAR_INPUTS = "data/modular-inputs"

PATH_ROLES = "authorization/roles/"

PATH_SAVED_SEARCHES = "saved/searches/"

PATH_STANZA = "configs/conf-%s/%s" # (file, stanza)

PATH_USERS = "authentication/users/"

PATH_RECEIVERS_STREAM = "/services/receivers/stream"

PATH_RECEIVERS_SIMPLE = "/services/receivers/simple"

PATH_STORAGE_PASSWORDS = "storage/passwords"

XNAMEF_ATOM = "{http://www.w3.org/2005/Atom}%s"

XNAME_ENTRY = XNAMEF_ATOM % "entry"

XNAME_CONTENT = XNAMEF_ATOM % "content"

MATCH_ENTRY_CONTENT = f"{XNAME_ENTRY}/{XNAME_CONTENT}/*"

class IllegalOperationException(Exception):

"""Thrown when an operation is not possible on the Splunk instance that a

:class:`Service` object is connected to."""

class IncomparableException(Exception):

"""Thrown when trying to compare objects (using ``==``, ``<``, ``>``, and

so on) of a type that doesn't support it."""

class AmbiguousReferenceException(ValueError):

"""Thrown when the name used to fetch an entity matches more than one entity."""

class InvalidNameException(Exception):

"""Thrown when the specified name contains characters that are not allowed

in Splunk entity names."""

class NoSuchCapability(Exception):

"""Thrown when the capability that has been referred to doesn't exist."""

class OperationError(Exception):

"""Raised for a failed operation, such as a timeout."""

class NotSupportedError(Exception):

"""Raised for operations that are not supported on a given object."""

def _trailing(template, *targets):

"""Substring of *template* following all *targets*.

**Example**::

template = "this is a test of the bunnies."

_trailing(template, "is", "est", "the") == " bunnies"

Each target is matched successively in the string, and the string

remaining after the last target is returned. If one of the targets

fails to match, a ValueError is raised.

:param template: Template to extract a trailing string from.

:type template: ``string``

:param targets: Strings to successively match in *template*.

:type targets: list of ``string``s

:return: Trailing string after all targets are matched.

:rtype: ``string``

:raises ValueError: Raised when one of the targets does not match.

"""

s = template

for t in targets:

n = s.find(t)

if n == -1:

raise ValueError("Target " + t + " not found in template.")

s = s[n + len(t) :]

return s

# Filter the given state content record according to the given arg list.

def _filter_content(content, *args):

if len(args) > 0:

return record((k, content[k]) for k in args)

return record(

(k, v) for k, v in content.items() if k not in ["eai:acl", "eai:attributes", "type"]

)

# Construct a resource path from the given base path + resource name

def _path(base, name):

if not base.endswith("/"):

base = base + "/"

return base + name

# Load an atom record from the body of the given response

# this will ultimately be sent to an xml ElementTree so we

# should use the xmlcharrefreplace option

def _load_atom(response, match=None):

return data.load(response.body.read().decode("utf-8", "xmlcharrefreplace"), match)

# Load an array of atom entries from the body of the given response

def _load_atom_entries(response):

r = _load_atom(response)

if "feed" in r:

# Need this to handle a random case in the REST API

if r.feed.get("totalResults") in [0, "0"]:

return []

entries = r.feed.get("entry", None)

if entries is None:

return None

return entries if isinstance(entries, list) else [entries]

# Unlike most other endpoints, the jobs endpoint does not return

# its state wrapped in another element, but at the top level.

# For example, in XML, it returns <entry>...</entry> instead of

# <feed><entry>...</entry></feed>.

entries = r.get("entry", None)

if entries is None:

return None

return entries if isinstance(entries, list) else [entries]

# Load the sid from the body of the given response

def _load_sid(response, output_mode):

if output_mode == "json":

json_obj = json.loads(response.body.read())

return json_obj.get("sid")

return _load_atom(response).response.sid

# Parse the given atom entry record into a generic entity state record

def _parse_atom_entry(entry):

title = entry.get("title", None)

elink = entry.get("link", [])

elink = elink if isinstance(elink, list) else [elink]

links = record((link.rel, link.href) for link in elink)

# Retrieve entity content values

content = entry.get("content", {})

# Host entry metadata

metadata = _parse_atom_metadata(content)

# Filter some of the noise out of the content record

content = record((k, v) for k, v in content.items() if k not in ["eai:acl", "eai:attributes"])

if "type" in content:

if isinstance(content["type"], list):

content["type"] = [t for t in content["type"] if t != "text/xml"]

# Unset type if it was only 'text/xml'

if len(content["type"]) == 0:

content.pop("type", None)

# Flatten 1 element list

if len(content["type"]) == 1:

content["type"] = content["type"][0]

else:

content.pop("type", None)

return record(

{

"title": title,

"links": links,

"access": metadata.access,

"fields": metadata.fields,

"content": content,

"updated": entry.get("updated"),

"published": entry.get("published"),

"author": entry.get("author"),

}

)

# Parse the metadata fields out of the given atom entry content record

def _parse_atom_metadata(content):

# Hoist access metadata

access = content.get("eai:acl", None)

# Hoist content metadata (and cleanup some naming)

attributes = content.get("eai:attributes", {})

fields = record(

{

"required": attributes.get("requiredFields", []),

"optional": attributes.get("optionalFields", []),

"wildcard": attributes.get("wildcardFields", []),

}

)

return record({"access": access, "fields": fields})

# kwargs: scheme, host, port, app, owner, username, password

def connect(**kwargs):

"""This function connects and logs in to a Splunk instance.

This function is a shorthand for :meth:`Service.login`.

The ``connect`` function makes one round trip to the server (for logging in).

:param host: The host name (the default is "localhost").

:type host: ``string``

:param port: The port number (the default is 8089).

:type port: ``integer``

:param scheme: The scheme for accessing the service (the default is "https").

:type scheme: "https" or "http"

:param verify: Enable (True) or disable (False) SSL verification for https connections.

:type verify: ``Boolean``

:param self_signed_certificate: Specifies if self signed certificate is used

:type self_signed_certificate: ``Boolean``

:param `key_file`: Path to a PEM-encoded private key.

:type key_file: ``string``

:param `cert_file`: Path to a PEM-encoded X509 certificate chain.

:type cert_file: ``string``

:param `context`: Custom SSLContext used with the HTTPSConnection, requires verify=True.

:type context: ``SSLContext``

:param `owner`: The owner context of the namespace (optional).

:type owner: ``string``

:param `app`: The app context of the namespace (optional).

:type app: ``string``

:param sharing: The sharing mode for the namespace (the default is "user").

:type sharing: "global", "system", "app", or "user"

:param `token`: The current session token (optional). Session tokens can be

shared across multiple service instances.

:type token: ``string``

:param cookie: A session cookie. When provided, you don't need to call :meth:`login`.

This parameter is only supported for Splunk 6.2+.

:type cookie: ``string``

:param autologin: When ``True``, automatically tries to log in again if the

session terminates.

:type autologin: ``boolean``

:param `username`: The Splunk account username, which is used to

authenticate the Splunk instance.

:type username: ``string``

:param `password`: The password for the Splunk account.

:type password: ``string``

:param retries: Number of retries for each HTTP connection (optional, the default is 0).

NOTE: THIS MAY INCREASE THE NUMBER OF ROUNDTRIP CONNECTIONS TO THE SPLUNK SERVER.

:type retries: ``int``

:param retryDelay: How long to wait between connection attempts if `retries` > 0 (optional, defaults to 10s).

:type retryDelay: ``int`` (in seconds)

:param `context`: The SSLContext that can be used when setting verify=True (optional)

:type context: ``SSLContext``

:return: An initialized :class:`Service` connection.

**Example**::

import splunklib.client as client

s = client.connect(...)

a = s.apps["my_app"]

...

"""

s = Service(**kwargs)

s.login()

return s

# In preparation for adding Storm support, we added an

# intermediary class between Service and Context. Storm's

# API is not going to be the same as enterprise Splunk's

# API, so we will derive both Service (for enterprise Splunk)

# and StormService for (Splunk Storm) from _BaseService, and

# put any shared behavior on it.

class _BaseService(Context):

pass

class Service(_BaseService):

"""A Pythonic binding to Splunk instances.

A :class:`Service` represents a binding to a Splunk instance on an

HTTP or HTTPS port. It handles the details of authentication, wire

formats, and wraps the REST API endpoints into something more

Pythonic. All of the low-level operations on the instance from

:class:`splunklib.binding.Context` are also available in case you need

to do something beyond what is provided by this class.

After creating a ``Service`` object, you must call its :meth:`login`

method before you can issue requests to Splunk.

Alternately, use the :func:`connect` function to create an already

authenticated :class:`Service` object, or provide a session token

when creating the :class:`Service` object explicitly (the same

token may be shared by multiple :class:`Service` objects).

:param host: The host name (the default is "localhost").

:type host: ``string``

:param port: The port number (the default is 8089).

:type port: ``integer``

:param scheme: The scheme for accessing the service (the default is "https").

:type scheme: "https" or "http"

:param verify: Enable (True) or disable (False) SSL verification for https connections.

:type verify: ``Boolean``

:param self_signed_certificate: Specifies if self signed certificate is used

:type self_signed_certificate: ``Boolean``

:param `key_file`: Path to a PEM-encoded private key.

:type key_file: ``string``

:param `cert_file`: Path to a PEM-encoded X509 certificate chain.

:type cert_file: ``string``

:param `context`: Custom SSLContext used with the HTTPSConnection, requires verify=True.

:type context: ``SSLContext``

:param `owner`: The owner context of the namespace (optional; use "-" for wildcard).

:type owner: ``string``

:param `app`: The app context of the namespace (optional; use "-" for wildcard).

:type app: ``string``

:param `token`: The current session token (optional). Session tokens can be

shared across multiple service instances.

:type token: ``string``

:param cookie: A session cookie. When provided, you don't need to call :meth:`login`.

This parameter is only supported for Splunk 6.2+.

:type cookie: ``string``

:param `username`: The Splunk account username, which is used to

authenticate the Splunk instance.

:type username: ``string``

:param `password`: The password, which is used to authenticate the Splunk

instance.

:type password: ``string``

:param retries: Number of retries for each HTTP connection (optional, the default is 0).

NOTE: THIS MAY INCREASE THE NUMBER OF ROUNDTRIP CONNECTIONS TO THE SPLUNK SERVER.

:type retries: ``int``

:param retryDelay: How long to wait between connection attempts if `retries` > 0 (optional, defaults to 10s).

:type retryDelay: ``int`` (in seconds)

:return: A :class:`Service` instance.

**Example**::

import splunklib.client as client

s = client.Service(username="boris", password="natasha", ...)

s.login()

# Or equivalently

s = client.connect(username="boris", password="natasha")

# Or if you already have a session token

s = client.Service(token="atg232342aa34324a")

# Or if you already have a valid cookie

s = client.Service(cookie="splunkd_8089=...")

"""

def __init__(self, **kwargs):

super().__init__(**kwargs)

self._splunk_version = None

self._kvstore_owner = None

self._instance_type = None

@property

def apps(self):

"""Returns the collection of applications that are installed on this instance of Splunk.

:return: A :class:`Collection` of :class:`Application` entities.

"""

return Collection(self, PATH_APPS, item=Application)

@property

def confs(self):

"""Returns the collection of configuration files for this Splunk instance.

:return: A :class:`Configurations` collection of

:class:`ConfigurationFile` entities.

"""

return Configurations(self)

@property

def capabilities(self):

"""Returns the list of system capabilities.

:return: A ``list`` of capabilities.

"""

response = self.get(PATH_CAPABILITIES)

return _load_atom(response, MATCH_ENTRY_CONTENT).capabilities

@property

def event_types(self):

"""Returns the collection of event types defined in this Splunk instance.

:return: An :class:`Entity` containing the event types.

"""

return Collection(self, PATH_EVENT_TYPES)

@property

def fired_alerts(self):

"""Returns the collection of alerts that have been fired on the Splunk

instance, grouped by saved search.

:return: A :class:`Collection` of :class:`AlertGroup` entities.

"""

return Collection(self, PATH_FIRED_ALERTS, item=AlertGroup)

@property

def indexes(self):

"""Returns the collection of indexes for this Splunk instance.

:return: An :class:`Indexes` collection of :class:`Index` entities.

"""

return Indexes(self, PATH_INDEXES, item=Index)

@property

def info(self):

"""Returns the information about this instance of Splunk.

:return: The system information, as key-value pairs.

:rtype: ``dict``

"""

response = self.get("/services/server/info")

return _filter_content(_load_atom(response, MATCH_ENTRY_CONTENT))

def input(self, path, kind=None):

"""Retrieves an input by path, and optionally kind.

:return: A :class:`Input` object.

"""

return Input(self, path, kind=kind).refresh()

@property

def inputs(self):

"""Returns the collection of inputs configured on this Splunk instance.

:return: An :class:`Inputs` collection of :class:`Input` entities.

"""

return Inputs(self)

def job(self, sid):

"""Retrieves a search job by sid.

:return: A :class:`Job` object.

"""

return Job(self, sid).refresh()

@property

def jobs(self):

"""Returns the collection of current search jobs.

:return: A :class:`Jobs` collection of :class:`Job` entities.

"""

return Jobs(self)

@property

def loggers(self):

"""Returns the collection of logging level categories and their status.

:return: A :class:`Loggers` collection of logging levels.

"""

return Loggers(self)

@property

def messages(self):

"""Returns the collection of service messages.

:return: A :class:`Collection` of :class:`Message` entities.

"""

return Collection(self, PATH_MESSAGES, item=Message)

@property

def modular_input_kinds(self):

"""Returns the collection of the modular input kinds on this Splunk instance.

:return: A :class:`ReadOnlyCollection` of :class:`ModularInputKind` entities.

"""

if self.splunk_version >= (5,):

return ReadOnlyCollection(self, PATH_MODULAR_INPUTS, item=ModularInputKind)

raise IllegalOperationException("Modular inputs are not supported before Splunk version 5.")

@property

def storage_passwords(self):

"""Returns the collection of the storage passwords on this Splunk instance.

:return: A :class:`ReadOnlyCollection` of :class:`StoragePasswords` entities.

"""

return StoragePasswords(self)

# kwargs: enable_lookups, reload_macros, parse_only, output_mode

def parse(self, query, **kwargs):

"""Parses a search query and returns a semantic map of the search.

:param query: The search query to parse.

:type query: ``string``

:param kwargs: Arguments to pass to the ``search/parser`` endpoint

(optional). Valid arguments are:

* "enable_lookups" (``boolean``): If ``True``, performs reverse lookups

to expand the search expression.

* "output_mode" (``string``): The output format (XML or JSON).

* "parse_only" (``boolean``): If ``True``, disables the expansion of

search due to evaluation of subsearches, time term expansion,

lookups, tags, eventtypes, and sourcetype alias.

* "reload_macros" (``boolean``): If ``True``, reloads macro

definitions from macros.conf.

:type kwargs: ``dict``

:return: A semantic map of the parsed search query.

"""

if not self.disable_v2_api:

return self.post("search/v2/parser", q=query, **kwargs)

return self.get("search/parser", q=query, **kwargs)

def restart(self, timeout=None):

"""Restarts this Splunk instance.

The service is unavailable until it has successfully restarted.

If a *timeout* value is specified, ``restart`` blocks until the service

resumes or the timeout period has been exceeded. Otherwise, ``restart`` returns

immediately.

:param timeout: A timeout period, in seconds.

:type timeout: ``integer``

"""

msg = {"value": f"Restart requested by {self.username} via the Splunk SDK for Python"}

# This message will be deleted once the server actually restarts.

self.messages.create(name="restart_required", **msg)

result = self.post("/services/server/control/restart")

if timeout is None:

return result

start = datetime.now()

diff = timedelta(seconds=timeout)

while datetime.now() - start < diff:

try:

self.login()

if self.restart_required:

# Prevent a burst of requests from bombarding Splunk.

# Splunk does not stop the API immediately when /services/server/control/restart

# responds, thus the login call (above) will still succeed until the server

# is actually stopped. Based on the presence of restart_required message,

# that we have added before calling restart, we know that the server did not stop yet.

sleep(1)

continue

else:

return result

except Exception:

sleep(1)

raise Exception("Operation time out.")

@property

def restart_required(self):

"""Indicates whether splunkd is in a state that requires a restart.

:return: A ``boolean`` that indicates whether a restart is required.

"""

response = self.get("messages").body.read()

messages = data.load(response)["feed"]

if "entry" not in messages:

result = False

else:

if isinstance(messages["entry"], dict):

titles = [messages["entry"]["title"]]

else:

titles = [x["title"] for x in messages["entry"]]

result = "restart_required" in titles

return result

@property

def roles(self):

"""Returns the collection of user roles.

:return: A :class:`Roles` collection of :class:`Role` entities.

"""

return Roles(self)

def search(self, query, **kwargs):

"""Runs a search using a search query and any optional arguments you

provide, and returns a `Job` object representing the search.

:param query: A search query.

:type query: ``string``

:param kwargs: Arguments for the search (optional):

* "output_mode" (``string``): Specifies the output format of the

results.

* "earliest_time" (``string``): Specifies the earliest time in the

time range to

search. The time string can be a UTC time (with fractional

seconds), a relative time specifier (to now), or a formatted

time string.

* "latest_time" (``string``): Specifies the latest time in the time

range to

search. The time string can be a UTC time (with fractional

seconds), a relative time specifier (to now), or a formatted

time string.

* "rf" (``string``): Specifies one or more fields to add to the

search.

:type kwargs: ``dict``

:rtype: class:`Job`

:returns: An object representing the created job.

"""

return self.jobs.create(query, **kwargs)

@property

def saved_searches(self):

"""Returns the collection of saved searches.

:return: A :class:`SavedSearches` collection of :class:`SavedSearch`

entities.

"""

return SavedSearches(self)

@property

def macros(self):

"""Returns the collection of macros.

:return: A :class:`Macros` collection of :class:`Macro`

entities.

"""

return Macros(self)

@property

def settings(self):

"""Returns the configuration settings for this instance of Splunk.

:return: A :class:`Settings` object containing configuration settings.

"""

return Settings(self)

@property

def splunk_version(self):

"""Returns the version of the splunkd instance this object is attached

to.

The version is returned as a tuple of the version components as

integers (for example, `(4,3,3)` or `(5,)`).

:return: A ``tuple`` of ``integers``.

"""

if self._splunk_version is None:

self._splunk_version = tuple(int(p) for p in self.info["version"].split("."))

return self._splunk_version

@property

def splunk_instance(self):

if self._instance_type is None:

splunk_info = self.info

if hasattr(splunk_info, "instance_type"):

self._instance_type = splunk_info["instance_type"]

else:

self._instance_type = ""

return self._instance_type

@property

def disable_v2_api(self):

if self.splunk_instance.lower() == "cloud":

return self.splunk_version < (9, 0, 2209)

return self.splunk_version < (9, 0, 2)

@property

def kvstore_owner(self):

"""Returns the KVStore owner for this instance of Splunk.

By default is the kvstore owner is not set, it will return "nobody"

:return: A string with the KVStore owner.

"""

if self._kvstore_owner is None:

self._kvstore_owner = "nobody"

return self._kvstore_owner

@kvstore_owner.setter

def kvstore_owner(self, value):

"""

kvstore is refreshed, when the owner value is changed

"""

self._kvstore_owner = value

self.kvstore

@property

def kvstore(self):

"""Returns the collection of KV Store collections.

sets the owner for the namespace, before retrieving the KVStore Collection

:return: A :class:`KVStoreCollections` collection of :class:`KVStoreCollection` entities.

"""

self.namespace["owner"] = self.kvstore_owner

return KVStoreCollections(self)

@property

def users(self):

"""Returns the collection of users.

:return: A :class:`Users` collection of :class:`User` entities.

"""

return Users(self)

class Endpoint:

"""This class represents individual Splunk resources in the Splunk REST API.

An ``Endpoint`` object represents a URI, such as ``/services/saved/searches``.

This class provides the common functionality of :class:`Collection` and

:class:`Entity` (essentially HTTP GET and POST methods).

"""

def __init__(self, service, path):

self.service = service

self.path = path

def get_api_version(self, path):

"""Return the API version of the service used in the provided path.

Args:

path (str): A fully-qualified endpoint path (for example, "/services/search/jobs").

Returns:

int: Version of the API (for example, 1)

"""

# Default to v1 if undefined in the path

# For example, "/services/search/jobs" is using API v1

api_version = 1

versionSearch = re.search(r"(?:servicesNS\/[^/]+\/[^/]+|services)\/[^/]+\/v(\d+)\/", path)

if versionSearch:

api_version = int(versionSearch.group(1))

return api_version

def get(self, path_segment="", owner=None, app=None, sharing=None, **query):

"""Performs a GET operation on the path segment relative to this endpoint.

This method is named to match the HTTP method. This method makes at least

one roundtrip to the server, one additional round trip for

each 303 status returned, plus at most two additional round

trips if

the ``autologin`` field of :func:`connect` is set to ``True``.

If *owner*, *app*, and *sharing* are omitted, this method takes a

default namespace from the :class:`Service` object for this :class:`Endpoint`.

All other keyword arguments are included in the URL as query parameters.

:raises AuthenticationError: Raised when the ``Service`` is not logged in.

:raises HTTPError: Raised when an error in the request occurs.

:param path_segment: A path segment relative to this endpoint.

:type path_segment: ``string``

:param owner: The owner context of the namespace (optional).

:type owner: ``string``

:param app: The app context of the namespace (optional).

:type app: ``string``

:param sharing: The sharing mode for the namespace (optional).

:type sharing: "global", "system", "app", or "user"

:param query: All other keyword arguments, which are used as query

parameters.

:type query: ``string``

:return: The response from the server.

:rtype: ``dict`` with keys ``body``, ``headers``, ``reason``,

and ``status``

**Example**::

import splunklib.client

s = client.service(...)

apps = s.apps

apps.get() == \\

{'body': ...a response reader object...,

'headers': [('content-length', '26208'),

('expires', 'Fri, 30 Oct 1998 00:00:00 GMT'),

('server', 'Splunkd'),

('connection', 'close'),

('cache-control', 'no-store, max-age=0, must-revalidate, no-cache'),

('date', 'Fri, 11 May 2012 16:30:35 GMT'),

('content-type', 'text/xml; charset=utf-8')],

'reason': 'OK',

'status': 200}

apps.get('nonexistant/path') # raises HTTPError

s.logout()

apps.get() # raises AuthenticationError

"""

# self.path to the Endpoint is relative in the SDK, so passing

# owner, app, sharing, etc. along will produce the correct

# namespace in the final request.

if path_segment.startswith("/"):

path = path_segment

else:

if not self.path.endswith("/") and path_segment != "":

self.path = self.path + "/"

path = self.service._abspath(

self.path + path_segment, owner=owner, app=app, sharing=sharing

)

# ^-- This was "%s%s" % (self.path, path_segment).

# That doesn't work, because self.path may be UrlEncoded.

# Get the API version from the path

api_version = self.get_api_version(path)

# Search API v2+ fallback to v1:

# - In v2+, /results_preview, /events and /results do not support search params.

# - Fallback from v2+ to v1 if Splunk Version is < 9.

# if api_version >= 2 and ('search' in query and path.endswith(tuple(["results_preview", "events", "results"])) or self.service.splunk_version < (9,)):

# path = path.replace(PATH_JOBS_V2, PATH_JOBS)

if api_version == 1:

if isinstance(path, UrlEncoded):

path = UrlEncoded(path.replace(PATH_JOBS_V2, PATH_JOBS), skip_encode=True)

else:

path = path.replace(PATH_JOBS_V2, PATH_JOBS)

return self.service.get(path, owner=owner, app=app, sharing=sharing, **query)

def post(self, path_segment="", owner=None, app=None, sharing=None, **query):

"""Performs a POST operation on the path segment relative to this endpoint.

This method is named to match the HTTP method. This method makes at least

one roundtrip to the server, one additional round trip for

each 303 status returned, plus at most two additional round trips if

the ``autologin`` field of :func:`connect` is set to ``True``.

If *owner*, *app*, and *sharing* are omitted, this method takes a

default namespace from the :class:`Service` object for this :class:`Endpoint`.

All other keyword arguments are included in the URL as query parameters.

:raises AuthenticationError: Raised when the ``Service`` is not logged in.

:raises HTTPError: Raised when an error in the request occurs.

:param path_segment: A path segment relative to this endpoint.

:type path_segment: ``string``

:param owner: The owner context of the namespace (optional).

:type owner: ``string``

:param app: The app context of the namespace (optional).

:type app: ``string``

:param sharing: The sharing mode of the namespace (optional).

:type sharing: ``string``

:param query: All other keyword arguments, which are used as query

parameters.

:type query: ``string``

:return: The response from the server.

:rtype: ``dict`` with keys ``body``, ``headers``, ``reason``,

and ``status``

**Example**::

import splunklib.client

s = client.service(...)

apps = s.apps

apps.post(name='boris') == \\

{'body': ...a response reader object...,

'headers': [('content-length', '2908'),

('expires', 'Fri, 30 Oct 1998 00:00:00 GMT'),

('server', 'Splunkd'),

('connection', 'close'),

('cache-control', 'no-store, max-age=0, must-revalidate, no-cache'),

('date', 'Fri, 11 May 2012 18:34:50 GMT'),

('content-type', 'text/xml; charset=utf-8')],

'reason': 'Created',

'status': 201}

apps.get('nonexistant/path') # raises HTTPError

s.logout()

apps.get() # raises AuthenticationError

"""

if path_segment.startswith("/"):

path = path_segment

else:

if not self.path.endswith("/") and path_segment != "":

self.path = self.path + "/"

path = self.service._abspath(

self.path + path_segment, owner=owner, app=app, sharing=sharing

)

# Get the API version from the path

api_version = self.get_api_version(path)

# Search API v2+ fallback to v1:

# - In v2+, /results_preview, /events and /results do not support search params.

# - Fallback from v2+ to v1 if Splunk Version is < 9.

# if api_version >= 2 and ('search' in query and path.endswith(tuple(["results_preview", "events", "results"])) or self.service.splunk_version < (9,)):

# path = path.replace(PATH_JOBS_V2, PATH_JOBS)

if api_version == 1:

if isinstance(path, UrlEncoded):

path = UrlEncoded(path.replace(PATH_JOBS_V2, PATH_JOBS), skip_encode=True)

else:

path = path.replace(PATH_JOBS_V2, PATH_JOBS)

return self.service.post(path, owner=owner, app=app, sharing=sharing, **query)

# kwargs: path, app, owner, sharing, state

class Entity(Endpoint):

"""This class is a base class for Splunk entities in the REST API, such as

saved searches, jobs, indexes, and inputs.

``Entity`` provides the majority of functionality required by entities.

Subclasses only implement the special cases for individual entities.

For example for saved searches, the subclass makes fields like ``action.email``,

``alert_type``, and ``search`` available.

An ``Entity`` is addressed like a dictionary, with a few extensions,