### Install Authzed Python Client

Source: https://github.com/authzed/authzed-py/blob/main/README.md

Install the authzed library using pip. This is the first step to using the client.

```sh
pip install authzed
```

--------------------------------

### Install authzed with pip

Source: https://github.com/authzed/authzed-py/blob/main/examples/materialize/QUICKSTART.md

Install the authzed library using pip.

```bash
# Or with pip
pip install authzed
```

--------------------------------

### Install authzed-py

Source: https://context7.com/authzed/authzed-py/llms.txt

Install the authzed-py library using pip or uv.

```bash
pip install authzed
# or with uv
uv add authzed
```

--------------------------------

### Install authzed with UV

Source: https://github.com/authzed/authzed-py/blob/main/examples/materialize/QUICKSTART.md

Install the authzed library using the UV package manager.

```bash
# Install authzed with UV
uv add authzed
```

--------------------------------

### Local Development with InsecureClient

Source: https://github.com/authzed/authzed-py/blob/main/examples/materialize/QUICKSTART.md

Connect to local Materialize instances without TLS using InsecureClient. This example fetches a limited number of permission sets.

```python
from authzed.api.materialize.v0 import (
    InsecureClient,
    LookupPermissionSetsRequest,
)

client = InsecureClient("localhost:50051", "your_local_token")

request = LookupPermissionSetsRequest(limit=100)
response_stream = client.LookupPermissionSets(request)

for response in response_stream:
    # Process responses
    pass
```

--------------------------------

### Async Pagination for Permission Sets

Source: https://github.com/authzed/authzed-py/blob/main/examples/materialize/QUICKSTART.md

Demonstrates how to paginate through permission sets asynchronously using `AsyncClient`. This is useful for fetching large numbers of permission sets efficiently. Ensure you have `grpcio` and `authzed-python` installed.

```python
import asyncio
import grpc
from authzed.api.materialize.v0 import (
    AsyncClient,
    LookupPermissionSetsRequest,
)
from grpcutil import bearer_token_credentials

async def paginate_permission_sets():
    client = AsyncClient(
        "grpc.authzed.com:443",
        bearer_token_credentials("your_materialize_token"),
    )

    cursor = None
    total_processed = 0
    batch_size = 1000

    while True:
        request = LookupPermissionSetsRequest(
            limit=batch_size,
            optional_starting_after_cursor=cursor,
        )

        count = 0

        try:
            response_stream = client.LookupPermissionSets(request)

            async for response in response_stream:
                # Process response
                change = response.change
                parent = change.parent_set
                print(f"Processed: {parent.object_type}:{parent.object_id}#{parent.permission_or_relation}")

                count += 1
                total_processed += 1

                # Update cursor for next batch
                if response.HasField("cursor"):
                    cursor = response.cursor

            print(f"Batch complete: {count} permission sets (total: {total_processed})")

            # If we got fewer results than limit, we're done
            if count < batch_size:
                break

        except grpc.aio.AioRpcError as e:
            print(f"Error: {e.code()}: {e.details()}")
            break

    print(f"Pagination complete! Total: {total_processed}")

# Run the async function
asyncio.run(paginate_permission_sets())
```

--------------------------------

### Run unit tests with pytest

Source: https://github.com/authzed/authzed-py/blob/main/CONTRIBUTING.md

Execute unit tests using `pytest` with verbose output. This command should be run after starting a spicedb testing server.

```sh
uv run pytest -vv .
```

--------------------------------

### Pass linters with ruff

Source: https://github.com/authzed/authzed-py/blob/main/CONTRIBUTING.md

Run the `ruff` linter to check code quality and style. This can be done via `uv run` or by installing `ruff` globally.

```sh
uv run ruff
```

```sh
pipx install ruff
ruff
```

--------------------------------

### authzed.api.v1.SyncClient / AsyncClient

Source: https://context7.com/authzed/authzed-py/llms.txt

Use SyncClient or AsyncClient to bypass auto-detection and get precise type-checking IDE support.

```APIDOC
## authzed.api.v1.SyncClient / AsyncClient

### Description
Use `SyncClient` or `AsyncClient` to bypass auto-detection and get precise type-checking IDE support.

### Usage
```python
from authzed.api.v1 import AsyncClient, SyncClient
from grpcutil import bearer_token_credentials, insecure_bearer_token_credentials

# Explicitly synchronous
sync_client = SyncClient(
    "grpc.authzed.com:443",
    bearer_token_credentials("t_your_token_here_1234567deadbeef"),
)

# Explicitly asynchronous (must be used inside an event loop)
import asyncio

async def main():
    async_client = AsyncClient(
        "grpc.authzed.com:443",
        bearer_token_credentials("t_your_token_here_1234567deadbeef"),
    )
    resp = await async_client.ReadSchema(ReadSchemaRequest())
    print(resp.schema_text)

asyncio.run(main())
```
```

--------------------------------

### Check Permissions with Authzed Client

Source: https://github.com/authzed/authzed-py/blob/main/README.md

Perform a permission check using the initialized Authzed client. This example checks if 'emilia' has 'reader' permission on 'blog/post:1'.

```python
from authzed.api.v1 import (
    CheckPermissionRequest,
    CheckPermissionResponse,
    ObjectReference,
    SubjectReference,
)


post_one = ObjectReference(object_type="blog/post", object_id="1")
emilia = SubjectReference(object=ObjectReference(
    object_type="blog/user",
    object_id="emilia",
)) # Is Emilia in the set of users that can read post #1?
resp = client.CheckPermission(CheckPermissionRequest(
    resource=post_one,
    permission="reader",
    subject=emilia,
))
assert resp.permissionship == CheckPermissionResponse.PERMISSIONSHIP_HAS_PERMISSION
```

--------------------------------

### Async Usage with AsyncClient

Source: https://github.com/authzed/authzed-py/blob/main/examples/materialize/QUICKSTART.md

Perform asynchronous operations with the Materialize API using AsyncClient. This example demonstrates fetching permission sets asynchronously.

