Skip to content

AMSDAL Integrations

The amsdal_integrations package allows you to build integrations with any third-party Python application, to collect and push data to AMSDAL.

There are two implemented clients:

You can find the the repository on Github here: amsdal_integrations. The tools directory contains prebuilt integrations for:

  • SQLAlchemy (Coming soon!)
  • Django ORM (Coming soon!)

AMSDAL SDK

amsdal_integrations.core.AmsdalIntegration

Source code in amsdal_integrations/core.py 
 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
class AmsdalIntegration:
    def __init__(
        self,
        config: IntegrationConfig,
        client_class: type[BaseClient] | None = None,
    ) -> None:
        """
        Initialize Amsdal Integration.

        :param config: Configuration for Amsdal Integration.
        :type config: IntegrationConfig
        :param client_class: The client class to use for communicating. Defaults to AmsdalClient.
        :type client_class: type[BaseClient]
        """
        client_class = client_class or AmsdalClient
        self.client = client_class(
            host=config.amsdal_host,
            auth=config.amsdal_auth,
            **config.client_extra,
        )

    def register_schema(
        self,
        schema: Schema,
        operation_id: str | None = None,
        *,
        skip_data_migrations: bool = False,
    ) -> None:
        """
        Register a new schema/model version in AMSDAL application.

        :param schema: The schema/model to register.
        :type schema: Schema
        :param operation_id: The operation id to use for this request. This is useful to prevent duplicated requests.
                             Defaults to None.
        :type operation_id: str, optional
        :param skip_data_migrations: Whether to skip data migrations if a new version of schema is registering.
                                     Defaults to False.
        :type skip_data_migrations: bool, optional
        """
        self.client.register_schema(schema, operation_id=operation_id, skip_data_migrations=skip_data_migrations)

    def unregister_schema(self, class_name: str, operation_id: str | None = None) -> None:
        """
        Unregister a schema/model from AMSDAL application.

        :param class_name: The name of the schema/model to unregister.
        :type class_name: str
        :param operation_id: The operation id to use for this request. This is useful to prevent duplicated requests.
                             Defaults to None.
        :type operation_id: str, optional
        """
        self.client.unregister_schema(class_name, operation_id=operation_id)

    def create(self, class_name: str, data: dict[str, Any], operation_id: str | None = None) -> None:
        """
        Create a new object/data in AMSDAL application by specific class name.

        :param class_name: The name of the schema/model to create.
        :type class_name: str
        :param data: The data to create.
        :type data: dict[str, Any]
        :param operation_id: The operation id to use for this request. This is useful to prevent duplicated requests.
                             Defaults to None.
        :type operation_id: str, optional
        """
        self.client.create(class_name, data, operation_id=operation_id)

    def update(self, class_name: str, object_id: str, data: dict[str, Any], operation_id: str | None = None) -> None:
        """
        Update an existing object/data in AMSDAL application by specific class name and object id.

        :param class_name: The name of the schema/model to update.
        :type class_name: str
        :param object_id: The id of the object to update.
        :type object_id: str
        :param data: The data to update.
        :type data: dict[str, Any]
        :param operation_id: The operation id to use for this request. This is useful to prevent duplicated requests.
                             Defaults to None.
        :type operation_id: str, optional
        """
        self.client.update(class_name, object_id, data, operation_id=operation_id)

    def delete(self, class_name: str, object_id: str, operation_id: str | None = None) -> None:
        """
        Delete an existing object/data in AMSDAL application by specific class name and object id.

        :param class_name: The name of the schema/model to delete.
        :type class_name: str
        :param object_id: The id of the object to delete.
        :type object_id: str
        :param operation_id: The operation id to use for this request. This is useful to prevent duplicated requests.
                             Defaults to None.
        :type operation_id: str, optional
        """
        self.client.delete(class_name, object_id, operation_id=operation_id)

register_schema 

register_schema(
    schema, operation_id=None, *, skip_data_migrations=False
)

Register a new schema/model version in AMSDAL application.

Parameters:

Name Type Description Default
schema Schema

The schema/model to register.

 required
operation_id str | None

The operation id to use for this request. This is useful to prevent duplicated requests. Defaults to None.

 None
skip_data_migrations bool

Whether to skip data migrations if a new version of schema is registering. Defaults to False.

 False
Source code in amsdal_integrations/core.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def register_schema(
    self,
    schema: Schema,
    operation_id: str | None = None,
    *,
    skip_data_migrations: bool = False,
) -> None:
    """
    Register a new schema/model version in AMSDAL application.

    :param schema: The schema/model to register.
    :type schema: Schema
    :param operation_id: The operation id to use for this request. This is useful to prevent duplicated requests.
                         Defaults to None.
    :type operation_id: str, optional
    :param skip_data_migrations: Whether to skip data migrations if a new version of schema is registering.
                                 Defaults to False.
    :type skip_data_migrations: bool, optional
    """
    self.client.register_schema(schema, operation_id=operation_id, skip_data_migrations=skip_data_migrations)

unregister_schema 

unregister_schema(class_name, operation_id=None)

Unregister a schema/model from AMSDAL application.

Parameters:

Name Type Description Default
class_name str

The name of the schema/model to unregister.

 required
operation_id str | None

The operation id to use for this request. This is useful to prevent duplicated requests. Defaults to None.

 None
