Skip to content

Python SDK

Eunomia offers a Python client that enables users to interact with the Eunomia server.

The client allows you to register resources and principals with their metadata to the Eunomia server, as well as verify permissions of principals performing actions on resources. These features simplify the integration of the Eunomia server into your Python applications.

Installation

Install the eunomia-sdk package via pip:

pip install eunomia-sdk

Usage

Standard API

Use the standard API for authorization checks in your application:

from eunomia_sdk import EunomiaClient

client = EunomiaClient(endpoint="http://localhost:8000")

# Check if a principal has permissions to perform an action on a resource
response = client.check(
    principal_uri="user:123",
    resource_uri="document:456",
    action="read",
)

print(f"Is allowed: {response.allowed}")

Admin API Usage

Use the admin API for server configuration and entity management:

from eunomia_sdk import EunomiaClient

# For admin API usage authentication via API key might be required
client = EunomiaClient(
    endpoint="http://localhost:8000",
    api_key="your-admin-api-key"  # or set WAY_API_KEY environment variable
)

# Register a resource with metadata
resource = client.register_entity(
    type="resource",
    uri="document:456",
    attributes={
        "name": "sensitive_document",
        "type": "document",
        "classification": "confidential",
    },
)

# Register a principal with metadata
principal = client.register_entity(
    type="principal",
    uri="user:123",
    attributes={
        "name": "user_123",
        "role": "analyst",
        "department": "research",
    },
)

Docs

eunomia_sdk.client.EunomiaClient

A client for interacting with the Eunomia server.

This client provides methods to register resources and principals, and to check permissions.

Parameters:

Name Type Description Default
endpoint str

The base URL endpoint of the Eunomia server. Defaults to "http://localhost:8000" if not provided.

 None
api_key str

The API key for authenticating with the server. Defaults to the environment variable "WAY_API_KEY" if not provided.

 None
