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 access control between principals and 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()

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 access to a resource
has_access = client.check_access(
    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, check access permissions, and retrieve allowed resources for a principal.

Parameters:

Name Type Description Default
server_host str

The base URL 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
class EunomiaClient:
    """
    A client for interacting with the Eunomia server.

    This client provides methods to register resources and principals,
    check access permissions, and retrieve allowed resources for a principal.

    Parameters
    ----------
    server_host : str, optional
        The base URL 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, server_host: str | None = None, api_key: str | None = None
    ) -> None:
        self._server_host = (
            server_host if server_host 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._server_host,
            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_access(
        self,
        principal_uri: str | None = None,
        resource_uri: str | None = None,
        principal_attributes: dict = {},
        resource_attributes: dict = {},
    ) -> bool:
        """
        Check whether a principal has access to 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.

        Returns
        -------
        bool
            True if the principal has access to the resource, False otherwise.

        Raises
        ------
        httpx.HTTPStatusError
            If the HTTP request returns an unsuccessful status code.
        """
        request = schemas.AccessRequest(
            principal=schemas.PrincipalAccess(
                uri=principal_uri, attributes=principal_attributes
            ),
            resource=schemas.ResourceAccess(
                uri=resource_uri, attributes=resource_attributes
            ),
        )
        response = self.client.post("/check-access", json=request.model_dump())
        self._handle_response(response)
        return bool(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", "principal" or "any".
        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("/register-entity", 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.post(
            "/update-entity", json=entity.model_dump(), params={"override": override}
        )
        self._handle_response(response)
        return schemas.EntityInDb.model_validate(response.json())

    def delete_entity(self, uri: str) -> None:
        """
        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.post("/delete-entity", params={"uri": uri})
        self._handle_response(response)
        return

    def create_policy(
        self, policy: schemas.Policy, filename: str | None = None
    ) -> None:
        """
        Create a new policy and save it to the local file system.

        Parameters
        ----------
        policy : schemas.Policy
            The policy to create.
        filename : str, optional
            The filename of the policy to create.

        Raises
        ------
        httpx.HTTPStatusError
            If the HTTP request returns an unsuccessful status code.
        """
        params = {} if filename is None else {"filename": filename}
        response = self.client.post(
            "/create-policy", json=policy.model_dump(), params=params
        )
        self._handle_response(response)
        return

check_access(principal_uri=None, resource_uri=None, principal_attributes={}, resource_attributes={})

Check whether a principal has access to 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.

 {}

Returns:

Type Description
bool

True if the principal has access to 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 
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
def check_access(
    self,
    principal_uri: str | None = None,
    resource_uri: str | None = None,
    principal_attributes: dict = {},
    resource_attributes: dict = {},
) -> bool:
    """
    Check whether a principal has access to 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.

    Returns
    -------
    bool
        True if the principal has access to the resource, False otherwise.

    Raises
    ------
    httpx.HTTPStatusError
        If the HTTP request returns an unsuccessful status code.
    """
    request = schemas.AccessRequest(
        principal=schemas.PrincipalAccess(
            uri=principal_uri, attributes=principal_attributes
        ),
        resource=schemas.ResourceAccess(
            uri=resource_uri, attributes=resource_attributes
        ),
    )
    response = self.client.post("/check-access", json=request.model_dump())
    self._handle_response(response)
    return bool(response.json())

create_policy(policy, filename=None)

Create a new policy and save it to the local file system.

Parameters:

Name Type Description Default
policy Policy

The policy to create.

 required
filename str

The filename of the policy to create.

 None

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 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def create_policy(
    self, policy: schemas.Policy, filename: str | None = None
) -> None:
    """
    Create a new policy and save it to the local file system.

    Parameters
    ----------
    policy : schemas.Policy
        The policy to create.
    filename : str, optional
        The filename of the policy to create.

    Raises
    ------
    httpx.HTTPStatusError
        If the HTTP request returns an unsuccessful status code.
    """
    params = {} if filename is None else {"filename": filename}
    response = self.client.post(
        "/create-policy", json=policy.model_dump(), params=params
    )
    self._handle_response(response)
    return

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 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def delete_entity(self, uri: str) -> None:
    """
    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.post("/delete-entity", params={"uri": uri})
    self._handle_response(response)
    return

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", "principal" or "any".

 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 
 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
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", "principal" or "any".
    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("/register-entity", 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 
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
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.post(
        "/update-entity", json=entity.model_dump(), params={"override": override}
    )
    self._handle_response(response)
    return schemas.EntityInDb.model_validate(response.json())