Source code in amsdal_integrations/core.py 
53
54
55
56
57
58
59
60
61
62
63
def unregister_schema(self, class_name: str, operation_id: str | None = None) -> None:
    """
    Unregister a schema/model from AMSDAL application.

    :param class_name: The name of the schema/model to unregister.
    :type class_name: str
    :param operation_id: The operation id to use for this request. This is useful to prevent duplicated requests.
                         Defaults to None.
    :type operation_id: str, optional
    """
    self.client.unregister_schema(class_name, operation_id=operation_id)

create 

create(class_name, data, operation_id=None)

Create a new object/data in AMSDAL application by specific class name.

Parameters:

Name Type Description Default
class_name str

The name of the schema/model to create.

 required
data dict[str, Any]

The data to create.

 required
operation_id str | None

The operation id to use for this request. This is useful to prevent duplicated requests. Defaults to None.

 None
Source code in amsdal_integrations/core.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
def create(self, class_name: str, data: dict[str, Any], operation_id: str | None = None) -> None:
    """
    Create a new object/data in AMSDAL application by specific class name.

    :param class_name: The name of the schema/model to create.
    :type class_name: str
    :param data: The data to create.
    :type data: dict[str, Any]
    :param operation_id: The operation id to use for this request. This is useful to prevent duplicated requests.
                         Defaults to None.
    :type operation_id: str, optional
    """
    self.client.create(class_name, data, operation_id=operation_id)

update 

update(class_name, object_id, data, operation_id=None)

Update an existing object/data in AMSDAL application by specific class name and object id.

Parameters:

Name Type Description Default
class_name str

The name of the schema/model to update.

 required
object_id str

The id of the object to update.

 required
data dict[str, Any]

The data to update.

 required
operation_id str | None

The operation id to use for this request. This is useful to prevent duplicated requests. Defaults to None.

 None
Source code in amsdal_integrations/core.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def update(self, class_name: str, object_id: str, data: dict[str, Any], operation_id: str | None = None) -> None:
    """
    Update an existing object/data in AMSDAL application by specific class name and object id.

    :param class_name: The name of the schema/model to update.
    :type class_name: str
    :param object_id: The id of the object to update.
    :type object_id: str
    :param data: The data to update.
    :type data: dict[str, Any]
    :param operation_id: The operation id to use for this request. This is useful to prevent duplicated requests.
                         Defaults to None.
    :type operation_id: str, optional
    """
    self.client.update(class_name, object_id, data, operation_id=operation_id)

delete 

delete(class_name, object_id, operation_id=None)

Delete an existing object/data in AMSDAL application by specific class name and object id.

Parameters:

Name Type Description Default
class_name str

The name of the schema/model to delete.

 required
object_id str

The id of the object to delete.

 required
operation_id str | None

The operation id to use for this request. This is useful to prevent duplicated requests. Defaults to None.

 None
Source code in amsdal_integrations/core.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def delete(self, class_name: str, object_id: str, operation_id: str | None = None) -> None:
    """
    Delete an existing object/data in AMSDAL application by specific class name and object id.

    :param class_name: The name of the schema/model to delete.
    :type class_name: str
    :param object_id: The id of the object to delete.
    :type object_id: str
    :param operation_id: The operation id to use for this request. This is useful to prevent duplicated requests.
                         Defaults to None.
    :type operation_id: str, optional
    """
    self.client.delete(class_name, object_id, operation_id=operation_id)

SDK Config

amsdal_integrations.data_classes.IntegrationConfig dataclass

Source code in amsdal_integrations/data_classes.py 
 8
 9
10
11
12
13
14
15
@dataclass
class IntegrationConfig:
    amsdal_host: str
    """The host of the destination AMSDAL application or of the running AMSDAL Agent. Example: http://localhost:8054"""
    amsdal_auth: Any
    """The auth mechanism that will be used to communicate with `amsdal_host`. Can be any httpx.Auth instance."""
    client_extra: dict[str, Any] = field(default_factory=dict)
    """Extra parameters that will be passed to the httpx.Client constructor. Example: {'timeout': 10}"""

amsdal_host instance-attribute 

amsdal_host

The host of the destination AMSDAL application or of the running AMSDAL Agent. Example: http://localhost:8054

amsdal_auth instance-attribute 

amsdal_auth

The auth mechanism that will be used to communicate with amsdal_host. Can be any httpx.Auth instance.

client_extra class-attribute instance-attribute 

client_extra = field(default_factory=dict)

Extra parameters that will be passed to the httpx.Client constructor. Example: {'timeout': 10}

Clients

amsdal_integrations.clients.amsdal_client.AmsdalClient

Bases: BaseClient

Source code in amsdal_integrations/clients/amsdal_client.py 
 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
