Oso Cloud Python Client

2.6.0 · active · verified Wed Apr 15

Oso Cloud is an authorization-as-a-service platform. The Python client (`oso-cloud`) provides a convenient wrapper around the Oso Cloud HTTP API, enabling applications to interact with the service for modeling, storing, enforcing, querying, and testing authorization logic using the declarative Polar policy language. The library is currently at version 2.6.0 and demonstrates a healthy release cadence, with the last version released less than a year ago.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to initialize the `oso-cloud` client, insert authorization facts, and perform permission checks and queries against the Oso Cloud service. It requires an active Oso Cloud account and an API key, which should be set as the `OSO_AUTH` environment variable. The examples assume a basic policy defining roles and permissions (e.g., an owner can read a repository) has been uploaded to your Oso Cloud instance.

import os
from oso_cloud import Oso, Value

# Initialize Oso Cloud client with an API key
# Ensure OSO_AUTH environment variable is set or pass it directly:
# oso = Oso(api_key=os.environ.get('OSO_AUTH', ''))
oso = Oso()

# Define example data
user = Value("User", "alice")
repository = Value("Repository", "my-repo")
organization = Value("Organization", "acme")

async def run_auth_checks():
    try:
        # Insert a fact: Alice has the role 'owner' on 'my-repo'
        await oso.insert(("has_role", user, "owner", repository))
        print(f"Inserted fact: Alice is owner of my-repo.")

        # Check if Alice can 'read' the 'my-repo' repository
        # This assumes a policy has been uploaded to Oso Cloud, e.g.,
        # allow(user: User, "read", repo: Repository) if has_role(user, "owner", repo);
        is_allowed = await oso.check(user, "read", repository)
        print(f"Can Alice 'read' my-repo? {is_allowed}")

        # Query for all repositories Alice can 'read'
        # Assumes a policy defining 'allow' rules.
        readable_repos = []
        async for result in oso.query(user, "read", Value("Repository")):
            if result.resource:
                readable_repos.append(result.resource.id)
        print(f"Repositories Alice can read: {readable_repos}")

        # Delete the previously inserted fact
        await oso.delete(("has_role", user, "owner", repository))
        print(f"Deleted fact: Alice is owner of my-repo.")

    except Exception as e:
        print(f"An error occurred: {e}")
        print("Make sure OSO_AUTH environment variable is set with your Oso Cloud API key.")

import asyncio
asyncio.run(run_auth_checks())

view raw JSON →