opa-python-client 2.0.4

Client for connection to the OPA service

OpaClient - Open Policy Agent Python Client

OpaClient is a Python client library designed to interact with the Open Policy Agent (OPA). It supports both synchronous and asynchronous requests, making it easy to manage policies, data, and evaluate rules in OPA servers.

Features

• Manage Policies: Create, update, retrieve, and delete policies.
• Manage Data: Create, update, retrieve, and delete data in OPA.
• Evaluate Policies: Use input data to evaluate policies and return decisions.
• Synchronous & Asynchronous: Choose between sync or async operations to suit your application.
• SSL/TLS Support: Communicate securely with SSL/TLS, including client certificates.
• Customizable: Use custom headers, timeouts, and other configurations.

Installation

You can install the OpaClient package via pip:

pip install opa-python-client

Quick Start

Synchronous Client Example

from opa_client.opa import OpaClient

# Initialize the OPA client
client = OpaClient(host='localhost', port=8181)

# Check the OPA server connection
try:
    print(client.check_connection())  # True
finally:
    client.close_connection()

or with client factory

from opa_client import create_opa_client

client = create_opa_client(host="localhost", port=8181)

Check OPA healthy. If you want check bundels or plugins, add query params for this.

from opa_client.opa import OpaClient

client = OpaClient()

print(client.check_health()) # response is  True or False
print(client.check_health({"bundle": True})) # response is  True or False
# If your diagnostic url different than default url, you can provide it.
print(client.check_health(diagnostic_url="http://localhost:8282/health"))  # response is  True or False
print(client.check_health(query={"bundle": True}, diagnostic_url="http://localhost:8282/health"))  # response is  True or False

Asynchronous Client Example

import asyncio
from opa_client.opa_async import AsyncOpaClient

async def main():
    async with AsyncOpaClient(host='localhost', port=8181) as client:
        result = await client.check_connection()
        print(result)

# Run the async main function
asyncio.run(main())

or with clien factory

from opa_client import create_opa_client

client = create_opa_client(async_mode=True,host="localhost", port=8181)

Secure Connection with SSL/TLS

You can use OpaClient with secure SSL/TLS connections, including mutual TLS (mTLS), by providing a client certificate and key.

Synchronous Client with SSL/TLS

from opa_client.opa import OpaClient

# Path to your certificate and private key
cert_path = '/path/to/client_cert.pem'
key_path = '/path/to/client_key.pem'

# Initialize the OPA client with SSL/TLS
client = OpaClient(
    host='your-opa-server.com',
    port=443,  # Typically for HTTPS
    ssl=True,
    cert=(cert_path, key_path)  # Provide the certificate and key as a tuple
)

# Check the OPA server connection
try:
    result = client.check_connection()
    print(result)
finally:
    client.close_connection()

Asynchronous Client with SSL/TLS

import asyncio
from opa_client.opa_async import AsyncOpaClient

# Path to your certificate and private key
cert_path = '/path/to/client_cert.pem'
key_path = '/path/to/client_key.pem'

async def main():
    # Initialize the OPA client with SSL/TLS
    async with AsyncOpaClient(
        host='your-opa-server.com',
        port=443,  # Typically for HTTPS
        ssl=True,
        cert=(cert_path, key_path)  # Provide the certificate and key as a tuple
    ) as client:
        # Check the OPA server connection
        result = await client.check_connection()
        print(result)

# Run the async main function
asyncio.run(main())

Usage

Policy Management

Create or Update a Policy

You can create or update a policy using the following syntax:

• Synchronous:

policy_name = 'example_policy'
policy_content = '''
package example

default allow = false

allow {
    input.user == "admin"
}
'''

client.update_policy_from_string(policy_content, policy_name)

• Asynchronous:

await client.update_policy_from_string(policy_content, policy_name)

Or from url:

• Synchronous:

policy_name = 'example_policy'

client.update_policy_from_url("http://opapolicyurlexample.test/example.rego", policy_name) 

• Asynchronous:

await client.update_policy_from_url("http://opapolicyurlexample.test/example.rego", policy_name) 

Update policy from rego file

client.update_opa_policy_fromfile("/your/path/filename.rego", endpoint="fromfile") # response is True

client.get_policies_list()

• Asynchronous:

await client.update_opa_policy_fromfile("/your/path/filename.rego", endpoint="fromfile") # response is True

await client.get_policies_list()

Retrieve a Policy

After creating a policy, you can retrieve it:

• Synchronous:

policy = client.get_policy('example_policy')
print(policy)
# or
policies = client.get_policies_list()
print(policies)

• Asynchronous:

policy = await client.get_policy('example_policy')
print(policy)

# or
policies = await client.get_policies_list()
print(policies)

Save policy to file from OPA service

client.policy_to_file(policy_name="example_policy",path="/your/path",filename="example.rego")

• Asynchronous:

await client.policy_to_file(policy_name="example_policy",path="/your/path",filename="example.rego")

Information about policy path and rules

print(client.get_policies_info())
#{'example_policy': {'path': 'http://localhost:8181/v1/data/example', 'rules': ['http://localhost:8181/v1/data/example/allow']}}

• Asynchronous:

print(await client.get_policies_info())
#{'example_policy': {'path': 'http://localhost:8181/v1/data/example', 'rules': ['http://localhost:8181/v1/data/example/allow']}}

Delete a Policy

You can delete a policy by name:

• Synchronous:

client.delete_policy('example_policy')

• Asynchronous:

await client.delete_policy('example_policy')

Data Management

Create or Update Data

You can upload arbitrary data to OPA:

• Synchronous:

data_name = 'users'
data_content = {
    "users": [
        {"name": "alice", "role": "admin"},
        {"name": "bob", "role": "user"}
    ]
}

client.update_or_create_data(data_content, data_name)

• Asynchronous:

await client.update_or_create_data(data_content, data_name)

Retrieve Data

You can fetch the data stored in OPA:

• Synchronous:

data = client.get_data('users')
print(data)
# You can use query params for additional info
# provenance - If parameter is true, response will include build/version info in addition to the result.
# metrics - Return query performance metrics in addition to result 
data = client.get_data('users',query_params={"provenance": True})
print(data) # {'provenance': {'version': '0.68.0', 'build_commit': 'db53d77c482676fadd53bc67a10cf75b3d0ce00b', 'build_timestamp': '2024-08-29T15:23:19Z', 'build_hostname': '3aae2b82a15f'}, 'result': {'users': [{'name': 'alice', 'role': 'admin'}, {'name': 'bob', 'role': 'user'}]}}


data = client.get_data('users',query_params={"metrics": True})
print(data) # {'metrics': {'counter_server_query_cache_hit': 0, 'timer_rego_external_resolve_ns': 7875, 'timer_rego_input_parse_ns': 875, 'timer_rego_query_compile_ns': 501083, 'timer_rego_query_eval_ns': 50250, 'timer_rego_query_parse_ns': 199917, 'timer_server_handler_ns': 1031291}, 'result': {'users': [{'name': 'alice', 'role': 'admin'}, {'name': 'bob', 'role': 'user'}]}}

• Asynchronous:

data = await client.get_data('users')
print(data)

Delete Data

To delete data from OPA:

• Synchronous:

client.delete_data('users')

• Asynchronous:

await client.delete_data('users')

Policy Evaluation

Check Permission (Policy Evaluation)

Evaluate a rule from a known package path. This is the recommended method for evaluating OPA decisions.

rego = """
package play

default hello = false

hello {
    m := input.message
    m == "world"
}
"""

check_data = {"message": "world"}

client.update_policy_from_string(rego, "test")
print(client.query_rule(input_data=check_data, package_path="play", rule_name="hello")) # {'result': True}

• Asynchronous:

rego = """
package play

default hello = false

hello {
    m := input.message
    m == "world"
}
"""

check_data = {"message": "world"}

await client.update_policy_from_string(rego, "test")
print(await client.query_rule(input_data=check_data, package_path="play", rule_name="hello")) # {'result': True}

You can evaluate policies with input data using check_permission.

⚠️ Deprecated: check_permission()

This method introspects the policy AST to construct a query path dynamically. It introduces unnecessary overhead and is not recommended for production use.

• Synchronous:

input_data = {"user": "admin"}
policy_name = 'example_policy'
rule_name = 'allow'

result = client.check_permission(input_data, policy_name, rule_name)
print(result)

🔥 Prefer query_rule() instead for better performance and maintainability.

⚠️ Deprecated: check_permission()

• Asynchronous:

input_data = {"user": "admin"}
policy_name = 'example_policy'
rule_name = 'allow'

result = await client.check_permission(input_data, policy_name, rule_name)
print(result)

🔥 Prefer query_rule() instead for better performance and maintainability.

Ad-hoc Queries

Execute ad-hoc queries directly:

• Synchronous:

data = {
    "user_roles": {
        "alice": [
            "admin"
        ],
        "bob": [
            "employee",
            "billing"
        ],
        "eve": [
            "customer"
        ]
    }
}
input_data = {"user": "admin"}
client.update_or_create_data(data, "userinfo")

result = client.ad_hoc_query(query="data.userinfo.user_roles[name]")
print(result) # {'result': [{'name': 'alice'}, {'name': 'bob'}, {'name': 'eve'}]}

• Asynchronous:

data = {
    "user_roles": {
        "alice": [
            "admin"
        ],
        "bob": [
            "employee",
            "billing"
        ],
        "eve": [
            "customer"
        ]
    }
}
input_data = {"user": "admin"}
await client.update_or_create_data(data, "userinfo")

result = await client.ad_hoc_query(query="data.userinfo.user_roles[name]")
print(result) # {'result': [{'name': 'alice'}, {'name': 'bob'}, {'name': 'eve'}]}

API Reference

Synchronous Client (OpaClient)

• check_connection(): Verify connection to OPA server.
• get_policies_list(): Get a list of all policies.
• get_policies_info(): Returns information about each policy, including policy path and policy rules.
• get_policy(policy_name): Fetch a specific policy.
• policy_to_file(policy_name): Save an OPA policy to a file..
• update_policy_from_string(policy_content, policy_name): Upload or update a policy using its string content.
• update_policy_from_url(url,endpoint): Update OPA policy by fetching it from a URL.
• update_policy_from_file(filepath,endpoint): Update OPA policy using a policy file.
• delete_policy(policy_name): Delete a specific policy.
• update_or_create_data(data_content, data_name): Create or update data in OPA.
• get_data(data_name): Retrieve data from OPA.
• delete_data(data_name): Delete data from OPA.
• check_permission(input_data, policy_name, rule_name): Evaluate a policy using input data.
• query_rule(input_data, package_path, rule_name): Query a specific rule in a package.
• ad_hoc_query(query, input_data): Run an ad-hoc query.

Asynchronous Client (AsyncOpaClient)

Same as the synchronous client, but all methods are asynchronous and must be awaited.

Contributing

Contributions are welcome! Feel free to open issues, fork the repo, and submit pull requests.

License

This project is licensed under the MIT License.