class AmsdalClient(BaseClient):
    def __init__(self, host: str, auth: Any = None, **kwargs: Any) -> None:
        super().__init__(host, auth, **kwargs)
        self._operation_ids: set[str] = set()
        self._client = httpx.Client(
            base_url=self._host,
            auth=self._auth,
            timeout=10.0,
            **self._params,
        )

    @contextmanager
    def unique_operation_id(self, operation_id: str | None) -> Iterator[bool]:
        if operation_id is not None and operation_id in self._operation_ids:
            logger.warning('Operation ID %s already exists', operation_id)
            yield False
            return

        yield True

        if operation_id is not None:
            self._operation_ids.add(operation_id)

    def register_schema(
        self,
        schema: Schema,
        operation_id: str | None = None,
        *,
        skip_data_migrations: bool = False,
    ) -> None:
        """
        Sends a new schema of your model version directly to the AMSDAL Application.

        :param schema: Schema of your model version.
        :type schema: Schema
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        :param skip_data_migrations: Skip data migrations.
        :type skip_data_migrations: bool
        """
        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = self._client.post(
                '/api/classes/',
                params={'skip_data_migrations': skip_data_migrations},
                json={'class_schema': asdict(schema)},
            )
            response.raise_for_status()

    def unregister_schema(self, class_name: str, operation_id: str | None = None) -> None:
        """
        Send a request to unregister a schema of your model version directly from AMSDAL Application.

        :param class_name: Name of the class.
        :type class_name: str
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = self._client.delete(
                f'/api/classes/{class_name}/',
            )
            response.raise_for_status()

    def create(self, class_name: str, data: dict[str, Any], operation_id: str | None = None) -> None:
        """
        Send a request to create an object of your model directly to AMSDAL Application.

        :param class_name: Name of the class.
        :type class_name: str
        :param data: Data of the object.
        :type data: dict[str, Any]
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = self._client.post(
                '/api/objects/',
                json=data,
                params={
                    'class_name': class_name,
                    'load_references': False,
                },
            )
            response.raise_for_status()

    def update(self, class_name: str, object_id: str, data: dict[str, Any], operation_id: str | None = None) -> None:
        """
        Send a request to update an object of your model directly to AMSDAL Application.

        :param class_name: Name of the class.
        :type class_name: str
        :param object_id: ID of the object.
        :type object_id: str
        :param data: Data of the object.
        :type data: dict[str, Any]
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        address = build_address_string(class_name, object_id, class_version='ALL')

        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = self._client.post(
                quote(f'/api/objects/{address}/'),
                params={'load_references': False},
                json=data,
            )
            response.raise_for_status()

    def delete(self, class_name: str, object_id: str, operation_id: str | None = None) -> None:
        """
        Send a request to delete an object of your model directly to AMSDAL Application.

        :param class_name: Name of the class.
        :type class_name: str
        :param object_id: ID of the object.
        :type object_id: str
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        address = build_address_string(class_name, object_id)

        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = self._client.delete(
                quote(f'/api/objects/{address}/'),
            )
            response.raise_for_status()

register_schema 

register_schema(
    schema, operation_id=None, *, skip_data_migrations=False
)

Sends a new schema of your model version directly to the AMSDAL Application.

Parameters:

Name Type Description Default
schema Schema

Schema of your model version.

 required
operation_id str | None

Operation ID for the request.

 None
skip_data_migrations bool

Skip data migrations.

 False
Source code in amsdal_integrations/clients/amsdal_client.py 
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
def register_schema(
    self,
    schema: Schema,
    operation_id: str | None = None,
    *,
    skip_data_migrations: bool = False,
) -> None:
    """
    Sends a new schema of your model version directly to the AMSDAL Application.

    :param schema: Schema of your model version.
    :type schema: Schema
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    :param skip_data_migrations: Skip data migrations.
    :type skip_data_migrations: bool
    """
    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = self._client.post(
            '/api/classes/',
            params={'skip_data_migrations': skip_data_migrations},
            json={'class_schema': asdict(schema)},
        )
        response.raise_for_status()

unregister_schema 

unregister_schema(class_name, operation_id=None)

Send a request to unregister a schema of your model version directly from AMSDAL Application.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/amsdal_client.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def unregister_schema(self, class_name: str, operation_id: str | None = None) -> None:
    """
    Send a request to unregister a schema of your model version directly from AMSDAL Application.

    :param class_name: Name of the class.
    :type class_name: str
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = self._client.delete(
            f'/api/classes/{class_name}/',
        )
        response.raise_for_status()

create 

create(class_name, data, operation_id=None)

Send a request to create an object of your model directly to AMSDAL Application.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
data dict[str, Any]

Data of the object.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/amsdal_client.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def create(self, class_name: str, data: dict[str, Any], operation_id: str | None = None) -> None:
    """
    Send a request to create an object of your model directly to AMSDAL Application.

    :param class_name: Name of the class.
    :type class_name: str
    :param data: Data of the object.
    :type data: dict[str, Any]
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = self._client.post(
            '/api/objects/',
            json=data,
            params={
                'class_name': class_name,
                'load_references': False,
            },
        )
        response.raise_for_status()

update 

update(class_name, object_id, data, operation_id=None)

Send a request to update an object of your model directly to AMSDAL Application.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
object_id str

ID of the object.

 required
data dict[str, Any]

Data of the object.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/amsdal_client.py 
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
def update(self, class_name: str, object_id: str, data: dict[str, Any], operation_id: str | None = None) -> None:
    """
    Send a request to update an object of your model directly to AMSDAL Application.

    :param class_name: Name of the class.
    :type class_name: str
    :param object_id: ID of the object.
    :type object_id: str
    :param data: Data of the object.
    :type data: dict[str, Any]
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    address = build_address_string(class_name, object_id, class_version='ALL')

    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = self._client.post(
            quote(f'/api/objects/{address}/'),
            params={'load_references': False},
            json=data,
        )
        response.raise_for_status()

delete 

delete(class_name, object_id, operation_id=None)

Send a request to delete an object of your model directly to AMSDAL Application.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
object_id str

ID of the object.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/amsdal_client.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def delete(self, class_name: str, object_id: str, operation_id: str | None = None) -> None:
    """
    Send a request to delete an object of your model directly to AMSDAL Application.

    :param class_name: Name of the class.
    :type class_name: str
    :param object_id: ID of the object.
    :type object_id: str
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    address = build_address_string(class_name, object_id)

    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = self._client.delete(
            quote(f'/api/objects/{address}/'),
        )
        response.raise_for_status()

