openfga-sdk

Python SDK for OpenFGA

![Socket Badge](https://badge.socket.dev/pypi/package/openfga-sdk/0.9.9

This is an autogenerated python SDK for OpenFGA. It provides a wrapper around the OpenFGA API definition.

Table of Contents

• About OpenFGA
• Resources
• Installation
• Getting Started
  • Initializing the API Client
  • Custom Headers
  • Get your Store ID
  • Calling the API
    • Stores
      • List All Stores
      • Create a Store
      • Get a Store
      • Delete a Store
    • Authorization Models
      • Read Authorization Models
      • Write Authorization Model
      • Read a Single Authorization Model
      • Read the Latest Authorization Model
    • Relationship Tuples
      • Read Relationship Tuple Changes (Watch)
      • Read Relationship Tuples
      • Write (Create and Delete) Relationship Tuples
    • Relationship Queries
      • Check
      • Batch Check
      • Expand
      • List Objects
      • Streamed List Objects
      • List Relations
      • List Users
    • Assertions
      • Read Assertions
      • Write Assertions
  • Retries
  • API Endpoints
  • Models
  • OpenTelemetry
• Contributing
  • Issues
  • Pull Requests
• License

About

OpenFGA is an open source Fine-Grained Authorization solution inspired by Google's Zanzibar paper. It was created by the FGA team at Auth0 based on Auth0 Fine-Grained Authorization (FGA), available under a permissive license (Apache-2) and welcomes community contributions.

OpenFGA is designed to make it easy for application builders to model their permission layer, and to add and integrate fine-grained authorization into their applications. OpenFGA’s design is optimized for reliability and low latency at a high scale.

Resources

• OpenFGA Documentation
• OpenFGA API Documentation
• X
• OpenFGA Community
• Zanzibar Academy
• Google's Zanzibar Paper (2019)

Installation

pip install

PyPI

The openfga_sdk is available to be downloaded via PyPI, you can install directly using:

pip3 install openfga_sdk

(you may need to run pip with root permission: sudo pip3 install openfga_sdk)

Then import the package:

import openfga_sdk

GitHub

The openfga_sdk is also hosted in GitHub, you can install directly using:

pip3 install https://github.com/openfga/python-sdk.git

(you may need to run pip with root permission: sudo pip3 install https://github.com/openfga/python-sdk.git)

Then import the package:

import openfga_sdk

Setuptools

Install via Setuptools.

python setup.py install --user

(or sudo python setup.py install to install the package for all users)

Then import the package:

import openfga_sdk

Getting Started

Initializing the API Client

Learn how to initialize your SDK

We strongly recommend you initialize the OpenFgaClient only once and then re-use it throughout your app, otherwise you will incur the cost of having to re-initialize multiple times or at every request, the cost of reduced connection pooling and re-use, and would be particularly costly in the client credentials flow, as that flow will be preformed on every request.

The OpenFgaClient will by default retry API requests up to 3 times on 429 and 5xx errors.

No Credentials

from openfga_sdk import ClientConfiguration, OpenFgaClient


async def main():
    configuration = ClientConfiguration(
        api_url=FGA_API_URL,  # required
        store_id=FGA_STORE_ID,  # optional, not needed when calling `CreateStore` or `ListStores`
        authorization_model_id=FGA_MODEL_ID,  # Optional, can be overridden per request
    )
    # Enter a context with an instance of the OpenFgaClient
    async with OpenFgaClient(configuration) as fga_client:
        api_response = await fga_client.read_authorization_models()
        await fga_client.close()
        return api_response

API Token

from openfga_sdk import ClientConfiguration, OpenFgaClient
from openfga_sdk.credentials import CredentialConfiguration, Credentials


async def main():
    configuration = ClientConfiguration(
        api_url=FGA_API_URL,  # required
        store_id=FGA_STORE_ID,  # optional, not needed when calling `CreateStore` or `ListStores`
        authorization_model_id=FGA_MODEL_ID,  # Optional, can be overridden per request
        credentials=Credentials(
            method='api_token',
            configuration=CredentialConfiguration(
                api_token=FGA_API_TOKEN,
            )
        )
    )
    # Enter a context with an instance of the OpenFgaClient
    async with OpenFgaClient(configuration) as fga_client:
        api_response = await fga_client.read_authorization_models()
        await fga_client.close()
        return api_response

Client Credentials

from openfga_sdk import ClientConfiguration, OpenFgaClient
from openfga_sdk.credentials import Credentials, CredentialConfiguration


async def main():
    configuration = ClientConfiguration(
        api_url=FGA_API_URL,  # required
        store_id=FGA_STORE_ID,  # optional, not needed when calling `CreateStore` or `ListStores`
        authorization_model_id=FGA_MODEL_ID,  # Optional, can be overridden per request
        credentials=Credentials(
            method='client_credentials',
            configuration=CredentialConfiguration(
                api_issuer=FGA_API_TOKEN_ISSUER,
                api_audience=FGA_API_AUDIENCE,
                client_id=FGA_CLIENT_ID,
                client_secret=FGA_CLIENT_SECRET,
            )
        )
    )
    # Enter a context with an instance of the OpenFgaClient
    async with OpenFgaClient(configuration) as fga_client:
        api_response = await fga_client.read_authorization_models()
        await fga_client.close()
        return api_response

Custom Headers

Default Headers

You can configure default headers that will be sent with every request by passing them to the ClientConfiguration:

from openfga_sdk import ClientConfiguration, OpenFgaClient


async def main():
    configuration = ClientConfiguration(
        api_url=FGA_API_URL,
        store_id=FGA_STORE_ID,
        authorization_model_id=FGA_MODEL_ID,
        headers={
            "X-Custom-Header": "default-value",
            "X-Request-Source": "my-app",
        },
    )

    async with OpenFgaClient(configuration) as fga_client:
        api_response = await fga_client.read_authorization_models()
        return api_response

Per-Request Headers

You can also send custom headers on a per-request basis by using the options parameter. Per-request headers will override any default headers with the same name.

from openfga_sdk import ClientConfiguration, OpenFgaClient
from openfga_sdk.client.models import ClientCheckRequest


async def main():
    configuration = ClientConfiguration(
        api_url=FGA_API_URL,
        store_id=FGA_STORE_ID,
        authorization_model_id=FGA_MODEL_ID,
    )

    async with OpenFgaClient(configuration) as fga_client:
        # Add custom headers to a specific request
        result = await fga_client.check(
            body=ClientCheckRequest(
                user="user:anne",
                relation="viewer",
                object="document:roadmap",
            ),
            options={
                "headers": {
                    "X-Request-ID": "123e4567-e89b-12d3-a456-426614174000",
                    "X-Custom-Header": "custom-value",  # Overrides default header value
                }
            }
        )
        return result

Synchronous Client

To run outside of an async context, the project exports a synchronous client from openfga_sdk.sync that supports all the credential types and calls, without requiring async/await.

from openfga_sdk.client import ClientConfiguration
from openfga_sdk.sync import OpenFgaClient


def main():
    configuration = ClientConfiguration(
        api_url=FGA_API_URL,  # required
        store_id=FGA_STORE_ID,  # optional, not needed when calling `CreateStore` or `ListStores`
        authorization_model_id=FGA_MODEL_ID,  # optional, can be overridden per request
    )
    # Enter a context with an instance of the OpenFgaClient
    with OpenFgaClient(configuration) as fga_client:
        api_response = fga_client.read_authorization_models()
        return api_response

Get your Store ID

You need your store id to call the OpenFGA API (unless it is to call the CreateStore or ListStores methods).

If your server is configured with authentication enabled, you also need to have your credentials ready.

Calling the API

Stores

List Stores

Get a paginated list of stores.

API Documentation

# from openfga_sdk.sync import OpenFgaClient

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

options = {"page_size": 25, "continuation_token": "eyJwayI6IkxBVEVTVF9OU0NPTkZJR19hdXRoMHN0b3JlIiwic2siOiIxem1qbXF3MWZLZExTcUoyN01MdTdqTjh0cWgifQ=="}
response = await fga_client.list_stores(options)
# response = ListStoresResponse(...)
# response.stores = [Store({"id": "01FQH7V8BEG3GPQW93KTRFR8JB", "name": "FGA Demo Store", "created_at": "2022-01-01T00:00:00.000Z", "updated_at": "2022-01-01T00:00:00.000Z"})]

Create Store

Create and initialize a store.

API Documentation

# from openfga_sdk import CreateStoreRequest, OpenFgaClient

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

body = CreateStoreRequest(
    name = "FGA Demo Store",
)
response = await fga_client.create_store(body)
# response.id = "01FQH7V8BEG3GPQW93KTRFR8JB"

Get Store

Get information about the current store.

API Documentation

Requires a client initialized with a storeId

# from openfga_sdk import OpenFgaClient

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

response = await fga_client.get_store()
# response = Store({"id": "01FQH7V8BEG3GPQW93KTRFR8JB", "name": "FGA Demo Store", "created_at": "2022-01-01T00:00:00.000Z", "updated_at": "2022-01-01T00:00:00.000Z"})

Delete Store

Delete a store.

API Documentation

Requires a client initialized with a storeId

# from openfga_sdk import OpenFgaClient

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

response = await fga_client.delete_store()

Authorization Models

Read Authorization Models

Read all authorization models in the store.

API Documentation

# from openfga_sdk import OpenFgaClient

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

options = {"page_size": 25, "continuation_token": "eyJwayI6IkxBVEVTVF9OU0NPTkZJR19hdXRoMHN0b3JlIiwic2siOiIxem1qbXF3MWZLZExTcUoyN01MdTdqTjh0cWgifQ=="}
response = await fga_client.read_authorization_models(options)
# response.authorization_models = [AuthorizationModel(id='01GXSA8YR785C4FYS3C0RTG7B1', schema_version = '1.1', type_definitions=type_definitions[...], AuthorizationModel(id='01GXSBM5PVYHCJNRNKXMB4QZTW', schema_version = '1.1', type_definitions=type_definitions[...])]

Write Authorization Model

Create a new authorization model.

API Documentation

Note: To learn how to build your authorization model, check the Docs at https://openfga.dev/docs.

Learn more about the OpenFGA configuration language.

You can use the OpenFGA Syntax Transformer to convert between the friendly DSL and the JSON authorization model.

# from openfga_sdk import (
#     Condition, ConditionParamTypeRef, Metadata, ObjectRelation, OpenFgaClient, RelationMetadata,
#     RelationReference, TypeDefinition, Userset, Usersets, WriteAuthorizationModelRequest
# )

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

user_type = TypeDefinition(type="user")

document_relations = dict(
    writer=Userset(this=dict()),
    viewer=Userset(
        union=Usersets(
            child=[
                Userset(this=dict()),
                Userset(computed_userset=ObjectRelation(
                    object="",
                    relation="writer",
                )),
            ],
        ),
    ),
)

document_metadata = Metadata(
        relations=dict(
            writer=RelationMetadata(
                directly_related_user_types=[
                    RelationReference(type="user"),
                    RelationReference(type="user", condition="ViewCountLessThan200"),
                ]
            ),
            viewer=RelationMetadata(
                directly_related_user_types=[
                    RelationReference(type="user"),
                ]
            )
        )
)

document_type = TypeDefinition(
    type="document",
    relations=document_relations,
    metadata=document_metadata
)

conditions = dict(
        ViewCountLessThan200=Condition(
            name="ViewCountLessThan200",
            expression="ViewCount < 200",
            parameters=dict(
                ViewCount=ConditionParamTypeRef(
                    type_name="TYPE_NAME_INT"
                ),
                Type=ConditionParamTypeRef(
                    type_name="TYPE_NAME_STRING"
                ),
                Name=ConditionParamTypeRef(
                    type_name="TYPE_NAME_STRING"
                ),
            )
        )
    )

body = WriteAuthorizationModelRequest(
    schema_version="1.1",
    type_definitions=[
        user_type,
        document_type
    ],
    conditions=conditions
)

response = await fga_client.write_authorization_model(body)
# response.authorization_model_id = "01GXSA8YR785C4FYS3C0RTG7B1"

Read a Single Authorization Model

Read a particular authorization model.

API Documentation

# from openfga_sdk import OpenFgaClient

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

options = {
    # You can rely on the model id set in the configuration or override it for this specific request
    "authorization_model_id": "01GXSA8YR785C4FYS3C0RTG7B1"
}

response = await fga_client.read_authorization_model(options)
# response.authorization_model =  AuthorizationModel(id='01GXSA8YR785C4FYS3C0RTG7B1', schema_version = '1.1', type_definitions=type_definitions[...])

Read the Latest Authorization Model

Reads the latest authorization model (note: this ignores the model id in configuration).

API Documentation

# from openfga_sdk import ClientConfiguration, OpenFgaClient

# Create the cofiguration object
# configuration = ClientConfiguration(
#     ...
#     authorization_model_id = FGA_MODEL_ID,
#     ...
# )

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

response = await fga_client.read_latest_authorization_model()
# response.authorization_model =  AuthorizationModel(id='01GXSA8YR785C4FYS3C0RTG7B1', schema_version = '1.1', type_definitions=type_definitions[...])

Relationship Tuples

Read Relationship Tuple Changes (Watch)

Reads the list of historical relationship tuple writes and deletes.

API Documentation

# from openfga_sdk import OpenFgaClient
# from openfga_sdk.client.models import ClientReadChangesRequest

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

options = {
    # You can rely on the model id set in the configuration or override it for this specific request
    "page_size": "25",
    "continuation_token": "eyJwayI6IkxBVEVTVF9OU0NPTkZJR19hdXRoMHN0b3JlIiwic2siOiIxem1qbXF3MWZLZExTcUoyN01MdTdqTjh0cWgifQ=="
}
body = ClientReadChangesRequest(type="document", start_time="2022-01-01T00:00:00Z")

response = await fga_client.read_changes(body, options)
# response.continuation_token = ...
# response.changes = [TupleChange(tuple_key=TupleKey(object="...",relation="...",user="..."),operation=TupleOperation("TUPLE_OPERATION_WRITE"),timestamp=datetime.fromisoformat("..."))]

Read Relationship Tuples

Reads the relationship tuples stored in the database. It does not evaluate nor exclude invalid tuples according to the authorization model.

API Documentation

# from openfga_sdk import OpenFgaClient, ReadRequestTupleKey

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

# Find if a relationship tuple stating that a certain user is a viewer of certain document
body = ReadRequestTupleKey(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    relation="viewer",
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
)

response = await fga_client.read(body)
# response = ReadResponse({"tuples": [Tuple({"key": TupleKey({"user":"...","relation":"...","object":"..."}), "timestamp": datetime.fromisoformat("...") })]})

# from openfga_sdk import OpenFgaClient, ReadRequestTupleKey

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

# Find all relationship tuples where a certain user has a relationship as any relation to a certain document
body = ReadRequestTupleKey(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
)

response = await fga_client.read(body)
# response = ReadResponse({"tuples": [Tuple({"key": TupleKey({"user":"...","relation":"...","object":"..."}), "timestamp": datetime.fromisoformat("...") })]})

# from openfga_sdk import OpenFgaClient, ReadRequestTupleKey

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

# Find all relationship tuples where a certain user is a viewer of any document
body = ReadRequestTupleKey(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    relation="viewer",
    object="document:",
)

response = await fga_client.read(body)
# response = ReadResponse({"tuples": [Tuple({"key": TupleKey({"user":"...","relation":"...","object":"..."}), "timestamp": datetime.fromisoformat("...") })]})

# from openfga_sdk import OpenFgaClient, ReadRequestTupleKey

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

# Find all relationship tuples where any user has a relationship as any relation with a particular document
body = ReadRequestTupleKey(
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
)

response = await fga_client.read(body)
# response = ReadResponse({"tuples": [Tuple({"key": TupleKey({"user":"...","relation":"...","object":"..."}), "timestamp": datetime.fromisoformat("...") })]})

# from openfga_sdk import OpenFgaClient, ReadRequestTupleKey

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

# Read all stored relationship tuples
body = ReadRequestTupleKey()

response = await api_instance.read(body)
# response = ReadResponse({"tuples": [Tuple({"key": TupleKey({"user":"...","relation":"...","object":"..."}), "timestamp": datetime.fromisoformat("...") })]})

Write (Create and Delete) Relationship Tuples

Create and/or delete relationship tuples to update the system state.

API Documentation

Transaction mode (default)

By default, write runs in a transaction mode where any invalid operation (deleting a non-existing tuple, creating an existing tuple, one of the tuples was invalid) or a server error will fail the entire operation.

# from openfga_sdk import OpenFgaClient, RelationshipCondition
# from openfga_sdk.client.models import ClientTuple, ClientWriteRequest

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

options = {
    # You can rely on the model id set in the configuration or override it for this specific request
    "authorization_model_id": "01GXSA8YR785C4FYS3C0RTG7B1"
}
body = ClientWriteRequest(
    writes=[
        ClientTuple(
            user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
            relation="viewer",
            object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
            condition=RelationshipCondition(
              name='ViewCountLessThan200',
              context=dict(
                  Name='Roadmap',
                  Type='Document',
              ),
            ),
        ),
        ClientTuple(
            user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
            relation="viewer",
            object="document:0192ab2d-d36e-7cb3-a4a8-5d1d67a300c5",
        ),
    ],
    deletes=[
        ClientTuple(
            user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
            relation="writer",
            object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
        ),
    ],
)

response = await fga_client.write(body, options)

Convenience write_tuples and delete_tuples methods are also available.

Non-transaction mode

The SDK will split the writes into separate requests and send them sequentially to avoid violating rate limits.

# from openfga_sdk import OpenFgaClient, RelationshipCondition
# from openfga_sdk.client.models import ClientTuple, ClientWriteRequest, WriteTransactionOpts

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

options = {
    # You can rely on the model id set in the configuration or override it for this specific request
    "authorization_model_id": "01GXSA8YR785C4FYS3C0RTG7B1",
    "transaction": WriteTransactionOpts(
        disabled=True,
        max_parallel_requests=10, # Maximum number of requests to issue in parallel
        max_per_chunk=1, # Maximum number of requests to be sent in a transaction in a particular chunk
    )
}
body = ClientWriteRequest(
    writes=[
        ClientTuple(
            user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
            relation="viewer",
            object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
        ),
        ClientTuple(
            user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
            relation="viewer",
            object="document:0192ab2d-d36e-7cb3-a4a8-5d1d67a300c5",
            condition=RelationshipCondition(
              name='ViewCountLessThan200',
              context=dict(
                  Name='Roadmap',
                  Type='Document',
              ),
            ),
        ),
    ],
    deletes=[
        ClientTuple(
            user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
            relation="writer",
            object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
        ),
    ],
)

response = await fga_client.write(body, options)

Conflict Options for Write Operations

OpenFGA v1.10.0+ supports conflict options for write operations to handle duplicate writes and missing deletes gracefully.

Example: Ignore duplicate writes and missing deletes

# from openfga_sdk import OpenFgaClient
# from openfga_sdk.client.models import ClientTuple, ClientWriteRequest
# from openfga_sdk.client.models.write_conflict_opts import (
#     ClientWriteRequestOnDuplicateWrites,
#     ClientWriteRequestOnMissingDeletes,
#     ConflictOptions,
# )

options = {
    "authorization_model_id": "01GXSA8YR785C4FYS3C0RTG7B1",
    "conflict": ConflictOptions(
        on_duplicate_writes=ClientWriteRequestOnDuplicateWrites.IGNORE, # Available options: ERROR, IGNORE
        on_missing_deletes=ClientWriteRequestOnMissingDeletes.IGNORE, # Available options: ERROR, IGNORE
    )
}

body = ClientWriteRequest(
    writes=[
        ClientTuple(
            user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
            relation="viewer",
            object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
        ),
    ],
    deletes=[
        ClientTuple(
            user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
            relation="writer",
            object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
        ),
    ],
)

response = await fga_client.write(body, options)

Relationship Queries

Check

Check if a user has a particular relation with an object.

API Documentation

# from openfga_sdk import OpenFgaClient
# from openfga_sdk.client import ClientCheckRequest

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

options = {
    # You can rely on the model id set in the configuration or override it for this specific request
    "authorization_model_id": "01GXSA8YR785C4FYS3C0RTG7B1"
}
body = ClientCheckRequest(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    relation="writer",
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
    context=dict(
        ViewCount=100
    ),
)

response = await fga_client.check(body, options)
# response.allowed = True

Batch Check

Similar to check, but instead of checking a single user-object relationship, accepts a list of relationships to check. Requires OpenFGA version 1.8.0 or greater.

API Documentation

Note: The order of batch_check results is not guaranteed to match the order of the checks provided. Use correlation_id to pair responses with requests.

# from openfga_sdk import OpenFgaClient
# from openfga_sdk.client.models import (
#   ClientTuple,
#   ClientBatchCheckItem,
#   ClientBatchCheckRequest,
# )
# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

options = {
    # You can rely on the model id set in the configuration or override it for this specific request
    "authorization_model_id": "01GXSA8YR785C4FYS3C0RTG7B1"
}
checks = [ClientBatchCheckItem(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    relation="viewer",
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
    contextual_tuples=[  # optional
        ClientTuple(
            user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
            relation="editor",
            object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
        ),
    ],
    context=dict(
        ViewCount=100
    )
), ClientBatchCheckItem(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    relation="admin",
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
    contextual_tuples=[  # optional
        ClientTuple(
            user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
            relation="editor",
            object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
        ),
    ]
), ClientBatchCheckItem(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    relation="creator",
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
), ClientBatchCheckItem(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    relation="deleter",
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
)]

response = await fga_client.batch_check(ClientBatchCheckRequest(checks=checks), options)
# response.result = [{
#   allowed: false,
#   correlation_id: "de3630c2-f9be-4ee5-9441-cb1fbd82ce75",
#   tuple: {
#     user: "user:81684243-9356-4421-8fbf-a4f8d36aa31b",
#     relation: "viewer",
#     object: "document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
#     contextual_tuples: [{
#       user: "user:81684243-9356-4421-8fbf-a4f8d36aa31b",
#       relation: "editor",
#       object: "document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a"
#     }],
#    context=dict(
#        ViewCount=100
#    )
#   }
# }, {
#   allowed: false,
#   correlation_id: "6d7c7129-9607-480e-bfd0-17c16e46b9ec",
#   tuple: {
#     user: "user:81684243-9356-4421-8fbf-a4f8d36aa31b",
#     relation: "admin",
#     object: "document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
#     contextual_tuples: [{
#       user: "user:81684243-9356-4421-8fbf-a4f8d36aa31b",
#       relation: "editor",
#       object: "document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a"
#     }]
#   }
# }, {
#   allowed: false,
#   correlation_id: "210899b9-6bc3-4491-bdd1-d3d79780aa31",
#   tuple: {
#     user: "user:81684243-9356-4421-8fbf-a4f8d36aa31b",
#     relation: "creator",
#     object: "document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
#   },
#   error: {
#       input_error: "validation_error",
#       message: "relation 'document#creator' not found"
#   }
# }, {
#   allowed: true,
#   correlation_id: "55cc1946-9fc3-4710-bd40-8fe2687ed8da",
#   request: {
#     user: "user:81684243-9356-4421-8fbf-a4f8d36aa31b",
#     relation: "deleter",
#     object: "document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
#   }},
# ]

If you are using an OpenFGA version less than 1.8.0, you can use client_batch_check, which calls check in parallel. It will return allowed: false if it encounters an error, and will return the error in the body. If 429s or 5xxs are encountered, the underlying check will retry up to 3 times before giving up.

# from openfga_sdk import OpenFgaClient
# from openfga_sdk.client import ClientCheckRequest
# from openfga_sdk.client.models import ClientTuple
# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

options = {
    # You can rely on the model id set in the configuration or override it for this specific request
    "authorization_model_id": "01GXSA8YR785C4FYS3C0RTG7B1"
}
body = [ClientCheckRequest(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    relation="viewer",
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
    contextual_tuples=[  # optional
        ClientTuple(
            user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
            relation="editor",
            object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
        ),
    ],
    context=dict(
        ViewCount=100
    )
), ClientCheckRequest(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    relation="admin",
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
    contextual_tuples=[  # optional
        ClientTuple(
            user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
            relation="editor",
            object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
        ),
    ]
), ClientCheckRequest(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    relation="creator",
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
), ClientCheckRequest(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    relation="deleter",
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
)]

response = await fga_client.client_batch_check(body, options)
# response.responses = [{
#   allowed: false,
#   request: {
#     user: "user:81684243-9356-4421-8fbf-a4f8d36aa31b",
#     relation: "viewer",
#     object: "document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
#     contextual_tuples: [{
#       user: "user:81684243-9356-4421-8fbf-a4f8d36aa31b",
#       relation: "editor",
#       object: "document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a"
#     }],
#    context=dict(
#        ViewCount=100
#    )
#   }
# }, {
#   allowed: false,
#   request: {
#     user: "user:81684243-9356-4421-8fbf-a4f8d36aa31b",
#     relation: "admin",
#     object: "document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
#     contextual_tuples: [{
#       user: "user:81684243-9356-4421-8fbf-a4f8d36aa31b",
#       relation: "editor",
#       object: "document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a"
#     }]
#   }
# }, {
#   allowed: false,
#   request: {
#     user: "user:81684243-9356-4421-8fbf-a4f8d36aa31b",
#     relation: "creator",
#     object: "document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
#   },
#   error: <FgaError ...>
# }, {
#   allowed: true,
#   request: {
#     user: "user:81684243-9356-4421-8fbf-a4f8d36aa31b",
#     relation: "deleter",
#     object: "document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
#   }},
# ]

Expand

Expands the relationships in userset tree format.

API Documentation

# from openfga_sdk import OpenFgaClient
# from openfga_sdk.client.models import ClientExpandRequest

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

options = {
    # You can rely on the model id set in the configuration or override it for this specific request
    "authorization_model_id": "01GXSA8YR785C4FYS3C0RTG7B1"
}
body = ClientExpandRequest(
    relation="viewer",
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
)

response = await fga_client.expand(body. options)
# response = ExpandResponse({"tree": UsersetTree({"root": Node({"name": "document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a#viewer", "leaf": Leaf({"users": Users({"users": ["user:81684243-9356-4421-8fbf-a4f8d36aa31b", "user:f52a4f7a-054d-47ff-bb6e-3ac81269988f"]})})})})})

List Objects

List the objects of a particular type a user has access to.

API Documentation

# from openfga_sdk import OpenFgaClient
# from openfga_sdk.client.models import ClientListObjectsRequest, ClientTuple

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

options = {
    # You can rely on the model id set in the configuration or override it for this specific request
    "authorization_model_id": "01GXSA8YR785C4FYS3C0RTG7B1"
}
body = ClientListObjectsRequest(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    relation="viewer",
    type="document",
    contextual_tuples=[  # optional
        ClientTuple(
            user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
            relation="writer",
            object="document:0192ab2d-d36e-7cb3-a4a8-5d1d67a300c5",
        ),
    ],
    context=dict(
        ViewCount=100
    )
)

response = await fga_client.list_objects(body)
# response.objects = ["document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a"]

Streamed List Objects

List the objects of a particular type a user has access to, using the streaming API.

API Documentation

# from openfga_sdk import OpenFgaClient
# from openfga_sdk.client.models import ClientListObjectsRequest

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

results = []

documents = ClientListObjectsRequest(
    type="document",
    relation="writer",
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
)

async for response in fga_client.streamed_list_objects(request):
    results.append(response)

# results = ["document:...", ...]

List Relations

List the relations a user has on an object.

# from openfga_sdk import OpenFgaClient
# from openfga_sdk.client.models import ClientListRelationsRequest

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

options = {
    # You can rely on the model id set in the configuration or override it for this specific request
    "authorization_model_id": "01GXSA8YR785C4FYS3C0RTG7B1"
}
body = ClientListRelationsRequest(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
    relations=["can_view", "can_edit", "can_delete", "can_rename"],
    contextual_tuples=[  # optional
        ClientTuple(
            user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
            relation="writer",
            object="document:0192ab2d-d36e-7cb3-a4a8-5d1d67a300c5",
        ),
    ],
    context=dict(
        ViewCount=100
    )
)

response = await fga_client.list_relations(body, options)

# response.relations = ["can_view", "can_edit"]

List Users

List the users who have a certain relation to a particular type.

API Documentation

from openfga_sdk import OpenFgaClient
from openfga_sdk.models.fga_object import FgaObject
from openfga_sdk.client.models import ClientTuple
from openfga_sdk.client.models.list_users_request import ClientListUsersRequest

configuration = ClientConfiguration(
    api_url=FGA_API_URL,
    # ...
)

async with OpenFgaClient(configuration) as api_client:
    options = {
        "authorization_model_id": "01GXSA8YR785C4FYS3C0RTG7B1"
    }

    request = ClientListUsersRequest(
        object=FgaObject(type="document", id="2021-budget"),
        relation="can_read",
        user_filters=[
            UserTypeFilter(type="user"),
            UserTypeFilter(type="team", relation="member"),
        ],
        context={},
        contextual_tuples=[
            ClientTuple(
                user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
                relation="editor",
                object="folder:product",
            ),
            ClientTuple(
                user="folder:product",
                relation="parent",
                object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
            ),
        ],
    )

    response = await api_client.list_users(request, options)

    # response.users = [{object: {type: "user", id: "81684243-9356-4421-8fbf-a4f8d36aa31b"}}, {userset: { type: "user" }}, ...]

Assertions

Read Assertions

Read assertions for a particular authorization model.

API Documentation

# from openfga_sdk import OpenFgaClient

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

options = {
    # You can rely on the model id set in the configuration or override it for this specific request
    "authorization_model_id": "01GXSA8YR785C4FYS3C0RTG7B1"
}

response = await fga_client.read_assertions(options)

Write Assertions

Update the assertions for a particular authorization model.

API Documentation

# from openfga_sdk import OpenFgaClient
# from openfga_sdk.client.models import ClientAssertion

# Initialize the fga_client
# fga_client = OpenFgaClient(configuration)

options = {
    # You can rely on the model id set in the configuration or override it for this specific request
    "authorization_model_id": "01GXSA8YR785C4FYS3C0RTG7B1"
}

body = [ClientAssertion(
    user="user:81684243-9356-4421-8fbf-a4f8d36aa31b",
    relation="viewer",
    object="document:0192ab2a-d83f-756d-9397-c5ed9f3cb69a",
    expectation=True,
)]

response = await fga_client.write_assertions(body, options)

Retries

If a network request fails with a 429 or 5xx error from the server, the SDK will automatically retry the request up to 3 times with a minimum wait time of 100 milliseconds between each attempt.

To customize this behavior, create a RetryParams object and assign values to the max_retry and min_wait_in_ms constructor parameters. max_retry determines the maximum number of retries (up to 15), while min_wait_in_ms sets the minimum wait time between retries in milliseconds.

Apply your custom retry values by passing the object to the ClientConfiguration constructor's retry_params parameter.

from openfga_sdk import ClientConfiguration, OpenFgaClient
from openfga_sdk.configuration import RetryParams
from os import environ

async def main():
    # Configure the client with custom retry settings
    config = ClientConfiguration(
        api_url=environ.get("FGA_API_URL"),
        retry_params=RetryParams(max_retry=3, min_wait_in_ms=250)
    )

    # Create a client instance and read authorization models
    async with OpenFgaClient(config) as client:
        return await client.read_authorization_models()

API Endpoints

Class	Method	HTTP request	Description
OpenFgaApi	batch_check	POST /stores/{store_id}/batch-check	Send a list of `check` operations in a single request
OpenFgaApi	check	POST /stores/{store_id}/check	Check whether a user is authorized to access an object
OpenFgaApi	create_store	POST /stores	Create a store
OpenFgaApi	delete_store	DELETE /stores/{store_id}	Delete a store
OpenFgaApi	expand	POST /stores/{store_id}/expand	Expand all relationships in userset tree format, and following userset rewrite rules. Useful to reason about and debug a certain relationship
OpenFgaApi	get_store	GET /stores/{store_id}	Get a store
OpenFgaApi	list_objects	POST /stores/{store_id}/list-objects	List all objects of the given type that the user has a relation with
OpenFgaApi	list_stores	GET /stores	List all stores
OpenFgaApi	list_users	POST /stores/{store_id}/list-users	List the users matching the provided filter who have a certain relation to a particular type.
OpenFgaApi	read	POST /stores/{store_id}/read	Get tuples from the store that matches a query, without following userset rewrite rules
OpenFgaApi	read_assertions	GET /stores/{store_id}/assertions/{authorization_model_id}	Read assertions for an authorization model ID
OpenFgaApi	read_authorization_model	GET /stores/{store_id}/authorization-models/{id}	Return a particular version of an authorization model
OpenFgaApi	read_authorization_models	GET /stores/{store_id}/authorization-models	Return all the authorization models for a particular store
OpenFgaApi	read_changes	GET /stores/{store_id}/changes	Return a list of all the tuple changes
OpenFgaApi	streamed_list_objects	POST /stores/{store_id}/streamed-list-objects	Stream all objects of the given type that the user has a relation with
OpenFgaApi	write	POST /stores/{store_id}/write	Add or delete tuples from the store
OpenFgaApi	write_assertions	PUT /stores/{store_id}/assertions/{authorization_model_id}	Upsert assertions for an authorization model ID
OpenFgaApi	write_authorization_model	POST /stores/{store_id}/authorization-models	Create a new authorization model

Models

Documentation For Models

• AbortedMessageResponse
• Any
• Assertion
• AssertionTupleKey
• AuthErrorCode
• AuthorizationModel
• BatchCheckItem
• BatchCheckRequest
• BatchCheckResponse
• BatchCheckSingleResult
• CheckError
• CheckRequest
• CheckRequestTupleKey
• CheckResponse
• Computed
• Condition
• ConditionMetadata
• ConditionParamTypeRef
• ConsistencyPreference
• ContextualTupleKeys
• CreateStoreRequest
• CreateStoreResponse
• Difference
• ErrorCode
• ExpandRequest
• ExpandRequestTupleKey
• ExpandResponse
• FgaObject
• ForbiddenResponse
• GetStoreResponse
• InternalErrorCode
• InternalErrorMessageResponse
• Leaf
• ListObjectsRequest
• ListObjectsResponse
• ListStoresResponse
• ListUsersRequest
• ListUsersResponse
• Metadata
• Node
• Nodes
• NotFoundErrorCode
• NullValue
• ObjectRelation
• PathUnknownErrorMessageResponse
• ReadAssertionsResponse
• ReadAuthorizationModelResponse
• ReadAuthorizationModelsResponse
• ReadChangesResponse
• ReadRequest
• ReadRequestTupleKey
• ReadResponse
• RelationMetadata
• RelationReference
• RelationshipCondition
• SourceInfo
• Status
• Store
• StreamResultOfStreamedListObjectsResponse
• StreamedListObjectsResponse
• Tuple
• TupleChange
• TupleKey
• TupleKeyWithoutCondition
• TupleOperation
• TupleToUserset
• TypeDefinition
• TypeName
• TypedWildcard
• UnauthenticatedResponse
• UnprocessableContentErrorCode
• UnprocessableContentMessageResponse
• User
• UserTypeFilter
• Users
• Userset
• UsersetTree
• UsersetTreeDifference
• UsersetTreeTupleToUserset
• UsersetUser
• Usersets
• ValidationErrorMessageResponse
• WriteAssertionsRequest
• WriteAuthorizationModelRequest
• WriteAuthorizationModelResponse
• WriteRequest
• WriteRequestDeletes
• WriteRequestWrites

OpenTelemetry

This SDK supports producing metrics that can be consumed as part of an OpenTelemetry setup. For more information, please see the documentation

Error Handling

The SDK provides comprehensive error handling with detailed error information and convenient helper methods.

Key features:

• Operation context in error messages (e.g., [write], [check])
• Detailed error codes and messages from the API
• Helper methods for error categorization (is_validation_error(), is_retryable(), etc.)

from openfga_sdk.exceptions import ApiException

try:
    await client.write([tuple])
except ApiException as e:
    print(f"Error: {e}")  # [write] HTTP 400 type 'invalid_type' not found (validation_error) [request-id: abc-123]
    
    if e.is_validation_error():
        print(f"Validation error: {e.error_message}")
    elif e.is_retryable():
        print(f"Temporary error - retrying... (Request ID: {e.request_id})")

Contributing

See CONTRIBUTING for details.

Author

OpenFGA

License

This project is licensed under the Apache-2.0 license. See the LICENSE file for more info.

The code in this repo was auto generated by OpenAPI Generator from a template based on the python legacy template, licensed under the Apache License 2.0.