1
+ import asyncio
2
+ import json
3
+ import os
4
+ import openfga_sdk
5
+ from openfga_sdk .client import OpenFgaClient
6
+ from openfga_sdk .credentials import Credentials , CredentialConfiguration
7
+
8
+ from app .core .config import settings
9
+
10
+
11
+ def build_openfga_client () -> OpenFgaClient :
12
+ """Build and return an OpenFGA client using settings configuration."""
13
+ openfga_client_config = openfga_sdk .ClientConfiguration (
14
+ api_url = settings .FGA_API_URL ,
15
+ store_id = settings .FGA_STORE_ID ,
16
+ credentials = Credentials (
17
+ method = "client_credentials" ,
18
+ configuration = CredentialConfiguration (
19
+ api_issuer = settings .FGA_API_TOKEN_ISSUER ,
20
+ api_audience = settings .FGA_API_AUDIENCE ,
21
+ client_id = settings .FGA_CLIENT_ID ,
22
+ client_secret = settings .FGA_CLIENT_SECRET ,
23
+ ),
24
+ ),
25
+ )
26
+ return OpenFgaClient (openfga_client_config )
27
+
28
+
29
+ async def main ():
30
+ """
31
+ Initializes the OpenFgaClient, writes an authorization model, and configures pre-defined tuples.
32
+
33
+ This function performs the following steps:
34
+ 1. Creates an instance of OpenFgaClient with the necessary configuration.
35
+ 2. Writes an authorization model with specified schema version and type definitions.
36
+ """
37
+
38
+ fga_client = build_openfga_client ()
39
+
40
+ # Define the authorization model
41
+ body_string = "{\" schema_version\" :\" 1.1\" ,\" type_definitions\" :[{\" type\" :\" user\" },{\" metadata\" :{\" relations\" :{\" can_view\" :{},\" owner\" :{\" directly_related_user_types\" :[{\" type\" :\" user\" }]},\" viewer\" :{\" directly_related_user_types\" :[{\" type\" :\" user\" },{\" type\" :\" user\" ,\" wildcard\" :{}}]}}},\" relations\" :{\" can_view\" :{\" union\" :{\" child\" :[{\" computedUserset\" :{\" relation\" :\" owner\" }},{\" computedUserset\" :{\" relation\" :\" viewer\" }}]}},\" owner\" :{\" this\" :{}},\" viewer\" :{\" this\" :{}}},\" type\" :\" doc\" }]}"
42
+ # Write the authorization model
43
+ model = await fga_client .write_authorization_model (json .loads (body_string ))
44
+
45
+ print (f'NEW MODEL ID: { model .authorization_model_id } ' )
46
+
47
+ # Properly close the client session
48
+ await fga_client .close ()
49
+
50
+
51
+ if __name__ == '__main__' :
52
+ asyncio .run (main ())
0 commit comments