Python SDK

Async and sync Python client for InferaDB.

Coming soon. The Python SDK is under active development. The API surface shown here is based on the Rust SDK and may change before release.

Async and sync clients for InferaDB’s authorization APIs. Requires Python 3.10+.

Installation

pip install inferadb

Authentication

Three authentication methods:

Method	Use Case	Security
Client Credentials (Ed25519 JWT)	Service-to-service	High
Bearer Token	User sessions, OAuth	Medium
API Key	Testing, simple integrations	Basic

Client Credentials (Recommended)

from inferadb import InferaDB, Ed25519PrivateKey

client = InferaDB(
    url="https://engine.inferadb.com",
    credentials={
        "client_id": "my-client",
        "private_key": Ed25519PrivateKey.from_pem_file("client.pem"),
        "certificate_id": "cert-id",
    },
)

Bearer Token

import os
from inferadb import InferaDB

client = InferaDB(
    url="https://engine.inferadb.com",
    token=os.environ["INFERADB_TOKEN"],
)

API Key

import os
from inferadb import InferaDB

client = InferaDB(
    url="https://engine.inferadb.com",
    api_key=os.environ["INFERADB_API_KEY"],
)

Permission Checks

vault = client.organization("my-org").vault("production")

allowed = await vault.check("user:alice", "can_edit", "document:readme")

With ABAC Context

allowed = await vault.check(
    "user:alice", "can_view", "document:readme",
    context={"ip_address": "10.0.0.1"},
)

Require — Raises on Deny

# Raises AccessDeniedError if permission is denied
await vault.require("user:alice", "can_edit", "document:readme")

With Consistency Token

allowed = await vault.check(
    "user:alice", "can_view", "document:readme",
    at_least_as_fresh=revision_token,
)

Batch Check

results = await vault.check_batch([
    ("user:alice", "can_edit", "document:readme"),
    ("user:bob", "can_view", "document:readme"),
])
assert results.all_allowed()

Sync API

from inferadb import InferaDBSync

client = InferaDBSync(url="https://engine.inferadb.com", token="...")
vault = client.organization("my-org").vault("production")

allowed = vault.check("user:alice", "can_edit", "document:readme")

Relationships

Write

# Returns a revision token
token = await vault.relationships.write(
    resource="document:readme",
    relation="editor",
    subject="user:alice",
)

Batch Write

await vault.relationships.write_batch([
    {"resource": "document:readme", "relation": "editor", "subject": "user:alice"},
    {"resource": "document:readme", "relation": "viewer", "subject": "user:bob"},
])

List

rels = await vault.relationships.list(resource="document:readme")

Delete

await vault.relationships.delete_where(
    resource="document:readme",
    relation="viewer",
    subject="user:bob",
)

Lookups

# What resources can Alice view?
resources = await vault.resources.accessible_by(
    "user:alice",
    permission="can_view",
    resource_type="document",
)

# Who can edit this document?
subjects = await vault.subjects.with_permission(
    "can_edit",
    resource="document:readme",
)

Testing

MockClient (Fastest)

Stub specific check results. Call assert_expectations() to verify all stubs were hit.

from inferadb.testing import MockClient

client = (
    MockClient()
    .on_check("user:alice", "can_edit", "document:readme").allow()
    .on_check("user:bob", "can_edit", "document:readme").deny()
    .on_check_any_subject("can_view", "document:readme").allow()
    .default_deny()
)

client.assert_expectations()

InMemoryClient (Full Policy Evaluation)

Evaluates policies locally with no I/O. Synchronous construction.

from inferadb.testing import InMemoryClient

client = InMemoryClient.with_schema_and_data(
    schema="""
    type document {
        relation viewer
        relation editor
        relation can_view = viewer | editor
    }
    """,
    data=[
        ("document:readme", "editor", "user:alice"),
        ("document:readme", "viewer", "user:bob"),
    ],
)

TestVault (Real Instance)

Runs against a live InferaDB instance. Auto-cleans up on GC; call vault.preserve() to keep data for debugging.

from inferadb.testing import TestVault

vault = await TestVault.create(org, schema=schema_ipl)

pytest Integration

import pytest
from inferadb.testing import InMemoryClient

@pytest.fixture
def authz():
    return InMemoryClient.with_schema_and_data(
        schema="...",
        data=[("document:readme", "editor", "user:alice")],
    )

async def test_alice_can_edit(authz):
    vault = authz.organization("test").vault("test")
    assert await vault.check("user:alice", "can_edit", "document:readme")

Requires asyncio_mode = "auto" in pyproject.toml.

Error Handling

from inferadb import AccessDeniedError, InferaDBError

try:
    await vault.require("user:alice", "can_edit", "document:readme")
except AccessDeniedError:
    # permission denied
    pass
except InferaDBError as e:
    if e.is_retriable:
        # retry after e.retry_after seconds
        pass
    print(e.kind, e.request_id)

ErrorKind values: "unauthorized", "forbidden", "not_found", "rate_limited", "schema_violation", "unavailable", "timeout", "invalid_argument".

Framework Integrations

FastAPI Dependency

from fastapi import Depends, Request
from inferadb.fastapi import require_permission

@app.get("/documents/{doc_id}")
async def get_document(
    doc_id: str,
    request: Request,
    _=Depends(require_permission(
        vault,
        subject=lambda req: f"user:{req.state.user_id}",
        permission="can_view",
    )),
):
    return {"id": doc_id}

Resource is resolved from path parameters automatically. Override with an explicit resource callback:

require_permission(
    vault,
    subject=lambda req: f"user:{req.state.user_id}",
    resource=lambda req: f"document:{req.path_params['doc_id']}",
    permission="can_view",
)

Django Decorator

from inferadb.django import require_permission

@require_permission(
    vault,
    subject=lambda request: f"user:{request.user.id}",
    resource=lambda request, pk: f"document:{pk}",
    permission="can_edit",
)
def update_document(request, pk):
    # only reached if authorized
    pass

Edit on GitHub (opens in new tab)