amsdal_integrations.clients.amsdal_client.AsyncAmsdalClient

Bases: AsyncBaseClient

Async version of the AMSDAL Client.

Source code in amsdal_integrations/clients/amsdal_client.py 
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
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
class AsyncAmsdalClient(AsyncBaseClient):
    """
    Async version of the AMSDAL Client.
    """

    def __init__(self, host: str, auth: Any = None, **kwargs: Any) -> None:
        super().__init__(host, auth, **kwargs)
        self._operation_ids: set[str] = set()
        self._client = httpx.AsyncClient(
            base_url=self._host,
            auth=self._auth,
            **self._params,
        )

    @contextmanager
    def unique_operation_id(self, operation_id: str | None) -> Iterator[bool]:
        if operation_id is not None and operation_id in self._operation_ids:
            logger.warning('Operation ID %s already exists', operation_id)
            yield False
            return

        yield True

        if operation_id is not None:
            self._operation_ids.add(operation_id)

    async def register_schema(
        self,
        schema: Schema,
        operation_id: str | None = None,
        *,
        skip_data_migrations: bool = False,
    ) -> None:
        """
        Sends a new schema of your model version directly to the AMSDAL Application.

        :param schema: Schema of your model version.
        :type schema: Schema
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        :param skip_data_migrations: Skip data migrations.
        :type skip_data_migrations: bool
        """
        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = await self._client.post(
                '/api/classes/',
                params={'skip_data_migrations': skip_data_migrations},
                json={'class_schema': asdict(schema)},
            )
            response.raise_for_status()

    async def unregister_schema(self, class_name: str, operation_id: str | None = None) -> None:
        """
        Send a request to unregister a schema of your model version directly from AMSDAL Application.

        :param class_name: Name of the class.
        :type class_name: str
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = await self._client.delete(
                f'/api/classes/{class_name}/',
            )
            response.raise_for_status()

    async def create(self, class_name: str, data: dict[str, Any], operation_id: str | None = None) -> None:
        """
        Send a request to create an object of your model directly to AMSDAL Application.

        :param class_name: Name of the class.
        :type class_name: str
        :param data: Data of the object.
        :type data: dict[str, Any]
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = await self._client.post(
                '/api/objects/',
                json=data,
                params={
                    'class_name': class_name,
                    'load_references': False,
                },
            )
            response.raise_for_status()

    async def update(
        self,
        class_name: str,
        object_id: str,
        data: dict[str, Any],
        operation_id: str | None = None,
    ) -> None:
        """
        Send a request to update an object of your model directly to AMSDAL Application.

        :param class_name: Name of the class.
        :type class_name: str
        :param object_id: ID of the object.
        :type object_id: str
        :param data: Data of the object.
        :type data: dict[str, Any]
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        address = build_address_string(class_name, object_id)

        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = await self._client.post(
                quote(f'/api/objects/{address}/'),
                params={'load_references': False},
                json=data,
            )
            response.raise_for_status()

    async def delete(self, class_name: str, object_id: str, operation_id: str | None = None) -> None:
        """
        Send a request to delete an object of your model directly to AMSDAL Application.

        :param class_name: Name of the class.
        :type class_name: str
        :param object_id: ID of the object.
        :type object_id: str
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        address = build_address_string(class_name, object_id)

        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = await self._client.delete(
                quote(f'/api/objects/{address}/'),
            )
            response.raise_for_status()

register_schema async 

register_schema(
    schema, operation_id=None, *, skip_data_migrations=False
)

Sends a new schema of your model version directly to the AMSDAL Application.

Parameters:

Name Type Description Default
schema Schema

Schema of your model version.

 required
operation_id str | None

Operation ID for the request.

 None
skip_data_migrations bool

Skip data migrations.

 False
Source code in amsdal_integrations/clients/amsdal_client.py 
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
async def register_schema(
    self,
    schema: Schema,
    operation_id: str | None = None,
    *,
    skip_data_migrations: bool = False,
) -> None:
    """
    Sends a new schema of your model version directly to the AMSDAL Application.

    :param schema: Schema of your model version.
    :type schema: Schema
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    :param skip_data_migrations: Skip data migrations.
    :type skip_data_migrations: bool
    """
    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = await self._client.post(
            '/api/classes/',
            params={'skip_data_migrations': skip_data_migrations},
            json={'class_schema': asdict(schema)},
        )
        response.raise_for_status()

unregister_schema async 

unregister_schema(class_name, operation_id=None)

Send a request to unregister a schema of your model version directly from AMSDAL Application.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/amsdal_client.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
async def unregister_schema(self, class_name: str, operation_id: str | None = None) -> None:
    """
    Send a request to unregister a schema of your model version directly from AMSDAL Application.

    :param class_name: Name of the class.
    :type class_name: str
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = await self._client.delete(
            f'/api/classes/{class_name}/',
        )
        response.raise_for_status()

create async 

create(class_name, data, operation_id=None)

Send a request to create an object of your model directly to AMSDAL Application.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
data dict[str, Any]

Data of the object.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/amsdal_client.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
async def create(self, class_name: str, data: dict[str, Any], operation_id: str | None = None) -> None:
    """
    Send a request to create an object of your model directly to AMSDAL Application.

    :param class_name: Name of the class.
    :type class_name: str
    :param data: Data of the object.
    :type data: dict[str, Any]
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = await self._client.post(
            '/api/objects/',
            json=data,
            params={
                'class_name': class_name,
                'load_references': False,
            },
        )
        response.raise_for_status()

