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-python package via pip:

pip install eunomia-sdk-python

Usage

Import the EunomiaClient class and create an instance of it:

from eunomia_sdk_python import EunomiaClient

client = EunomiaClient(
    endpoint="http://localhost:8000",
    api_key="my-api-key",
)

You can then use the client to interact with the Eunomia server:

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

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

# Check if a principal has permissions to perform an action on a resource
is_allowed = client.check(
    principal_uri=principal.uri,
    resource_uri=resource.uri
)

SDK Docs

eunomia_sdk_python.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_python/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
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",
    ) -> bool:
        """
        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
        -------
        bool
            True if the principal has permissions to perform the action on the resource, False otherwise.

        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 bool(response.json())

    def bulk_check(self, check_requests: list[schemas.CheckRequest]) -> list[bool]:
        """
        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[bool]
            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 [bool(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(
            "/fetchers/internal/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"/fetchers/internal/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"/fetchers/internal/entities/{uri}")
        self._handle_response(response)
        return response.json()

    def create_policy(self, request: schemas.CheckRequest, name: str) -> schemas.Policy:
        """
        Create a new policy 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(
            "/policies", json=request.model_dump(), params={"name": name}
        )
        self._handle_response(response)
        return schemas.Policy.model_validate(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[bool]

The list of results of the check requests.
Source code in pkgs/sdks/python/src/eunomia_sdk_python/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
def bulk_check(self, check_requests: list[schemas.CheckRequest]) -> list[bool]:
    """
    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[bool]
        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 [bool(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
bool

True if the principal has permissions to perform the action on the resource, False otherwise.

Raises:

Type Description
HTTPStatusError

If the HTTP request returns an unsuccessful status code.
Source code in pkgs/sdks/python/src/eunomia_sdk_python/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",
) -> bool:
    """
    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
    -------
    bool
        True if the principal has permissions to perform the action on the resource, False otherwise.

    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 bool(response.json())

create_policy(request, name)

Create a new policy 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_python/client.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def create_policy(self, request: schemas.CheckRequest, name: str) -> schemas.Policy:
    """
    Create a new policy 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(
        "/policies", 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_python/client.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
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"/fetchers/internal/entities/{uri}")
    self._handle_response(response)
    return 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_python/client.py 
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
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(
        "/fetchers/internal/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_python/client.py 
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
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"/fetchers/internal/entities/{uri}",
        json=entity.model_dump(),
        params={"override": override},
    )
    self._handle_response(response)
    return schemas.EntityInDb.model_validate(response.json())