```python
import asyncio
from authzed.api.materialize.v0 import (
    AsyncClient,
    LookupPermissionSetsRequest,
)
from grpcutil import bearer_token_credentials

async def fetch_permission_sets():
    client = AsyncClient(
        "grpc.authzed.com:443",
        bearer_token_credentials("your_materialize_token"),
    )
    request = LookupPermissionSetsRequest(limit=1000)

    response_stream = client.LookupPermissionSets(request)

    async for response in response_stream:
        # Process responses asynchronously
        pass

asyncio.run(fetch_permission_sets())
```

--------------------------------

### Explicit SyncClient and AsyncClient variants

Source: https://context7.com/authzed/authzed-py/llms.txt

Use `SyncClient` or `AsyncClient` to bypass auto-detection and get precise type-checking IDE support. The asynchronous client must be used inside an event loop.

```python
from authzed.api.v1 import AsyncClient, SyncClient
from grpcutil import bearer_token_credentials, insecure_bearer_token_credentials

# Explicitly synchronous
sync_client = SyncClient(
    "grpc.authzed.com:443",
    bearer_token_credentials("t_your_token_here_1234567deadbeef"),
)

# Explicitly asynchronous (must be used inside an event loop)
import asyncio

async def main():
    async_client = AsyncClient(
        "grpc.authzed.com:443",
        bearer_token_credentials("t_your_token_here_1234567deadbeef"),
    )
    resp = await async_client.ReadSchema(ReadSchemaRequest())
    print(resp.schema_text)

asyncio.run(main())
```

--------------------------------

### Bulk Import and Export Relationships with Python

Source: https://context7.com/authzed/authzed-py/llms.txt

Use `BulkImportRelationships` for streaming import and `BulkExportRelationships` for streaming export of full dataset snapshots. Ensure you have the necessary client setup and schema defined.

```python
from authzed.api.v1 import (
    BulkExportRelationshipsRequest,
    BulkImportRelationshipsRequest,
    Client, Consistency, ObjectReference, Relationship, SubjectReference, WriteSchemaRequest,
)
from grpcutil import insecure_bearer_token_credentials

client = Client("localhost:50051", insecure_bearer_token_credentials("t_your_token_here_1234567deadbeef"))

schema = """
    definition user {}
    definition post {
        relation reader: user
        relation writer: user
        permission view  = reader + writer
        permission write = writer
    }
"""
client.WriteSchema(WriteSchemaRequest(schema=schema))

emilia   = SubjectReference(object=ObjectReference(object_type="user", object_id="emilia"))
beatrice = SubjectReference(object=ObjectReference(object_type="user", object_id="beatrice"))
post_one = ObjectReference(object_type="post", object_id="1")

# Import two relationships in one request
import_resp = client.BulkImportRelationships(
    req for req in [
        BulkImportRelationshipsRequest(relationships=[
            Relationship(resource=post_one, relation="writer", subject=emilia),
            Relationship(resource=post_one, relation="reader", subject=beatrice),
        ])
    ]
)
assert import_resp.num_loaded == 2

# Export all relationships
rels = []
for page in client.BulkExportRelationships(
    BulkExportRelationshipsRequest(consistency=Consistency(fully_consistent=True))
):
    rels.extend(page.relationships)

print(len(rels))  # 2
for r in rels:
    print(f"{r.resource.object_type}:{r.resource.object_id}#{r.relation} -> {r.subject.object_id}")
# post:1#writer -> emilia
# post:1#reader -> beatrice
```

--------------------------------

### Sync all dependencies with uv

Source: https://github.com/authzed/authzed-py/blob/main/CONTRIBUTING.md

Execute the `uv sync` command to update all project dependencies to their latest compatible versions.

```sh
uv sync
```

--------------------------------

### Initial Backfill with LookupPermissionSets

Source: https://github.com/authzed/authzed-py/blob/main/examples/materialize/QUICKSTART.md

Use LookupPermissionSets to perform the initial backfill of permission sets. This code fetches permission sets and prints their parent information.

```python
from authzed.api.materialize.v0 import (
    Client,
    LookupPermissionSetsRequest,
)
from grpcutil import bearer_token_credentials

# Create client
client = Client(
    "grpc.authzed.com:443",
    bearer_token_credentials("your_materialize_token"),
)

# Fetch permission sets
request = LookupPermissionSetsRequest(limit=1000)
response_stream = client.LookupPermissionSets(request)

for response in response_stream:
    change = response.change
    parent = change.parent_set

    # Store in your database
    print(f"Parent: {parent.object_type}:{parent.object_id}#{parent.permission_or_relation}")

    if change.HasField("child_member"):
        member = change.child_member
        print(f"  -> Member: {member.object_type}:{member.object_id}")
```

--------------------------------

### Check Permission with Consistency

Source: https://context7.com/authzed/authzed-py/llms.txt

Demonstrates checking permissions with different consistency requirements. Use 'at_least_as_fresh' for consistency with recent writes, 'at_exact_snapshot' for a specific revision, or 'fully_consistent' for the most up-to-date but potentially slower check.

```python
from authzed.api.v1 import CheckPermissionRequest, Consistency, ZedToken

# At least as fresh as a previous write (common pattern: use written_at token)
write_resp = client.WriteRelationships(...)
req = CheckPermissionRequest(
    consistency=Consistency(at_least_as_fresh=write_resp.written_at), ...
)

# Pinned to an exact snapshot
req = CheckPermissionRequest(
    consistency=Consistency(at_exact_snapshot=ZedToken(token="GhYK...")), ...
)

# Fully consistent — always hits the primary (slowest but always current)
req = CheckPermissionRequest(
    consistency=Consistency(fully_consistent=True), ...
)
```

--------------------------------

### Materialize API Client Initialization

Source: https://context7.com/authzed/authzed-py/llms.txt

Initializes the Materialize API client for streaming permission set changes. Requires a gRPC endpoint and bearer token credentials.

```python
from authzed.api.materialize.v0 import Client, LookupPermissionSetsRequest
from grpcutil import bearer_token_credentials

client = Client(
    "grpc.authzed.com:443",
    bearer_token_credentials("your_materialize_token"),
)
```

--------------------------------

