11
11
# License for the specific language governing permissions and limitations
12
12
# under the License.
13
13
14
- from typing import Optional , List
14
+ from typing import Optional
15
15
16
16
import requests
17
- from openfga_sdk import ClientConfiguration
18
- from openfga_sdk .credentials import Credentials , CredentialConfiguration
19
- from openfga_sdk .sync import OpenFgaClient
17
+
20
18
from testcontainers .core .container import DockerContainer
21
19
from testcontainers .core .waiting_utils import wait_container_is_ready
22
20
21
+ no_client = False
22
+ try :
23
+ from openfga_sdk import ClientConfiguration
24
+ from openfga_sdk .credentials import CredentialConfiguration , Credentials
25
+ from openfga_sdk .sync import OpenFgaClient
26
+ except ImportError :
27
+ no_client = True
28
+
29
+ class OpenFgaClient :
30
+ pass
31
+
32
+
23
33
_DEFAULT_RUN_COMMAND = "run"
24
34
25
35
@@ -41,13 +51,13 @@ class OpenFGAContainer(DockerContainer):
41
51
42
52
# pylint: disable=too-many-arguments
43
53
def __init__ (
44
- self ,
45
- image : str = "openfga/openfga:latest" ,
46
- preshared_keys : Optional [List [str ]] = None ,
47
- playground_port : int = 3000 ,
48
- http_port : int = 8080 ,
49
- grpc_port : int = 8081 ,
50
- cmd : str = _DEFAULT_RUN_COMMAND ,
54
+ self ,
55
+ image : str = "openfga/openfga:latest" ,
56
+ preshared_keys : Optional [list [str ]] = None ,
57
+ playground_port : int = 3000 ,
58
+ http_port : int = 8080 ,
59
+ grpc_port : int = 8081 ,
60
+ cmd : str = _DEFAULT_RUN_COMMAND ,
51
61
) -> None :
52
62
super ().__init__ (image = image )
53
63
self .preshared_keys = preshared_keys
@@ -60,7 +70,7 @@ def __init__(
60
70
def _configure (self ) -> None :
61
71
if self .preshared_keys :
62
72
self .cmd += " --authn-method=preshared"
63
- self .cmd += f" --authn-preshared-keys=\" { ',' .join (self .preshared_keys )} \" "
73
+ self .cmd += f' --authn-preshared-keys=" { "," .join (self .preshared_keys )} "'
64
74
self .with_command (self .cmd )
65
75
66
76
def get_api_url (self ) -> str :
@@ -77,14 +87,17 @@ def start(self) -> "OpenFGAContainer":
77
87
self ._readiness_probe ()
78
88
return self
79
89
80
- def get_preshared_keys (self ) -> Optional [List [str ]]:
90
+ def get_preshared_keys (self ) -> Optional [list [str ]]:
81
91
return self .preshared_keys
82
92
83
- def get_client (self ) -> OpenFgaClient :
93
+ def get_client (self ) -> "OpenFgaClient" :
94
+ if no_client :
95
+ raise NotImplementedError ("failed to import openfga_sdk: is python < 3.10?" )
96
+
84
97
credentials = None
85
98
if preshared_keys := self .get_preshared_keys ():
86
99
credentials = Credentials (
87
- method = ' api_token' ,
100
+ method = " api_token" ,
88
101
configuration = CredentialConfiguration (
89
102
api_token = preshared_keys [0 ],
90
103
),
0 commit comments