update async 

update(class_name, object_id, data, operation_id=None)

Send a request to update an object of your model directly to AMSDAL Application.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
object_id str

ID of the object.

 required
data dict[str, Any]

Data of the object.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/amsdal_client.py 
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
async def update(
    self,
    class_name: str,
    object_id: str,
    data: dict[str, Any],
    operation_id: str | None = None,
) -> None:
    """
    Send a request to update an object of your model directly to AMSDAL Application.

    :param class_name: Name of the class.
    :type class_name: str
    :param object_id: ID of the object.
    :type object_id: str
    :param data: Data of the object.
    :type data: dict[str, Any]
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    address = build_address_string(class_name, object_id)

    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = await self._client.post(
            quote(f'/api/objects/{address}/'),
            params={'load_references': False},
            json=data,
        )
        response.raise_for_status()

delete async 

delete(class_name, object_id, operation_id=None)

Send a request to delete an object of your model directly to AMSDAL Application.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
object_id str

ID of the object.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/amsdal_client.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
async def delete(self, class_name: str, object_id: str, operation_id: str | None = None) -> None:
    """
    Send a request to delete an object of your model directly to AMSDAL Application.

    :param class_name: Name of the class.
    :type class_name: str
    :param object_id: ID of the object.
    :type object_id: str
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    address = build_address_string(class_name, object_id)

    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = await self._client.delete(
            quote(f'/api/objects/{address}/'),
        )
        response.raise_for_status()

amsdal_integrations.clients.agent_client.AgentClient

Bases: BaseClient

Agent Client that communicate with the AMSDAL Agent to avoid any performance issues.

Source code in amsdal_integrations/clients/agent_client.py 
 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
class AgentClient(BaseClient):
    """
    Agent Client that communicate with the AMSDAL Agent to avoid any performance issues.
    """

    def __init__(self, host: str, auth: Any = None, **kwargs: Any) -> None:
        super().__init__(host, auth, **kwargs)
        self._operation_ids: set[str] = set()
        self._client = httpx.Client(
            base_url=self._host,
            auth=self._auth,
            **self._params,
        )

    @contextmanager
    def unique_operation_id(self, operation_id: str | None) -> Iterator[bool]:
        if operation_id is not None and operation_id in self._operation_ids:
            logger.warning('Operation ID %s already exists', operation_id)
            yield False
            return

        yield True

        if operation_id is not None:
            self._operation_ids.add(operation_id)

    def register_schema(
        self,
        schema: Schema,
        operation_id: str | None = None,
        *,
        skip_data_migrations: bool = False,
    ) -> None:
        """
        Sends a new schema of your model version to the AMSDAL Agent.

        :param schema: Schema of your model version.
        :type schema: Schema
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        :param skip_data_migrations: Skip data migrations.
        :type skip_data_migrations: bool
        """
        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            logger.info('Agent Client: Registering class %s', schema.title)
            response = self._client.post(
                '/data',
                json={
                    'action': 'register_schema',
                    'params': {
                        'skip_data_migrations': skip_data_migrations,
                    },
                    'data': {'class_schema': asdict(schema)},
                },
            )
            response.raise_for_status()

    def unregister_schema(self, class_name: str, operation_id: str | None = None) -> None:
        """
        Send a request to unregister a schema of your model version from the AMSDAL Agent.

        :param class_name: Name of the class.
        :type class_name: str
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            logger.info('Agent Client: Unregister class %s', class_name)
            response = self._client.request(
                'DELETE',
                '/data',
                json={
                    'action': 'unregister_schema',
                    'path_params': [class_name],
                },
            )
            response.raise_for_status()

    def create(self, class_name: str, data: dict[str, Any], operation_id: str | None = None) -> None:
        """
        Send a request to create an object of your model to the AMSDAL Agent.

        :param class_name: Name of the class.
        :type class_name: str
        :param data: Data of the object.
        :type data: dict[str, Any]
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            logger.info('Agent Client: Create object for class %s', class_name)
            response = self._client.post(
                '/data',
                json={
                    'action': 'object_create',
                    'data': data,
                    'params': {
                        'class_name': class_name,
                        'load_references': False,
                    },
                },
            )
            response.raise_for_status()

    def update(self, class_name: str, object_id: str, data: dict[str, Any], operation_id: str | None = None) -> None:
        """
        Send a request to update an object of your model to the AMSDAL Agent.

        :param class_name: Name of the class.
        :type class_name: str
        :param object_id: ID of the object.
        :type object_id: str
        :param data: Data of the object.
        :type data: dict[str, Any]
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        address = build_address_string(class_name, object_id)

        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            logger.info('Agent Client: Update object for class %s', class_name)
            response = self._client.post(
                '/data',
                json={
                    'action': 'object_update',
                    'data': data,
                    'path_params': [address],
                    'params': {'load_references': False},
                },
            )
            response.raise_for_status()

    def delete(self, class_name: str, object_id: str, operation_id: str | None = None) -> None:
        """
        Send a request to delete an object of your model to the AMSDAL Agent.

        :param class_name: Name of the class.
        :type class_name: str
        :param object_id: ID of the object.
        :type object_id: str
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        address = build_address_string(class_name, object_id)

        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            logger.info('Agent Client: Delete object %s', address)
            response = self._client.request(
                'DELETE',
                '/data',
                json={
                    'action': 'object_delete',
                    'path_params': [address],
                },
            )
            response.raise_for_status()

register_schema 

register_schema(
    schema, operation_id=None, *, skip_data_migrations=False
)