### Async Materialize API Initial Backfill

Source: https://context7.com/authzed/authzed-py/llms.txt

Performs an initial backfill of permission sets asynchronously using the Materialize API. Handles potential gRPC errors during the process.

```python
import asyncio
import grpc
from authzed.api.materialize.v0 import AsyncClient, LookupPermissionSetsRequest
from grpcutil import bearer_token_credentials

async def backfill_async():
    client = AsyncClient(
        "grpc.authzed.com:443",
        bearer_token_credentials("your_materialize_token"),
    )
    cursor = None
    batch_size = 1000
    total = 0

    while True:
        request = LookupPermissionSetsRequest(
            limit=batch_size,
            optional_starting_after_cursor=cursor,
        )
        count = 0
        try:
            async for resp in client.LookupPermissionSets(request):
                parent = resp.change.parent_set
                print(f"{parent.object_type}:{parent.object_id}#{parent.permission_or_relation}")
                count += 1
                total += 1
                if resp.HasField("cursor"):
                    cursor = resp.cursor
        except grpc.aio.AioRpcError as e:
            print(f"RPC error {e.code()}: {e.details()}")
            break
        print(f"Batch: {count} (total: {total})")
        if count < batch_size:
            break
    print(f"Done — {total} permission sets")

asyncio.run(backfill_async())
```

--------------------------------

### Add a new dependency with uv

Source: https://github.com/authzed/authzed-py/blob/main/CONTRIBUTING.md

Use the `uv add` command to incorporate a new library into the project's dependencies.

```sh
uv add library
```

--------------------------------

### Materialize API Initial Backfill

Source: https://context7.com/authzed/authzed-py/llms.txt

Performs an initial backfill of all existing permission sets using the Materialize API. It pages through results until all sets are retrieved.

```python
cursor = None
total = 0
while True:
    request = LookupPermissionSetsRequest(
        limit=1000,
        optional_starting_after_cursor=cursor,
    )
    count = 0
    for resp in client.LookupPermissionSets(request):
        parent = resp.change.parent_set
        print(f"{parent.object_type}:{parent.object_id}#{parent.permission_or_relation}")
        count += 1
        total += 1
        if resp.HasField("cursor"):
            cursor = resp.cursor
    print(f"Batch: {count} (total: {total})")
    if count < 1000:
        break
print(f"Backfill complete — {total} permission sets")
```

--------------------------------

### Initialize Insecure Authzed Client

Source: https://github.com/authzed/authzed-py/blob/main/README.md

Initialize an insecure Authzed client, suitable for development or testing environments where TLS is not enforced. Requires a gRPC endpoint and a secret token.

```python
from authzed.api.v1 import InsecureClient

client = InsecureClient(
    "spicedb:50051",
    "my super secret token"
)
```

--------------------------------

### Auto-detecting v1 client for sync/async

Source: https://context7.com/authzed/authzed-py/llms.txt

The standard entry point for the v1 API. It automatically detects whether it's running inside an asyncio event loop and creates the appropriate gRPC channel (async or sync).

```python
from authzed.api.v1 import Client
from grpcutil import bearer_token_credentials

# Sync context — uses grpc.secure_channel
client = Client(
    "grpc.authzed.com:443",
    bearer_token_credentials("t_your_token_here_1234567deadbeef"),
)

# Async context — uses grpc.aio.secure_channel (call inside an async function/event loop)
import asyncio

async def main():
    client = Client(
        "grpc.authzed.com:443",
        bearer_token_credentials("t_your_token_here_1234567deadbeef"),
    )
    # client is now an async gRPC client
```

--------------------------------

### Watch Relationship and Schema Changes with Python

Source: https://context7.com/authzed/authzed-py/llms.txt

Subscribe to a live stream of mutations using `client.Watch`. Each `WatchResponse` includes `RelationshipUpdate` objects and a `ZedToken` for checkpointing. Optionally filter by object types.

```python
from authzed.api.v1 import Client, WatchRequest, ZedToken
from grpcutil import bearer_token_credentials

client = Client("grpc.authzed.com:443", bearer_token_credentials("t_your_token_here_1234567deadbeef"))

# Optionally filter to specific object types; omit for all changes
stream = client.Watch(
    WatchRequest(
        optional_object_types=["blog/post"],
        # optional_start_cursor=ZedToken(token="<checkpoint_token>"),
    )
)

for resp in stream:
    for update in resp.updates:
        op   = update.operation  # OPERATION_CREATE / OPERATION_DELETE / OPERATION_TOUCH
        rel  = update.relationship
        print(f"[{op}] {rel.resource.object_type}:{rel.resource.object_id}"
              f"#{rel.relation} -> {rel.subject.object.object_id}")
    print(f"Checkpoint: {resp.changes_through.token}")
```

--------------------------------

### Local Materialize Dev Client

Source: https://context7.com/authzed/authzed-py/llms.txt

Initializes and uses an insecure client for local development with the Materialize API. Connects to a local gRPC server.

```python
from authzed.api.materialize.v0 import InsecureClient, LookupPermissionSetsRequest

client = InsecureClient("localhost:50051", "your_local_token")

for resp in client.LookupPermissionSets(LookupPermissionSetsRequest(limit=100)):
    change = resp.change
    print(change.parent_set.object_type, change.parent_set.object_id)
```

--------------------------------

### Client.LookupPermissionSets

Source: https://context7.com/authzed/authzed-py/llms.txt

Performs an initial backfill by paginating through all existing permission sets. This method is used to retrieve a snapshot of all permission sets at a given point in time.

```APIDOC
## Client.LookupPermissionSets

### Description
Retrieves a paginated list of all permission sets. This is typically used for an initial backfill of a local data store.

### Method
`LookupPermissionSets(request: LookupPermissionSetsRequest)`

### Parameters
#### Request Body
- **limit** (int) - Required - The maximum number of permission sets to return in this batch.
- **optional_starting_after_cursor** (bytes) - Optional - A cursor to start retrieving permission sets after. If not provided, retrieval starts from the beginning.
```

--------------------------------

### Run commands with uv