Source code in pkgs/sdks/python/src/eunomia_sdk/client.py 
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
class EunomiaClient:
    """
    A client for interacting with the Eunomia server.

    This client provides methods to register resources and principals,
    and to check permissions.

    Parameters
    ----------
    endpoint : str, optional
        The base URL endpoint of the Eunomia server.
        Defaults to "http://localhost:8000" if not provided.
    api_key : str, optional
        The API key for authenticating with the server.
        Defaults to the environment variable "WAY_API_KEY" if not provided.
    """

    def __init__(self, endpoint: str | None = None, api_key: str | None = None) -> None:
        self._endpoint = endpoint if endpoint is not None else "http://localhost:8000"
        self._api_key = (
            api_key if api_key is not None else os.getenv("WAY_API_KEY", None)
        )

        headers = {}
        if self._api_key is not None:
            headers["WAY-API-KEY"] = self._api_key

        self.client = httpx.Client(base_url=self._endpoint, headers=headers, timeout=60)

    def _handle_response(self, response: httpx.Response) -> None:
        try:
            response.raise_for_status()
            return
        except httpx.HTTPStatusError as e:
            raise httpx.HTTPStatusError(
                f"{e}\nResponse: {e.response.text}",
                request=e.request,
                response=e.response,
            ) from None

    def check(
        self,
        principal_uri: str | None = None,
        resource_uri: str | None = None,
        principal_attributes: dict = {},
        resource_attributes: dict = {},
        action: str = "access",
    ) -> schemas.CheckResponse:
        """
        Check whether a principal has permissions to perform an action on a specific resource.

        Parameters
        ----------
        principal_uri : str, optional
            The identifier of the principal. Can be provided for registered principals to automatically retrieve attributes.
        resource_uri : str, optional
            The identifier of the resource. Can be provided for registered resources to automatically retrieve attributes.
        principal_attributes : dict, optional
            The attributes of the principal. Shall be provided if the principal is not registered.
        resource_attributes : dict, optional
            The attributes of the resource. Shall be provided if the resource is not registered.
        action : str, optional
            The action to check permissions for. Defaults to "access".

        Returns
        -------
        schemas.CheckResponse
            The response containing the allowed flag and the reason for the decision.

        Raises
        ------
        httpx.HTTPStatusError
            If the HTTP request returns an unsuccessful status code.
        """
        request = schemas.CheckRequest(
            principal=schemas.PrincipalCheck(
                uri=principal_uri, attributes=principal_attributes
            ),
            resource=schemas.ResourceCheck(
                uri=resource_uri, attributes=resource_attributes
            ),
            action=action,
        )
        response = self.client.post("/check", json=request.model_dump())
        self._handle_response(response)
        return schemas.CheckResponse.model_validate(response.json())

    def bulk_check(
        self, check_requests: list[schemas.CheckRequest]
    ) -> list[schemas.CheckResponse]:
        """
        Perform a set of permission checks in a single request.

        Parameters
        ----------
        check_requests : list[schemas.CheckRequest]
            The list of check requests to perform.

        Returns
        -------
        list[schemas.CheckResponse]
            The list of results of the check requests.
        """
        response = self.client.post(
            "/check/bulk",
            json=[
                schemas.CheckRequest.model_validate(request).model_dump()
                for request in check_requests
            ],
        )
        self._handle_response(response)
        return [
            schemas.CheckResponse.model_validate(result) for result in response.json()
        ]

    def register_entity(
        self, type: enums.EntityType, attributes: dict, uri: str | None = None
    ) -> schemas.EntityInDb:
        """
        Register a new entity with the Eunomia server.

        This method registers a new entity with its attributes to the Eunomia server.
        If no uri identifier is provided, the server will generate a random UUID.

        Parameters
        ----------
        type : enums.EntityType
            The type of entity to register. Either "resource" or "principal".
        attributes : dict
            The attributes to associate with the entity.
        uri : str | None, optional
            The uri for the entity. If not provided, the server will generate a random UUID.

        Returns
        -------
        schemas.EntityInDb
            The newly registered entity.

        Raises
        ------
        httpx.HTTPStatusError
            If the HTTP request returns an unsuccessful status code.
        """
        entity = schemas.EntityCreate(type=type, attributes=attributes, uri=uri)
        response = self.client.post(
            "/admin/fetchers/registry/entities", json=entity.model_dump()
        )
        self._handle_response(response)
        return schemas.EntityInDb.model_validate(response.json())

    def update_entity(
        self, uri: str, attributes: dict, override: bool = False
    ) -> schemas.EntityInDb:
        """
        Update the attributes of an existing entity.

        Parameters
        ----------
        uri : str
            The uri of the entity to update.
        attributes : dict
            The attributes to update.
        override : bool, default=False
            If True, the existing attributes are deleted and the new attributes are added.
            If False, the existing attributes are maintaned or updated in case of overlap,
            and the additional new attributes are added.

        Returns
        -------
        schemas.EntityInDb
            The updated entity.

        Raises
        ------
        httpx.HTTPStatusError
            If the HTTP request returns an unsuccessful status code.
        """
        entity = schemas.EntityUpdate(uri=uri, attributes=attributes)
        response = self.client.put(
            f"/admin/fetchers/registry/entities/{uri}",
            json=entity.model_dump(),
            params={"override": override},
        )
        self._handle_response(response)
        return schemas.EntityInDb.model_validate(response.json())

    def delete_entity(self, uri: str) -> bool:
        """
        Delete an entity from the Eunomia server.

        Parameters
        ----------
        uri : str
            The uri of the entity to delete.

        Raises
        ------
        httpx.HTTPStatusError
            If the HTTP request returns an unsuccessful status code.
        """
        response = self.client.delete(f"/admin/fetchers/registry/entities/{uri}")
        self._handle_response(response)
        return response.json()

    def create_policy(self, request: schemas.Policy) -> schemas.Policy:
        """
        Create a new policy and store it in the Eunomia server.

        Parameters
        ----------
        request : schemas.Policy
            The policy to create.

        Raises
        ------
        httpx.HTTPStatusError
            If the HTTP request returns an unsuccessful status code.
        """
        response = self.client.post("/admin/policies", json=request.model_dump())
        self._handle_response(response)
        return schemas.Policy.model_validate(response.json())

    def create_simple_policy(
        self, request: schemas.CheckRequest, name: str
    ) -> schemas.Policy:
        """
        Create a new simple policy with a single rule and store it in the Eunomia server.

        Parameters
        ----------
        request : schemas.CheckRequest
            The request to create the policy from.
        name : str
            The name of the policy.

        Raises
        ------
        httpx.HTTPStatusError
            If the HTTP request returns an unsuccessful status code.
        """
        response = self.client.post(
            "/admin/policies/simple", json=request.model_dump(), params={"name": name}
        )
        self._handle_response(response)
        return schemas.Policy.model_validate(response.json())

    def get_policies(self) -> list[schemas.Policy]:
        """
        Get all policies from the Eunomia server.

        Returns
        -------
        list[schemas.Policy]
            The list of all policies.

        Raises
        ------
        httpx.HTTPStatusError
            If the HTTP request returns an unsuccessful status code.
        """
        response = self.client.get("/admin/policies")
        self._handle_response(response)
        return [schemas.Policy.model_validate(policy) for policy in response.json()]

    def delete_policy(self, name: str) -> bool:
        """
        Delete a policy from the Eunomia server.

        Parameters
        ----------
        name : str
            The name of the policy to delete.

        Returns
        -------
        bool
            True if the policy was successfully deleted.

        Raises
        ------
        httpx.HTTPStatusError
            If the HTTP request returns an unsuccessful status code.
        """
        response = self.client.delete(f"/admin/policies/{name}")
        self._handle_response(response)
        return response.json()