Sends a new schema of your model version to the AMSDAL Agent.

Parameters:

Name Type Description Default
schema Schema

Schema of your model version.

 required
operation_id str | None

Operation ID for the request.

 None
skip_data_migrations bool

Skip data migrations.

 False
Source code in amsdal_integrations/clients/agent_client.py 
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
def register_schema(
    self,
    schema: Schema,
    operation_id: str | None = None,
    *,
    skip_data_migrations: bool = False,
) -> None:
    """
    Sends a new schema of your model version to the AMSDAL Agent.

    :param schema: Schema of your model version.
    :type schema: Schema
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    :param skip_data_migrations: Skip data migrations.
    :type skip_data_migrations: bool
    """
    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        logger.info('Agent Client: Registering class %s', schema.title)
        response = self._client.post(
            '/data',
            json={
                'action': 'register_schema',
                'params': {
                    'skip_data_migrations': skip_data_migrations,
                },
                'data': {'class_schema': asdict(schema)},
            },
        )
        response.raise_for_status()

unregister_schema 

unregister_schema(class_name, operation_id=None)

Send a request to unregister a schema of your model version from the AMSDAL Agent.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/agent_client.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def unregister_schema(self, class_name: str, operation_id: str | None = None) -> None:
    """
    Send a request to unregister a schema of your model version from the AMSDAL Agent.

    :param class_name: Name of the class.
    :type class_name: str
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        logger.info('Agent Client: Unregister class %s', class_name)
        response = self._client.request(
            'DELETE',
            '/data',
            json={
                'action': 'unregister_schema',
                'path_params': [class_name],
            },
        )
        response.raise_for_status()

create 

create(class_name, data, operation_id=None)

Send a request to create an object of your model to the AMSDAL Agent.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
data dict[str, Any]

Data of the object.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/agent_client.py 
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
def create(self, class_name: str, data: dict[str, Any], operation_id: str | None = None) -> None:
    """
    Send a request to create an object of your model to the AMSDAL Agent.

    :param class_name: Name of the class.
    :type class_name: str
    :param data: Data of the object.
    :type data: dict[str, Any]
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        logger.info('Agent Client: Create object for class %s', class_name)
        response = self._client.post(
            '/data',
            json={
                'action': 'object_create',
                'data': data,
                'params': {
                    'class_name': class_name,
                    'load_references': False,
                },
            },
        )
        response.raise_for_status()

update 

update(class_name, object_id, data, operation_id=None)

Send a request to update an object of your model to the AMSDAL Agent.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
object_id str

ID of the object.

 required
data dict[str, Any]

Data of the object.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/agent_client.py 
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
def update(self, class_name: str, object_id: str, data: dict[str, Any], operation_id: str | None = None) -> None:
    """
    Send a request to update an object of your model to the AMSDAL Agent.

    :param class_name: Name of the class.
    :type class_name: str
    :param object_id: ID of the object.
    :type object_id: str
    :param data: Data of the object.
    :type data: dict[str, Any]
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    address = build_address_string(class_name, object_id)

    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        logger.info('Agent Client: Update object for class %s', class_name)
        response = self._client.post(
            '/data',
            json={
                'action': 'object_update',
                'data': data,
                'path_params': [address],
                'params': {'load_references': False},
            },
        )
        response.raise_for_status()

delete 

delete(class_name, object_id, operation_id=None)

Send a request to delete an object of your model to the AMSDAL Agent.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
object_id str

ID of the object.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/agent_client.py 
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 delete(self, class_name: str, object_id: str, operation_id: str | None = None) -> None:
    """
    Send a request to delete an object of your model to the AMSDAL Agent.

    :param class_name: Name of the class.
    :type class_name: str
    :param object_id: ID of the object.
    :type object_id: str
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    address = build_address_string(class_name, object_id)

    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        logger.info('Agent Client: Delete object %s', address)
        response = self._client.request(
            'DELETE',
            '/data',
            json={
                'action': 'object_delete',
                'path_params': [address],
            },
        )
        response.raise_for_status()

amsdal_integrations.clients.agent_client.AsyncAgentClient

Bases: AsyncBaseClient

Async Agent Client that communicate with the AMSDAL Agent to avoid any performance issues.