Source: https://github.com/authzed/authzed-py/blob/main/CONTRIBUTING.md

Utilize `uv run` to execute project-specific commands, such as linters or test runners, within the project's environment.

```sh
uv run pytest
```

--------------------------------

### Lookup Resources a Subject Can Access

Source: https://context7.com/authzed/authzed-py/llms.txt

Streams all resources of a given type that a subject can access. Useful for 'what can this user see?' lists. Ensure consistency is set appropriately.

```python
from authzed.api.v1 import (
    Client, Consistency, LookupResourcesRequest, ObjectReference, SubjectReference,
)
from grpcutil import bearer_token_credentials

client = Client("grpc.authzed.com:443", bearer_token_credentials("t_your_token_here_1234567deadbeef"))

emilia = SubjectReference(object=ObjectReference(object_type="user", object_id="emilia"))

stream = client.LookupResources(
    LookupResourcesRequest(
        resource_object_type="post",
        permission="write",
        subject=emilia,
        consistency=Consistency(fully_consistent=True),
    )
)

accessible_posts = [resp.resource_object_id for resp in stream]
print(accessible_posts)  # ['post-one', 'post-two']
```

--------------------------------

### Initialize Authzed Client with Bearer Token

Source: https://github.com/authzed/authzed-py/blob/main/README.md

Initialize the Authzed client for API v1. Requires a gRPC endpoint and bearer token credentials. Replace 't_your_token_here_1234567deadbeef' with your actual API token.

```python
from authzed.api.v1 import Client
from grpcutil import bearer_token_credentials


client = Client(
    "grpc.authzed.com:443",
    bearer_token_credentials("t_your_token_here_1234567deadbeef"),
)
```

--------------------------------

### authzed.api.v1.Client

Source: https://context7.com/authzed/authzed-py/llms.txt

The standard entry point for the v1 API. Detects at construction time whether it is running inside an asyncio event loop and creates the appropriate gRPC channel (async or sync) automatically.

```APIDOC
## authzed.api.v1.Client

### Description
The standard entry point for the v1 API. Detects at construction time whether it is running inside an `asyncio` event loop and creates the appropriate gRPC channel (async or sync) automatically. Implements `SchemaServiceStub`, `PermissionsServiceStub`, `ExperimentalServiceStub`, and `WatchServiceStub`.

### Usage
```python
from authzed.api.v1 import Client
from grpcutil import bearer_token_credentials

# Sync context — uses grpc.secure_channel
client = Client(
    "grpc.authzed.com:443",
    bearer_token_credentials("t_your_token_here_1234567deadbeef"),
)

# Async context — uses grpc.aio.secure_channel (call inside an async function/event loop)
import asyncio

async def main():
    client = Client(
        "grpc.authzed.com:443",
        bearer_token_credentials("t_your_token_here_1234567deadbeef"),
    )
    # client is now an async gRPC client

asyncio.run(main())
```
```

--------------------------------

### AsyncClient.LookupPermissionSets

Source: https://context7.com/authzed/authzed-py/llms.txt

Asynchronous version of `LookupPermissionSets` for initial backfilling. Allows for non-blocking retrieval of permission sets.

```APIDOC
## AsyncClient.LookupPermissionSets

### Description
An asynchronous method to retrieve paginated permission sets, suitable for initial data backfilling in non-blocking applications.

### Method
`async for resp in client.LookupPermissionSets(request: LookupPermissionSetsRequest)`

### Parameters
#### Request Body
- **limit** (int) - Required - The maximum number of permission sets to return in this batch.
- **optional_starting_after_cursor** (bytes) - Optional - A cursor to start retrieving permission sets after. If not provided, retrieval starts from the beginning.
```

--------------------------------

### client.LookupResources

Source: https://context7.com/authzed/authzed-py/llms.txt

Streaming RPC that returns every resource of a given type on which the subject holds the requested permission. Useful for building "what can this user see?" lists.

```APIDOC
## `client.LookupResources` — Find all resources a subject can access

### Description
Streaming RPC that returns every resource of a given type on which the subject holds the requested permission. Useful for building "what can this user see?" lists.

### Method
POST (Assumed, as it's a lookup operation)

### Endpoint
`/authzed.api.v1.PermissionService/LookupResources` (Assumed gRPC endpoint)

### Parameters
#### Request Body
- **resource_object_type** (string) - Required - The type of the resources to look up.
- **permission** (string) - Required - The permission to check.
- **subject** (SubjectReference) - Required - The subject for which to find accessible resources.
  - **object** (ObjectReference) - Required - The object representing the subject.
    - **object_type** (string) - Required - The type of the subject object.
    - **object_id** (string) - Required - The ID of the subject object.
- **consistency** (Consistency) - Optional - The consistency level for the lookup.
  - **fully_consistent** (boolean) - Optional - If true, ensures a fully consistent read.

### Request Example
```python
from authzed.api.v1 import Consistency, LookupResourcesRequest, ObjectReference, SubjectReference

emilia = SubjectReference(object=ObjectReference(object_type="user", object_id="emilia"))

request = LookupResourcesRequest(
    resource_object_type="post",
    permission="write",
    subject=emilia,
    consistency=Consistency(fully_consistent=True),
)
```

### Response
#### Success Response (Stream)
- **resource_object_id** (string) - The ID of a resource the subject can access.
```

--------------------------------

### client.WriteSchema

Source: https://context7.com/authzed/authzed-py/llms.txt

Upserts the entire schema for the permissions system using SpiceZDL. Returns a WriteSchemaResponse containing a written_at ZedToken.

```APIDOC
## `client.WriteSchema` — Define or update the permissions schema

Upserts the entire schema for the permissions system. The schema uses SpiceZDL (Zanzibar Definition Language) to declare object types, relations, and computed permissions. Returns a `WriteSchemaResponse` containing a `written_at` `ZedToken`.

```python
from authzed.api.v1 import Client, WriteSchemaRequest
from grpcutil import bearer_token_credentials

SCHEMA = """
definition blog/user {}

definition blog/post {
    relation reader: blog/user
    relation writer: blog/user

    permission read  = reader + writer
    permission write = writer
}
"""

