This is an autogenerated python SDK for OpenFGA. It provides a wrapper around the OpenFGA API definition.
- About OpenFGA
- Resources
- Installation
- Getting Started
- Contributing
- License
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.
- OpenFGA Documentation
- OpenFGA API Documentation
- OpenFGA Community
- Zanzibar Academy
- Google's Zanzibar Paper (2019)
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
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
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
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.
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
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
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
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
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.
Get a paginated list of stores.
# 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 and initialize a store.
# 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 information about the current store.
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 a store.
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()
Read all authorization models in the store.
# 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[...])]
Create a new authorization model.
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 particular authorization model.
# 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[...])
Reads the latest authorization model (note: this ignores the model id in configuration).
# 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[...])
Reads the list of historical relationship tuple writes and deletes.
# 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("..."))]
Reads the relationship tuples stored in the database. It does not evaluate nor exclude invalid tuples according to the authorization model.
# 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("...") })]})
Create and/or delete relationship tuples to update the system state.
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.
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)
Check if a user has a particular relation with an object.
# 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
Run a set of checks. Batch Check 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.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",
# }},
# ]
Expands the relationships in userset tree format.
# 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 the objects of a particular type a user has access to.
# 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"]
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 the users who have a certain relation to a particular type.
from openfga_sdk import OpenFgaClient
from openfga_sdk.models.fga_object import FgaObject
from openfga_sdk.client.models import ClientListUsersRequest, ClientTuple
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" }}, ...]
Read assertions for a particular authorization model.
# 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)
Update the assertions for a particular authorization model.
# 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)
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()
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 | 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 |
- 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
- 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
This SDK supports producing metrics that can be consumed as part of an OpenTelemetry setup. For more information, please see the documentation
If you have found a bug or if you have a feature request, please report them on the sdk-generator repo issues section. Please do not report security vulnerabilities on the public GitHub issue tracker.
While we accept Pull Requests on this repository, the SDKs are autogenerated so please consider additionally submitting your Pull Requests to the sdk-generator and linking the two PRs together and to the corresponding issue. This will greatly assist the OpenFGA team in being able to give timely reviews as well as deploying fixes and updates to our other SDKs as well.
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.