Reference

Main interface

All of the functionality can be accessed by instantiating opa.OPAClient.

class opa.OPAClient(url: str = 'http://localhost:8181', verify: bool = True, retries: int = 5, token: Optional[str] = None)

Initialize a new OPA client.

Parameters

• url – (Optional) URL to OPA server. Defaults to http://localhost:8181.

• verify – (Optional) Dictates whether SSL certificates should be verfied or not.

• token – (Optional) Token used to authorize client with OPA server.

• retries – (Optional) Number of times requests should be retried before giving up.

check_custom_health_rule(rule: str)

Run custom rule in the system.health policy. The check will fail if the rule evaluates to false or does not exist.

check_health(bundles: bool = False, plugins: bool = False, exclude_plugins: Optional[list[str]] = None, retries: Optional[int] = None) → bool

Check that the connection to the OPA server is healthy.

Parameters

• bundles – (Optional) Account for bundles during health check.

• plugins – (Optional) Account for plugins during health check.

• exclude_plugins – (Optional) Plugins to exclude from check. Only valid if plugins is true.

• retries – (Optional) Override client retry setting.

check_liveness() → bool

Check liveness by making sure the live rule in the system.health policy exists and evaluates to true.

check_policy(package: str, input: Optional[dict[str, Any]] = None, raw: bool = False, pretty: bool = False, provenance: bool = False, instrument: bool = False, strict: bool = False, explain: Optional[Literal['notes', 'fails', 'full', 'debug']] = None, metrics: bool = False) → dict[str, Any]

Request a decision for the specified named policy.

Parameters

• package – Package that defines the policy. May include rule name.

• input – (Optional) Python dict providing input for the policy.

• raw – (Optional) Whether to return the raw response or just use the result-part of the object returned from the server.

• pretty – (Optional) Make the result pretty.

• provenance – (Optional) Include provenance information in the response. Will always return a raw value.

• instrument – (Optional) Instrument query evaluation and include details in response. Will always return a raw value.

• strict – (Optional) Treat built-in-function call errors as fatal.

• explain – (Optional) Include query explanation in response. Will always return a raw value.

• metrics – (Optional) Include query performance metrics in response. Will always return a raw value.

check_readiness() → bool

Check readiness by making sure the ready rule in the system.health policy exists and evaluates to true.

delete_document(package: str) → dict[str, Any]

Delete a document.

Parameters

package – Package path to the document.

delete_policy(id: str) → dict[str, Any]

Delete a policy.

Parameters

id – Id of the policy.

get_config() → dict[str, Any]

Return OPA’s active configuration as a dict.

get_default_policy() → dict[str, Any]

Attempts to retrieve the default policy. Will raise PolicyNotFound if no policy with the package signature system could be found.

get_document(package: str) → dict[str, Any]

Find and return a document.

Parameters

package – Package path to the document.

get_policy(id: str) → dict[str, Any]

Get a specific policy.

Parameters

id – Id of the policy.

list_documents() → list[dict[str, Any]]

List all available documents.

list_policies() → list[dict[str, Any]]

List all policies.

package_path(package: str) → str

Normalize and return a package path that can be used in HTTP requests. Replaces ‘.’ with ‘/’ and strips leading slash.

Parameters

package – Package path to normalize.

parse_url(url: str) → str

Parse and perform basic validation of supplied url.

Url

URL to parse. With or without scheme.

query(query: str, input: dict[str, Any], pretty: bool = False, explain: Optional[Literal['notes', 'fails', 'full', 'debug']] = None, metrics: bool = False) → dict[str, Any]

Execute an ad-hoc query and return bindings for variables found in the query.

Parameters

• query – The query to execute.

• input – Python dict providing input for the query.

• pretty – (Optional) Make the result pretty.

• explain – (Optional) Include query explanation in response.

• metrics – (Optional) Include query performance metrics in response.

request(verb: str, path: str, json: Optional[Any] = None, params: Optional[Any] = None, data: Optional[Any] = None, retries: Optional[int] = None) → Response

Make a request to OPA server. Used primarily internally.

Parameters

• verb – The HTTP verb to use.

• path – Path to make request to.

• json – (Optional) JSON serializable object to send.

• params – (Optional) Dictionary to send in the query string.

• data – (Optional) Arbitrary object to send.

• retries – (Optional) Override client retry setting.

save_document(package: str, data: dict[str, Any]) → None

Create or update a document.

Parameters

• package – Package path used to access the data.

• data – JSON serializable object containing data.

save_policy(id: str, policy: str) → dict[str, Any]

Create or update a policy.

Parameters

• id – Id of the policy.

• policy – Policy written in rego.

Exceptions

exception opa.exceptions.ConnectionError

Generic connection error.

exception opa.exceptions.DocumentNotFound

The document could not be found.

exception opa.exceptions.InvalidDocument

The document provided was not valid.

exception opa.exceptions.InvalidPolicy

The policy provided was not valid.

exception opa.exceptions.InvalidPolicyRequest

The policy-request was not valid.

exception opa.exceptions.InvalidURL

The URL was somehow invalid.

exception opa.exceptions.OPAException

Base exception all other opa-client exceptions are derived from.

exception opa.exceptions.PolicyNotFound

The policy could not be found.

exception opa.exceptions.PolicyRequestError

There was an error requesting the policy.

exception opa.exceptions.Unauthorized

The request to OPA was not authorized.