client = Client("grpc.authzed.com:443", bearer_token_credentials("t_your_token_here_1234567deadbeef"))
resp = client.WriteSchema(WriteSchemaRequest(schema=SCHEMA))
print(resp.written_at.token)   # e.g. "GhYKEzE3NDc..."
```
```

--------------------------------

### Check Permission with Caveats

Source: https://context7.com/authzed/authzed-py/llms.txt

Use `client.CheckPermission` with a `google.protobuf.struct_pb2.Struct` for `context` to evaluate parameterized caveats. The response will be `PERMISSIONSHIP_CONDITIONAL_PERMISSION` if context is missing.

```python
from google.protobuf.struct_pb2 import Struct
from authzed.api.v1 import (
    CheckPermissionRequest, CheckPermissionResponse,
    Client, Consistency, ObjectReference, SubjectReference,
)
from grpcutil import bearer_token_credentials

# Schema includes: caveat likes_harry_potter(likes bool) { likes == true }
# and permission view_as_fan = caveated_reader + writer

client = Client("grpc.authzed.com:443", bearer_token_credentials("t_your_token_here_1234567deadbeef"))
post_one = ObjectReference(object_type="post", object_id="post-one")
beatrice = SubjectReference(object=ObjectReference(object_type="user", object_id="beatrice"))

# With context: likes = true → PERMISSIONSHIP_HAS_PERMISSION
ctx = Struct()
ctx.update({"likes": True})
resp = client.CheckPermission(CheckPermissionRequest(
    resource=post_one, permission="view_as_fan",
    subject=beatrice, consistency=Consistency(fully_consistent=True),
    context=ctx,
))
assert resp.permissionship == CheckPermissionResponse.PERMISSIONSHIP_HAS_PERMISSION

# With context: likes = false → PERMISSIONSHIP_NO_PERMISSION
ctx.update({"likes": False})
resp = client.CheckPermission(CheckPermissionRequest(
    resource=post_one, permission="view_as_fan",
    subject=beatrice, consistency=Consistency(fully_consistent=True),
    context=ctx,
))
assert resp.permissionship == CheckPermissionResponse.PERMISSIONSHIP_NO_PERMISSION

# Without context → PERMISSIONSHIP_CONDITIONAL_PERMISSION
resp = client.CheckPermission(CheckPermissionRequest(
    resource=post_one, permission="view_as_fan",
    subject=beatrice, consistency=Consistency(fully_consistent=True),
))
assert resp.permissionship == CheckPermissionResponse.PERMISSIONSHIP_CONDITIONAL_PERMISSION
assert "likes" in resp.partial_caveat_info.missing_required_context
```

--------------------------------

### Watch

Source: https://context7.com/authzed/authzed-py/llms.txt

Subscribes to a live stream of relationship and schema changes. Each WatchResponse contains a list of RelationshipUpdate objects and a changes_through ZedToken for checkpointing.

```APIDOC
## `client.Watch` — Stream relationship and schema changes

Subscribes to a live stream of mutations. Each `WatchResponse` contains a list of `RelationshipUpdate` objects and a `changes_through` `ZedToken` for checkpointing.

```python
from authzed.api.v1 import Client, WatchRequest, ZedToken
from grpcutil import bearer_token_credentials

client = Client("grpc.authzed.com:443", bearer_token_credentials("t_your_token_here_1234567deadbeef"))

# Optionally filter to specific object types; omit for all changes
stream = client.Watch(
    WatchRequest(
        optional_object_types=["blog/post"],
        # optional_start_cursor=ZedToken(token="<checkpoint_token>"),
    )
)

for resp in stream:
    for update in resp.updates:
        op   = update.operation  # OPERATION_CREATE / OPERATION_DELETE / OPERATION_TOUCH
        rel  = update.relationship
        print(f"[{op}] {rel.resource.object_type}:{rel.resource.object_id}"
              f"#{rel.relation} -> {rel.subject.object.object_id}")
    print(f"Checkpoint: {resp.changes_through.token}")
```
```

--------------------------------

### Local/test credentials with insecure bearer token

Source: https://context7.com/authzed/authzed-py/llms.txt

Creates credentials for local or Docker-based SpiceDB instances where TLS is not used. This is for testing purposes only.

```python
from grpcutil import insecure_bearer_token_credentials

creds = insecure_bearer_token_credentials("sometoken")
# Use with Client, SyncClient, AsyncClient, or InsecureClient targeting localhost
```

--------------------------------

### InsecureClient.LookupPermissionSets

Source: https://context7.com/authzed/authzed-py/llms.txt

A client for local development that connects without TLS. Used for testing and local Materialize instances.

```APIDOC
## InsecureClient.LookupPermissionSets

### Description
Provides access to Materialize functionality over an insecure connection, intended for local development and testing environments.

### Method
`LookupPermissionSets(request: LookupPermissionSetsRequest)`

### Parameters
#### Request Body
- **limit** (int) - Required - The maximum number of permission sets to return in this batch.
- **optional_starting_after_cursor** (bytes) - Optional - A cursor to start retrieving permission sets after. If not provided, retrieval starts from the beginning.
```

--------------------------------

### Local development InsecureClient

Source: https://context7.com/authzed/authzed-py/llms.txt

Skips TLS entirely for local environments where the gRPC server is not on `localhost` but TLS is unavailable. Accepts a raw token string.

```python
from authzed.api.v1 import InsecureClient

client = InsecureClient(
    "spicedb:50051",       # Docker service name or remote host without TLS
    "my super secret token",
)
resp = client.ReadSchema(ReadSchemaRequest())
```

--------------------------------

### Synchronous Pagination with Cursor

Source: https://github.com/authzed/authzed-py/blob/main/examples/materialize/QUICKSTART.md

Handle large datasets with cursor-based pagination for synchronous operations. This loop processes permission sets in batches until all are retrieved.

