See offical documentation page Open Policy Agent
$ pip install OPA-python-client>>>fromopa_client.opaimportOpaClient>>>client=OpaClient() # default host='localhost', port=8181, version='v1'>>>client.check_connection()
'Yes I"m here :)'>>>test_policy="""... package play... ... import data.testapi.testdata... ... default hello = false... ... hello {... m := input.message... testdata[i] == m... }... """>>>client.update_opa_policy_fromstring(test_policy, "testpolicy")
True>>>client.get_policies_list()
['testpolicy']
>>>data= ["world", "hello"]
>>>client.update_or_create_opa_data(data, "testapi/testdata")
True>>>check_data= {"input": {"message": "hello"}}
>>>client.check_permission(input_data=check_data, policy_name="testpolicy", rule_name="hello")
{'result': True}fromopa_client.opaimportOpaClientclient=OpaClient() # default host='localhost', port=8181, version='v1'client.check_connection() # response is Yes I'm here :)# Ensure the connection is closed correctly by deleting the clientdelclientfromopa_client.opaimportOpaClientclient=OpaClient(
host="https://192.168.99.100",
port=8181,
version="v1",
ssl=True,
cert="/your/certificate/file/path/mycert.crt",
)
client.check_connection() # response is Yes I'm here :)delclientfromopa_client.opaimportOpaClientclient=OpaClient() # default host='localhost', port=8181, version='v1'client.update_opa_policy_fromfile("/your/path/filename.rego", endpoint="fromfile") # response is Trueclient.get_policies_list() # response is ["fromfile"]delclientfromopa_client.opaimportOpaClientclient=OpaClient() # default host='localhost', port=8181, version='v1'client.update_opa_policy_fromurl("http://opapolicyurlexample.test/example.rego", endpoint="fromurl") # response is Trueclient.get_policies_list() # response is ["fromfile","fromurl"]delclientfromopa_client.opaimportOpaClientclient=OpaClient() # default host='localhost', port=8181, version='v1'client.delete_opa_policy("fromfile") # response is Trueclient.get_policies_list() # response is [fromurl"]delclientfromopa_client.opaimportOpaClientclient=OpaClient() # default host='localhost', port=8181, version='v1'print(client.get_opa_raw_data("testapi/testdata")) # response is {'result': ['world', 'hello']}delclientfromopa_client.opaimportOpaClientclient=OpaClient() # default host='localhost', port=8181, version='v1'client.opa_policy_to_file(policy_name="fromurl",path="/your/path",filename="example.rego") # response is Truedelclientfromopa_client.opaimportOpaClientclient=OpaClient() # default host='localhost', port=8181, version='v1'client.delete_opa_data("testapi") # response is Truedelclientfromopa_client.opaimportOpaClientclient=OpaClient() # default host='localhost', port=8181, version='v1'client.get_policies_info()
# response is {'testpolicy': {'path': ['http://your-opa-service/v1/data/play'], 'rules': ['http://your-opa-service/v1/data/play/hello']}delclientfromopa_client.opaimportOpaClientclient=OpaClient() # default host='localhost', port=8181, version='v1'permission_you_want_check= {"input": {"message": "hello"}}
client.check_permission(input_data=permission_you_want_check, policy_name="testpolicy", rule_name="hello")
# response is {'result': True}delclient