Source code in amsdal_integrations/clients/agent_client.py 
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
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
class AsyncAgentClient(AsyncBaseClient):
    """
    Async Agent Client that communicate with the AMSDAL Agent to avoid any performance issues.
    """

    def __init__(self, host: str, auth: Any = None, **kwargs: Any) -> None:
        super().__init__(host, auth, **kwargs)
        self._operation_ids: set[str] = set()
        self._client = httpx.AsyncClient(
            base_url=self._host,
            auth=self._auth,
            **self._params,
        )

    @contextmanager
    def unique_operation_id(self, operation_id: str | None) -> Iterator[bool]:
        if operation_id is not None and operation_id in self._operation_ids:
            logger.warning('Operation ID %s already exists', operation_id)
            yield False
            return

        yield True

        if operation_id is not None:
            self._operation_ids.add(operation_id)

    async def register_schema(
        self,
        schema: Schema,
        operation_id: str | None = None,
        *,
        skip_data_migrations: bool = False,
    ) -> None:
        """
        Sends a new schema of your model version to the AMSDAL Agent.

        :param schema: Schema of your model version.
        :type schema: Schema
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        :param skip_data_migrations: Skip data migrations.
        :type skip_data_migrations: bool
        """
        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = await self._client.post(
                '/data',
                json={
                    'action': 'register_schema',
                    'params': {
                        'skip_data_migrations': skip_data_migrations,
                    },
                    'data': asdict(schema),
                },
            )
            response.raise_for_status()

    async def unregister_schema(self, class_name: str, operation_id: str | None = None) -> None:
        """
        Send a request to unregister a schema of your model version from the AMSDAL Agent.

        :param class_name: Name of the class.
        :type class_name: str
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = await self._client.request(
                'DELETE',
                '/data',
                json={
                    'action': 'unregister_schema',
                    'path_params': [class_name],
                },
            )
            response.raise_for_status()

    async def create(self, class_name: str, data: dict[str, Any], operation_id: str | None = None) -> None:
        """
        Send a request to create an object of your model to the AMSDAL Agent.

        :param class_name: Name of the class.
        :type class_name: str
        :param data: Data of the object.
        :type data: dict[str, Any]
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = await self._client.post(
                '/data',
                json={
                    'action': 'object_create',
                    'data': data,
                    'params': {
                        'class_name': class_name,
                        'load_references': False,
                    },
                },
            )
            response.raise_for_status()

    async def update(
        self,
        class_name: str,
        object_id: str,
        data: dict[str, Any],
        operation_id: str | None = None,
    ) -> None:
        """
        Send a request to update an object of your model to the AMSDAL Agent.

        :param class_name: Name of the class.
        :type class_name: str
        :param object_id: ID of the object.
        :type object_id: str
        :param data: Data of the object.
        :type data: dict[str, Any]
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        address = build_address_string(class_name, object_id)

        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = await self._client.post(
                '/data',
                json={
                    'action': 'object_update',
                    'data': data,
                    'path_params': [address],
                    'params': {'load_references': False},
                },
            )
            response.raise_for_status()

    async def delete(self, class_name: str, object_id: str, operation_id: str | None = None) -> None:
        """
        Send a request to delete an object of your model to the AMSDAL Agent.

        :param class_name: Name of the class.
        :type class_name: str
        :param object_id: ID of the object.
        :type object_id: str
        :param operation_id: Operation ID for the request.
        :type operation_id: str
        """
        address = build_address_string(class_name, object_id)

        with self.unique_operation_id(operation_id) as do_request:
            if not do_request:
                return

            response = await self._client.request(
                'DELETE',
                '/data',
                json={
                    'action': 'object_delete',
                    'path_params': [address],
                },
            )
            response.raise_for_status()

register_schema async 

register_schema(
    schema, operation_id=None, *, skip_data_migrations=False
)

Sends a new schema of your model version to the AMSDAL Agent.

Parameters:

Name Type Description Default
schema Schema

Schema of your model version.

 required
operation_id str | None

Operation ID for the request.

 None
skip_data_migrations bool

Skip data migrations.

 False
Source code in amsdal_integrations/clients/agent_client.py 
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
async def register_schema(
    self,
    schema: Schema,
    operation_id: str | None = None,
    *,
    skip_data_migrations: bool = False,
) -> None:
    """
    Sends a new schema of your model version to the AMSDAL Agent.

    :param schema: Schema of your model version.
    :type schema: Schema
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    :param skip_data_migrations: Skip data migrations.
    :type skip_data_migrations: bool
    """
    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = await self._client.post(
            '/data',
            json={
                'action': 'register_schema',
                'params': {
                    'skip_data_migrations': skip_data_migrations,
                },
                'data': asdict(schema),
            },
        )
        response.raise_for_status()

unregister_schema async 

unregister_schema(class_name, operation_id=None)

Send a request to unregister a schema of your model version from the AMSDAL Agent.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/agent_client.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
async def unregister_schema(self, class_name: str, operation_id: str | None = None) -> None:
    """
    Send a request to unregister a schema of your model version from the AMSDAL Agent.

    :param class_name: Name of the class.
    :type class_name: str
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = await self._client.request(
            'DELETE',
            '/data',
            json={
                'action': 'unregister_schema',
                'path_params': [class_name],
            },
        )
        response.raise_for_status()

create async 

create(class_name, data, operation_id=None)

Send a request to create an object of your model to the AMSDAL Agent.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
data dict[str, Any]

Data of the object.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/agent_client.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
async def create(self, class_name: str, data: dict[str, Any], operation_id: str | None = None) -> None:
    """
    Send a request to create an object of your model to the AMSDAL Agent.

    :param class_name: Name of the class.
    :type class_name: str
    :param data: Data of the object.
    :type data: dict[str, Any]
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = await self._client.post(
            '/data',
            json={
                'action': 'object_create',
                'data': data,
                'params': {
                    'class_name': class_name,
                    'load_references': False,
                },
            },
        )
        response.raise_for_status()

update async 

update(class_name, object_id, data, operation_id=None)

Send a request to update an object of your model to the AMSDAL Agent.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
object_id str

ID of the object.

 required
data dict[str, Any]

Data of the object.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/agent_client.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
async def update(
    self,
    class_name: str,
    object_id: str,
    data: dict[str, Any],
    operation_id: str | None = None,
) -> None:
    """
    Send a request to update an object of your model to the AMSDAL Agent.

    :param class_name: Name of the class.
    :type class_name: str
    :param object_id: ID of the object.
    :type object_id: str
    :param data: Data of the object.
    :type data: dict[str, Any]
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    address = build_address_string(class_name, object_id)

    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = await self._client.post(
            '/data',
            json={
                'action': 'object_update',
                'data': data,
                'path_params': [address],
                'params': {'load_references': False},
            },
        )
        response.raise_for_status()