```python
from authzed.api.materialize.v0 import (
    Client,
    LookupPermissionSetsRequest,
)
from grpcutil import bearer_token_credentials

client = Client(
    "grpc.authzed.com:443",
    bearer_token_credentials("your_materialize_token"),
)

cursor = None
total_processed = 0

while True:
    request = LookupPermissionSetsRequest(
        limit=1000,
        optional_starting_after_cursor=cursor,
    )

    count = 0
    for response in client.LookupPermissionSets(request):
        # Process response
        change = response.change
        parent = change.parent_set
        print(f"Processed: {parent.object_type}:{parent.object_id}#{parent.permission_or_relation}")

        count += 1
        total_processed += 1
        cursor = response.cursor if response.HasField("cursor") else None

    print(f"Batch complete: {count} permission sets (total: {total_processed})")

    # If we got fewer results than limit, we're done
    if count < 1000:
        break

print(f"Pagination complete! Total: {total_processed}")
```

--------------------------------

### client.CheckPermission with caveats

Source: https://context7.com/authzed/authzed-py/llms.txt

Evaluates permissions with parameterized conditions (caveats) supplied via context. If context is missing, it returns PERMISSIONSHIP_CONDITIONAL_PERMISSION.

```APIDOC
## `client.CheckPermission` with caveats — Contextual permission evaluation

Caveats are parameterized conditions evaluated at check time. Pass a `google.protobuf.struct_pb2.Struct` as `context` to supply caveat parameters. When context is missing, the response is `PERMISSIONSHIP_CONDITIONAL_PERMISSION`.

```python
from google.protobuf.struct_pb2 import Struct
from authzed.api.v1 import (
    CheckPermissionRequest, CheckPermissionResponse,
    Client, Consistency, ObjectReference, SubjectReference,
)
from grpcutil import bearer_token_credentials

# Schema includes: caveat likes_harry_potter(likes bool) { likes == true }
# and permission view_as_fan = caveated_reader + writer

client = Client("grpc.authzed.com:443", bearer_token_credentials("t_your_token_here_1234567deadbeef"))
post_one = ObjectReference(object_type="post", object_id="post-one")
beatrice = SubjectReference(object=ObjectReference(object_type="user", object_id="beatrice"))

# With context: likes = true → PERMISSIONSHIP_HAS_PERMISSION
ctx = Struct()
ctx.update({"likes": True})
resp = client.CheckPermission(CheckPermissionRequest(
    resource=post_one, permission="view_as_fan",
    subject=beatrice, consistency=Consistency(fully_consistent=True),
    context=ctx,
))
assert resp.permissionship == CheckPermissionResponse.PERMISSIONSHIP_HAS_PERMISSION

# With context: likes = false → PERMISSIONSHIP_NO_PERMISSION
ctx.update({"likes": False})
resp = client.CheckPermission(CheckPermissionRequest(
    resource=post_one, permission="view_as_fan",
    subject=beatrice, consistency=Consistency(fully_consistent=True),
    context=ctx,
))
assert resp.permissionship == CheckPermissionResponse.PERMISSIONSHIP_NO_PERMISSION

# Without context → PERMISSIONSHIP_CONDITIONAL_PERMISSION
resp = client.CheckPermission(CheckPermissionRequest(
    resource=post_one, permission="view_as_fan",
    subject=beatrice, consistency=Consistency(fully_consistent=True),
))
assert resp.permissionship == CheckPermissionResponse.PERMISSIONSHIP_CONDITIONAL_PERMISSION
assert "likes" in resp.partial_caveat_info.missing_required_context
```
```

--------------------------------

### Write Schema using authzed-py

Source: https://context7.com/authzed/authzed-py/llms.txt

Upserts the entire permissions schema using Zanzibar Definition Language (ZED). Requires a Client instance and a WriteSchemaRequest containing the schema string. Returns a ZedToken.

```python
from authzed.api.v1 import Client, WriteSchemaRequest
from grpcutil import bearer_token_credentials

SCHEMA = """
definition blog/user {}

definition blog/post {
    relation reader: blog/user
    relation writer: blog/user

    permission read  = reader + writer
    permission write = writer
}
"""

client = Client("grpc.authzed.com:443", bearer_token_credentials("t_your_token_here_1234567deadbeef"))
resp = client.WriteSchema(WriteSchemaRequest(schema=SCHEMA))
print(resp.written_at.token)   # e.g. "GhYKEzE3NDc..."

```

--------------------------------

### Production TLS credentials with bearer token

Source: https://context7.com/authzed/authzed-py/llms.txt

Creates gRPC channel credentials combining SSL/TLS with a Bearer token for production environments. Optionally supply a custom CA certificate chain.

```python
from grpcutil import bearer_token_credentials

# Standard TLS connection with a bearer token
creds = bearer_token_credentials("t_your_token_here_1234567deadbeef")

# Optionally supply a custom CA certificate chain (PEM bytes)
with open("ca-cert.pem", "rb") as f:
    creds = bearer_token_credentials("t_your_token_here_1234567deadbeef", certChain=f.read())
```

--------------------------------

### Update generated Protobuf code

Source: https://github.com/authzed/authzed-py/blob/main/CONTRIBUTING.md

Regenerate Protobuf code by executing the `buf.gen.yaml` file. This process uses `buf` and is configured by the Buf Registry ref specified in the shebang.

```sh
./buf.gen.yaml
```

--------------------------------

### Client.WatchPermissionSets

Source: https://context7.com/authzed/authzed-py/llms.txt

Streams incremental permission set changes after an initial backfill. This method provides a live stream of `PermissionSetChange` deltas to keep a local store synchronized.

```APIDOC
## Client.WatchPermissionSets

### Description
Streams real-time changes to permission sets. Use this after an initial backfill to apply incremental updates (`SET_OPERATION_ADDED`, `SET_OPERATION_REMOVED`) to a local data store.

### Method
`WatchPermissionSets(request: WatchPermissionSetsRequest)`

### Parameters
#### Request Body
- **optional_starting_after** (ZedToken) - Optional - The revision token to start watching from. Typically the last revision from a `LookupPermissionSets` call.
```

--------------------------------

### authzed.api.v1.InsecureClient

Source: https://context7.com/authzed/authzed-py/llms.txt