bulk_check(check_requests)

Perform a set of permission checks in a single request.

Parameters:

Name Type Description Default
check_requests list[CheckRequest]

The list of check requests to perform.

 required

Returns:

Type Description
list[CheckResponse]

The list of results of the check requests.
Source code in pkgs/sdks/python/src/eunomia_sdk/client.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def bulk_check(
    self, check_requests: list[schemas.CheckRequest]
) -> list[schemas.CheckResponse]:
    """
    Perform a set of permission checks in a single request.

    Parameters
    ----------
    check_requests : list[schemas.CheckRequest]
        The list of check requests to perform.

    Returns
    -------
    list[schemas.CheckResponse]
        The list of results of the check requests.
    """
    response = self.client.post(
        "/check/bulk",
        json=[
            schemas.CheckRequest.model_validate(request).model_dump()
            for request in check_requests
        ],
    )
    self._handle_response(response)
    return [
        schemas.CheckResponse.model_validate(result) for result in response.json()
    ]

check(principal_uri=None, resource_uri=None, principal_attributes={}, resource_attributes={}, action='access')

Check whether a principal has permissions to perform an action on a specific resource.

Parameters:

Name Type Description Default
principal_uri str

The identifier of the principal. Can be provided for registered principals to automatically retrieve attributes.

 None
resource_uri str

The identifier of the resource. Can be provided for registered resources to automatically retrieve attributes.

 None
principal_attributes dict

The attributes of the principal. Shall be provided if the principal is not registered.

 {}
resource_attributes dict

The attributes of the resource. Shall be provided if the resource is not registered.

 {}
action str

The action to check permissions for. Defaults to "access".

 'access'

Returns:

Type Description
CheckResponse

The response containing the allowed flag and the reason for the decision.

Raises:

Type Description
HTTPStatusError