delete async 

delete(class_name, object_id, operation_id=None)

Send a request to delete an object of your model to the AMSDAL Agent.

Parameters:

Name Type Description Default
class_name str

Name of the class.

 required
object_id str

ID of the object.

 required
operation_id str | None

Operation ID for the request.

 None
Source code in amsdal_integrations/clients/agent_client.py 
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
async def delete(self, class_name: str, object_id: str, operation_id: str | None = None) -> None:
    """
    Send a request to delete an object of your model to the AMSDAL Agent.

    :param class_name: Name of the class.
    :type class_name: str
    :param object_id: ID of the object.
    :type object_id: str
    :param operation_id: Operation ID for the request.
    :type operation_id: str
    """
    address = build_address_string(class_name, object_id)

    with self.unique_operation_id(operation_id) as do_request:
        if not do_request:
            return

        response = await self._client.request(
            'DELETE',
            '/data',
            json={
                'action': 'object_delete',
                'path_params': [address],
            },
        )
        response.raise_for_status()

Data classes

amsdal_integrations.data_classes.Schema dataclass

Source code in amsdal_integrations/data_classes.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@dataclass
class Schema:
    title: str
    """The name of the schema/model/class. Should be valid Python class name. Example: User"""
    type: str = 'object'
    """The type of the schema. Should be `object` for now."""
    properties: dict[str, PropertySchema] = field(default_factory=dict)
    """The properties of the schema. Example: {'name': {'type': 'string', 'default': 'John Doe'}}"""
    required: list[str] = field(default_factory=list)
    """The required properties of the schema. Example: ['name']"""
    indexed: list[str] = field(default_factory=list)
    """The indexed properties of the schema. Example: ['name']"""
    unique: list[list[str]] = field(default_factory=list)
    """The unique properties of the schema. Example: [['name', 'email']]"""
    custom_code: str = ''
    """The custom code of the schema. Example: `def __str__(self): return self.name`"""

title instance-attribute 

title

The name of the schema/model/class. Should be valid Python class name. Example: User

type class-attribute instance-attribute 

type = 'object'

The type of the schema. Should be object for now.

properties class-attribute instance-attribute 

properties = field(default_factory=dict)

The properties of the schema. Example: {'name': {'type': 'string', 'default': 'John Doe'}}

required class-attribute instance-attribute 

required = field(default_factory=list)

The required properties of the schema. Example: ['name']

indexed class-attribute instance-attribute 

indexed = field(default_factory=list)

The indexed properties of the schema. Example: ['name']

unique class-attribute instance-attribute 

unique = field(default_factory=list)

The unique properties of the schema. Example: [['name', 'email']]

custom_code class-attribute instance-attribute 

custom_code = ''

The custom code of the schema. Example: def __str__(self): return self.name

amsdal_integrations.data_classes.PropertySchema dataclass

Source code in amsdal_integrations/data_classes.py 
42
43
44
45
46
47
48
49
50
51
52
53
@dataclass
class PropertySchema:
    type: str
    """The type of the property. Should be valid AMSDAL type. Example: `string`"""
    default: Any
    """The default value of the property. Example: `John Doe`"""
    title: str | None = None
    """The title of the property. Example: `Name`"""
    items: TypeSchema | DictItem | None = None
    """The items of the property in case of type is `array` or `dictionary`. Example: `{'type': 'string'}`"""
    options: list[OptionSchema] | None = None
    """The options of available values for this property."""

type instance-attribute 

type

The type of the property. Should be valid AMSDAL type. Example: string

default instance-attribute 

default

The default value of the property. Example: John Doe

title class-attribute instance-attribute 

title = None

The title of the property. Example: Name

items class-attribute instance-attribute 

items = None

The items of the property in case of type is array or dictionary. Example: {'type': 'string'}

options class-attribute instance-attribute 

options = None

The options of available values for this property.

amsdal_integrations.data_classes.OptionSchema dataclass

Source code in amsdal_integrations/data_classes.py 
34
35
36
37
38
39
@dataclass
class OptionSchema:
    key: str
    """The key of the option. Example: `John Doe`"""
    value: str
    """The value of the option. Example: `John Doe`"""

key instance-attribute 

key

The key of the option. Example: John Doe

value instance-attribute 

value

The value of the option. Example: John Doe

amsdal_integrations.data_classes.TypeSchema dataclass

Source code in amsdal_integrations/data_classes.py 
26
27
28
29
30
31
@dataclass
class TypeSchema:
    type: str
    """The type of the schema. Should be valid AMSDAL type. Example: `string`"""
    items: Optional[Union[DictItem, 'TypeSchema']] = None  # noqa: UP007
    """The items of the schema in case of type is `array` or `dictionary`. Example: `{'type': 'string'}`"""

type instance-attribute 

type

The type of the schema. Should be valid AMSDAL type. Example: string

items class-attribute instance-attribute 

items = None

The items of the schema in case of type is array or dictionary. Example: {'type': 'string'}

amsdal_integrations.data_classes.DictItem dataclass

Source code in amsdal_integrations/data_classes.py 
18
19
20
21
22
23
@dataclass
class DictItem:
    key: 'TypeSchema'
    """The key of the dictionary. Example: `{'type': 'string'}`"""
    value: 'TypeSchema'
    """The value of the dictionary. Example: `{'type': 'string'}`"""

key instance-attribute 

key

The key of the dictionary. Example: {'type': 'string'}

value instance-attribute 

value

The value of the dictionary. Example: {'type': 'string'}