Skips TLS entirely for Docker Compose and other local environments where the gRPC server is not on localhost but TLS is unavailable.

```APIDOC
## authzed.api.v1.InsecureClient

### Description
Skips TLS entirely for Docker Compose and other local environments where the gRPC server is not on `localhost` but TLS is unavailable. Accepts a raw token string (no credential wrapper needed).

### Usage
```python
from authzed.api.v1 import InsecureClient

client = InsecureClient(
    "spicedb:50051",       # Docker service name or remote host without TLS
    "my super secret token",
)
resp = client.ReadSchema(ReadSchemaRequest())
```
```

--------------------------------

### grpcutil.insecure_bearer_token_credentials

Source: https://context7.com/authzed/authzed-py/llms.txt

Creates credentials suitable for local or Docker-based SpiceDB instances where TLS is not used. For testing only.

```APIDOC
## grpcutil.insecure_bearer_token_credentials

### Description
Creates credentials suitable for local or Docker-based SpiceDB instances where TLS is not used. **For testing only.**

### Usage
```python
from grpcutil import insecure_bearer_token_credentials

creds = insecure_bearer_token_credentials("sometoken")
# Use with Client, SyncClient, AsyncClient, or InsecureClient targeting localhost
```
```

--------------------------------

### Bulk Check Permissions

Source: https://context7.com/authzed/authzed-py/llms.txt

Use `client.BulkCheckPermission` to check multiple resource/permission/subject triples in a single RPC. The response `pairs` list mirrors the input order, with each item containing an individual `permissionship` result.

```python
from authzed.api.v1 import (
    BulkCheckPermissionRequest,
    BulkCheckPermissionRequestItem,
    CheckPermissionResponse,
    Client,
    Consistency,
    ObjectReference,
    SubjectReference,
)
from grpcutil import bearer_token_credentials

client = Client("grpc.authzed.com:443", bearer_token_credentials("t_your_token_here_1234567deadbeef"))

post_one = ObjectReference(object_type="blog/post", object_id="1")
beatrice = SubjectReference(object=ObjectReference(object_type="blog/user", object_id="beatrice"))

resp = client.BulkCheckPermission(
    BulkCheckPermissionRequest(
        consistency=Consistency(fully_consistent=True),
        items=[
            BulkCheckPermissionRequestItem(resource=post_one, permission="view",  subject=beatrice),
            BulkCheckPermissionRequestItem(resource=post_one, permission="write", subject=beatrice),
        ],
    )
)

assert len(resp.pairs) == 2
# pairs[0] corresponds to "view", pairs[1] to "write"
print(resp.pairs[0].item.permissionship)   # PERMISSIONSHIP_HAS_PERMISSION
print(resp.pairs[1].item.permissionship)   # PERMISSIONSHIP_NO_PERMISSION (if beatrice is reader only)
```

--------------------------------

### Control Read Freshness with Consistency in Python

Source: https://context7.com/authzed/authzed-py/llms.txt

Use the `Consistency` message in read operations to control read freshness versus latency. Options include minimizing latency (potentially stale reads) or anchoring to a known revision for fully consistent reads.

```python
from authzed.api.v1 import CheckPermissionRequest, Consistency, ZedToken

# Fastest: may read a slightly stale snapshot
req = CheckPermissionRequest(
    consistency=Consistency(minimize_latency=True), ...
)

```

--------------------------------

### Bulk Import Relationships

Source: https://context7.com/authzed/authzed-py/llms.txt

Imports a large number of relationships using a client-streaming RPC. SpiceDB commits each batch as a separate transaction. Requires schema to be written first.

```python
from itertools import batched
from authzed.api.v1 import Client, ObjectReference, Relationship, SubjectReference, WriteSchemaRequest
from authzed.api.v1.permission_service_pb2 import ImportBulkRelationshipsRequest
from grpcutil import insecure_bearer_token_credentials

client = Client("localhost:50051", insecure_bearer_token_credentials("sometoken"))

schema = """
definition user {}
definition resource {
    relation viewer: user
    permission view = viewer
}
"""
client.WriteSchema(WriteSchemaRequest(schema=schema))

def relationship_generator(n):
    for i in range(1, n + 1):
        yield Relationship(
            resource=ObjectReference(object_type="resource", object_id=str(i)),
            relation="viewer",
            subject=SubjectReference(
                object=ObjectReference(object_type="user", object_id="our_user")
            ),
        )

TOTAL = 1_000
PER_TRANSACTION = 100
PER_CHUNK = 10

for txn_batch in batched(relationship_generator(TOTAL), PER_TRANSACTION):
    resp = client.ImportBulkRelationships(
        ImportBulkRelationshipsRequest(relationships=chunk)
        for chunk in batched(txn_batch, PER_CHUNK)
    )
    print(f"Loaded {resp.num_loaded} relationships")
# Loaded 100 relationships  (×10 times)
```

--------------------------------

### grpcutil.bearer_token_credentials

Source: https://context7.com/authzed/authzed-py/llms.txt

Creates composite gRPC channel credentials combining SSL/TLS with a Bearer token. Used for all connections to hosted Authzed or remote SpiceDB instances.

```APIDOC
## grpcutil.bearer_token_credentials

### Description
Creates composite gRPC channel credentials combining SSL/TLS with a Bearer token. Used for all connections to hosted Authzed or remote SpiceDB instances.

### Usage
```python
from grpcutil import bearer_token_credentials

# Standard TLS connection with a bearer token
creds = bearer_token_credentials("t_your_token_here_1234567deadbeef")

# Optionally supply a custom CA certificate chain (PEM bytes)
with open("ca-cert.pem", "rb") as f:
    creds = bearer_token_credentials("t_your_token_here_1234567deadbeef", certChain=f.read())
```
```

--------------------------------

### client.BulkCheckPermission

Source: https://context7.com/authzed/authzed-py/llms.txt

Efficiently checks multiple permission requests in a single RPC call. The response pairs mirror the input order, providing the `permissionship` for each item.