If the HTTP request returns an unsuccessful status code.
Source code in pkgs/sdks/python/src/eunomia_sdk/client.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def check(
    self,
    principal_uri: str | None = None,
    resource_uri: str | None = None,
    principal_attributes: dict = {},
    resource_attributes: dict = {},
    action: str = "access",
) -> schemas.CheckResponse:
    """
    Check whether a principal has permissions to perform an action on a specific resource.

    Parameters
    ----------
    principal_uri : str, optional
        The identifier of the principal. Can be provided for registered principals to automatically retrieve attributes.
    resource_uri : str, optional
        The identifier of the resource. Can be provided for registered resources to automatically retrieve attributes.
    principal_attributes : dict, optional
        The attributes of the principal. Shall be provided if the principal is not registered.
    resource_attributes : dict, optional
        The attributes of the resource. Shall be provided if the resource is not registered.
    action : str, optional
        The action to check permissions for. Defaults to "access".

    Returns
    -------
    schemas.CheckResponse
        The response containing the allowed flag and the reason for the decision.

    Raises
    ------
    httpx.HTTPStatusError
        If the HTTP request returns an unsuccessful status code.
    """
    request = schemas.CheckRequest(
        principal=schemas.PrincipalCheck(
            uri=principal_uri, attributes=principal_attributes
        ),
        resource=schemas.ResourceCheck(
            uri=resource_uri, attributes=resource_attributes
        ),
        action=action,
    )
    response = self.client.post("/check", json=request.model_dump())
    self._handle_response(response)
    return schemas.CheckResponse.model_validate(response.json())

create_policy(request)

Create a new policy and store it in the Eunomia server.

Parameters:

Name Type Description Default
request Policy

The policy to create.

 required

Raises:

Type Description
HTTPStatusError

If the HTTP request returns an unsuccessful status code.
Source code in pkgs/sdks/python/src/eunomia_sdk/client.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def create_policy(self, request: schemas.Policy) -> schemas.Policy:
    """
    Create a new policy and store it in the Eunomia server.

    Parameters
    ----------
    request : schemas.Policy
        The policy to create.

    Raises
    ------
    httpx.HTTPStatusError
        If the HTTP request returns an unsuccessful status code.
    """
    response = self.client.post("/admin/policies", json=request.model_dump())
    self._handle_response(response)
    return schemas.Policy.model_validate(response.json())

create_simple_policy(request, name)

Create a new simple policy with a single rule and store it in the Eunomia server.

Parameters:

Name Type Description Default
request CheckRequest

The request to create the policy from.

 required
name str

The name of the policy.

 required

Raises:

Type Description
HTTPStatusError

If the HTTP request returns an unsuccessful status code.
Source code in pkgs/sdks/python/src/eunomia_sdk/client.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def create_simple_policy(
    self, request: schemas.CheckRequest, name: str
) -> schemas.Policy:
    """
    Create a new simple policy with a single rule and store it in the Eunomia server.

    Parameters
    ----------
    request : schemas.CheckRequest
        The request to create the policy from.
    name : str
        The name of the policy.

    Raises
    ------
    httpx.HTTPStatusError
        If the HTTP request returns an unsuccessful status code.
    """
    response = self.client.post(
        "/admin/policies/simple", json=request.model_dump(), params={"name": name}
    )
    self._handle_response(response)
    return schemas.Policy.model_validate(response.json())

delete_entity(uri)

Delete an entity from the Eunomia server.

Parameters:

Name Type Description Default
uri str

The uri of the entity to delete.

 required

Raises:

Type Description
HTTPStatusError

If the HTTP request returns an unsuccessful status code.
Source code in pkgs/sdks/python/src/eunomia_sdk/client.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def delete_entity(self, uri: str) -> bool:
    """
    Delete an entity from the Eunomia server.

    Parameters
    ----------
    uri : str
        The uri of the entity to delete.

    Raises
    ------
    httpx.HTTPStatusError
        If the HTTP request returns an unsuccessful status code.
    """
    response = self.client.delete(f"/admin/fetchers/registry/entities/{uri}")
    self._handle_response(response)
    return response.json()

delete_policy(name)

Delete a policy from the Eunomia server.

Parameters:

Name Type Description Default
name str

The name of the policy to delete.

 required

Returns:

Type Description
bool

True if the policy was successfully deleted.

Raises:

Type Description
HTTPStatusError

If the HTTP request returns an unsuccessful status code.
Source code in pkgs/sdks/python/src/eunomia_sdk/client.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
def delete_policy(self, name: str) -> bool:
    """
    Delete a policy from the Eunomia server.

    Parameters
    ----------
    name : str
        The name of the policy to delete.

    Returns
    -------
    bool
        True if the policy was successfully deleted.

    Raises
    ------
    httpx.HTTPStatusError
        If the HTTP request returns an unsuccessful status code.
    """
    response = self.client.delete(f"/admin/policies/{name}")
    self._handle_response(response)
    return response.json()

