oso-cloud

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())