```APIDOC
## `client.BulkCheckPermission` — Check multiple permissions in one RPC

Checks a list of resource/permission/subject triples in a single request. Returns a `BulkCheckPermissionResponse` with a `pairs` list that mirrors the input order, each carrying the individual `permissionship` result.

```python
from authzed.api.v1 import (
    BulkCheckPermissionRequest,
    BulkCheckPermissionRequestItem,
    CheckPermissionResponse,
    Client,
    Consistency,
    ObjectReference,
    SubjectReference,
)
from grpcutil import bearer_token_credentials

client = Client("grpc.authzed.com:443", bearer_token_credentials("t_your_token_here_1234567deadbeef"))

post_one = ObjectReference(object_type="blog/post", object_id="1")
beatrice = SubjectReference(object=ObjectReference(object_type="blog/user", object_id="beatrice"))

resp = client.BulkCheckPermission(
    BulkCheckPermissionRequest(
        consistency=Consistency(fully_consistent=True),
        items=[
            BulkCheckPermissionRequestItem(resource=post_one, permission="view",  subject=beatrice),
            BulkCheckPermissionRequestItem(resource=post_one, permission="write", subject=beatrice),
        ],
    )
)

assert len(resp.pairs) == 2
# pairs[0] corresponds to "view", pairs[1] to "write"
print(resp.pairs[0].item.permissionship)   # PERMISSIONSHIP_HAS_PERMISSION
print(resp.pairs[1].item.permissionship)   # PERMISSIONSHIP_NO_PERMISSION (if beatrice is reader only)
```
```

--------------------------------

### Watch for Updates with WatchPermissionSets

Source: https://github.com/authzed/authzed-py/blob/main/examples/materialize/QUICKSTART.md

After backfilling, watch for ongoing changes using WatchPermissionSets. This snippet handles adding and removing permission set operations.

```python
from authzed.api.materialize.v0 import (
    Client,
    WatchPermissionSetsRequest,
    PermissionSetChange,
)
from grpcutil import bearer_token_credentials

client = Client(
    "grpc.authzed.com:443",
    bearer_token_credentials("your_materialize_token"),
)

request = WatchPermissionSetsRequest(
    optional_starting_after=last_revision,  # From your last LookupPermissionSets backfill
)

response_stream = client.WatchPermissionSets(request)

for response in response_stream:
    change = response.change

    # Handle the operation type
    if change.operation == PermissionSetChange.SET_OPERATION_ADDED:
        # Add to materialized view
        pass
    elif change.operation == PermissionSetChange.SET_OPERATION_REMOVED:
        # Remove from materialized view
        pass
```

--------------------------------

### Consistency

Source: https://context7.com/authzed/authzed-py/llms.txt

Controls read freshness versus latency for read operations. Allows choosing between minimizing latency, anchoring to a known revision, or guaranteeing fully consistent reads.

```APIDOC
## `Consistency` — Control read freshness vs. latency

Every read operation accepts a `Consistency` message that governs which SpiceDB snapshot is used. Choose between minimizing latency, anchoring to a known revision, or guaranteeing fully consistent reads.

```python
from authzed.api.v1 import CheckPermissionRequest, Consistency, ZedToken

# Fastest: may read a slightly stale snapshot
req = CheckPermissionRequest(
    consistency=Consistency(minimize_latency=True), ...
)


```

--------------------------------

### Watch Materialize Permission Set Changes

Source: https://context7.com/authzed/authzed-py/llms.txt

Streams incremental permission set changes from the Materialize API after an initial backfill. Apply ADDED and REMOVED operations to keep a local store synchronized.

```python
from authzed.api.materialize.v0 import (
    Client, PermissionSetChange, WatchPermissionSetsRequest,
)
from grpcutil import bearer_token_credentials

client = Client("grpc.authzed.com:443", bearer_token_credentials("your_materialize_token"))

# last_revision is the ZedToken saved from the final LookupPermissionSets response
stream = client.WatchPermissionSets(
    WatchPermissionSetsRequest(optional_starting_after=last_revision)
)

for resp in stream:
    if resp.HasField("change"):
        change = resp.change
        parent = change.parent_set
        if change.operation == PermissionSetChange.SET_OPERATION_ADDED:
            # INSERT into your local permissions table
            print(f"ADD  {parent.object_type}:{parent.object_id}#{parent.permission_or_relation}")
        elif change.operation == PermissionSetChange.SET_OPERATION_REMOVED:
            # DELETE from your local permissions table
            print(f"DEL  {parent.object_type}:{parent.object_id}#{parent.permission_or_relation}")
    if resp.HasField("completed_revision"):
        # Safe to advance your checkpoint cursor
        last_revision = resp.completed_revision
```

--------------------------------

### client.LookupSubjects

Source: https://context7.com/authzed/authzed-py/llms.txt

Streaming RPC that returns every subject of a given type that holds the requested permission on a specific resource. Useful for building ACL lists and audit logs.

```APIDOC
## `client.LookupSubjects` — Find all subjects that can access a resource

### Description
Streaming RPC that returns every subject of a given type that holds the requested permission on a specific resource. Useful for building ACL lists and audit logs.

### Method
POST (Assumed, as it's a lookup operation)

### Endpoint
`/authzed.api.v1.PermissionService/LookupSubjects` (Assumed gRPC endpoint)

### Parameters
#### Request Body
- **resource** (ObjectReference) - Required - The resource for which to find subjects.
  - **object_type** (string) - Required - The type of the resource.
  - **object_id** (string) - Required - The ID of the resource.
- **permission** (string) - Required - The permission to check.
- **subject_object_type** (string) - Required - The type of subjects to look for.
- **consistency** (Consistency) - Optional - The consistency level for the lookup.
  - **fully_consistent** (boolean) - Optional - If true, ensures a fully consistent read.

### Request Example
```python
from authzed.api.v1 import Consistency, LookupSubjectsRequest, ObjectReference

post_one = ObjectReference(object_type="post", object_id="post-one")

request = LookupSubjectsRequest(
    resource=post_one,
    permission="view",
    subject_object_type="user",
    consistency=Consistency(fully_consistent=True),
)
```

### Response
#### Success Response (Stream)
- **subject_object_id** (string) - The ID of a subject that can access the resource.
```