get_policies()

Get all policies from the Eunomia server.

Returns:

Type Description
list[Policy]

The list of all policies.

Raises:

Type Description
HTTPStatusError

If the HTTP request returns an unsuccessful status code.
Source code in pkgs/sdks/python/src/eunomia_sdk/client.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def get_policies(self) -> list[schemas.Policy]:
    """
    Get all policies from the Eunomia server.

    Returns
    -------
    list[schemas.Policy]
        The list of all policies.

    Raises
    ------
    httpx.HTTPStatusError
        If the HTTP request returns an unsuccessful status code.
    """
    response = self.client.get("/admin/policies")
    self._handle_response(response)
    return [schemas.Policy.model_validate(policy) for policy in response.json()]

register_entity(type, attributes, uri=None)

Register a new entity with the Eunomia server.

This method registers a new entity with its attributes to the Eunomia server. If no uri identifier is provided, the server will generate a random UUID.

Parameters:

Name Type Description Default
type EntityType

The type of entity to register. Either "resource" or "principal".

 required
attributes dict

The attributes to associate with the entity.

 required
uri str | None

The uri for the entity. If not provided, the server will generate a random UUID.

 None

Returns:

Type Description
EntityInDb

The newly registered entity.

Raises:

Type Description
HTTPStatusError

If the HTTP request returns an unsuccessful status code.
Source code in pkgs/sdks/python/src/eunomia_sdk/client.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def register_entity(
    self, type: enums.EntityType, attributes: dict, uri: str | None = None
) -> schemas.EntityInDb:
    """
    Register a new entity with the Eunomia server.

    This method registers a new entity with its attributes to the Eunomia server.
    If no uri identifier is provided, the server will generate a random UUID.

    Parameters
    ----------
    type : enums.EntityType
        The type of entity to register. Either "resource" or "principal".
    attributes : dict
        The attributes to associate with the entity.
    uri : str | None, optional
        The uri for the entity. If not provided, the server will generate a random UUID.

    Returns
    -------
    schemas.EntityInDb
        The newly registered entity.

    Raises
    ------
    httpx.HTTPStatusError
        If the HTTP request returns an unsuccessful status code.
    """
    entity = schemas.EntityCreate(type=type, attributes=attributes, uri=uri)
    response = self.client.post(
        "/admin/fetchers/registry/entities", json=entity.model_dump()
    )
    self._handle_response(response)
    return schemas.EntityInDb.model_validate(response.json())

update_entity(uri, attributes, override=False)

Update the attributes of an existing entity.

Parameters:

Name Type Description Default
uri str

The uri of the entity to update.

 required
attributes dict

The attributes to update.

 required
override bool

If True, the existing attributes are deleted and the new attributes are added. If False, the existing attributes are maintaned or updated in case of overlap, and the additional new attributes are added.

 False

Returns:

Type Description
EntityInDb

The updated entity.

Raises:

Type Description
HTTPStatusError

If the HTTP request returns an unsuccessful status code.
Source code in pkgs/sdks/python/src/eunomia_sdk/client.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def update_entity(
    self, uri: str, attributes: dict, override: bool = False
) -> schemas.EntityInDb:
    """
    Update the attributes of an existing entity.

    Parameters
    ----------
    uri : str
        The uri of the entity to update.
    attributes : dict
        The attributes to update.
    override : bool, default=False
        If True, the existing attributes are deleted and the new attributes are added.
        If False, the existing attributes are maintaned or updated in case of overlap,
        and the additional new attributes are added.

    Returns
    -------
    schemas.EntityInDb
        The updated entity.

    Raises
    ------
    httpx.HTTPStatusError
        If the HTTP request returns an unsuccessful status code.
    """
    entity = schemas.EntityUpdate(uri=uri, attributes=attributes)
    response = self.client.put(
        f"/admin/fetchers/registry/entities/{uri}",
        json=entity.model_dump(),
        params={"override": override},
    )
    self._handle_response(response)
    return schemas.EntityInDb.model_validate(response.json())