AAS HTTP Client Documentation
Loading...
Searching...
No Matches
sm_implementation.py
Go to the documentation of this file.
1"""Implementation of Submodel related API calls."""
2
3import json
4import logging
5from typing import TYPE_CHECKING
6
7import requests
8from pydantic import BaseModel
9
10if TYPE_CHECKING:
11 from aas_http_client.classes.client.aas_client import AasHttpClient
12
13from aas_http_client.utilities.constants import LogIntensity
14from aas_http_client.utilities.encoder import encode_base_64
16 STATUS_CODE_200,
17 STATUS_CODE_201,
18 STATUS_CODE_204,
19 STATUS_CODE_404,
20 log_response,
21)
22
23_logger = logging.getLogger(__name__)
24
25
27 """Implementation of Submodel related API calls."""
28
29 def __init__(self, client: "AasHttpClient"):
30 """Initializes the SmImplementation with the given parameters."""
31 self._client = client
32
33 session = client.get_session()
34 if session is None:
35 raise ValueError(
36 "HTTP session is not initialized in the client. Call 'initialize()' method of the client before creating SubmodelRegistryImplementation instance."
37 )
38
39 self._session: requests.Session = session
40
41 # GET /submodels/{submodelIdentifier}
42 def get_submodel_by_id(self, submodel_identifier: str, level: str = "", extent: str = "") -> dict | None:
43 """Returns a specific Submodel.
44
45 :param submodel_identifier: The Submodels unique id
46 :param level: Determines the structural depth of the respective resource content. Available values : deep, core
47 :param extent: Determines to which extent the resource is being serialized. Available values : withBlobValue, withoutBlobValue
48 :return: Submodel data or None if an error occurred
49 """
50 if not self._client.encoded_ids:
51 submodel_identifier = encode_base_64(submodel_identifier)
53 url = f"{self._client.base_url}/submodels/{submodel_identifier}"
54
55 params: dict[str, str] = {}
56 if level:
57 params["level"] = level
58 if extent:
59 params["extent"] = extent
60
61 self._client.set_token()
62
63 try:
64 response = self._session.get(url, params=params, timeout=self._client.time_out)
65 _logger.debug(f"Call REST API url '{response.url}'")
66
67 if response.status_code == STATUS_CODE_404:
68 if self._client.get_log_intensity() == LogIntensity.HIGH:
69 _logger.warning(f"Submodel with id '{submodel_identifier}' not found.")
70 elif self._client.get_log_intensity() == LogIntensity.STANDARD:
71 _logger.debug(f"Submodel with id '{submodel_identifier}' not found.")
72
73 _logger.debug(response.text)
74 return None
75
76 if response.status_code != STATUS_CODE_200:
77 log_response(response)
78 return None
79
80 except requests.exceptions.RequestException as e:
81 _logger.error(f"Error call REST API: {e}")
82 return None
83
84 content = response.content.decode("utf-8")
85 return json.loads(content)
86
87 # PUT /submodels/{submodelIdentifier}
88 def put_submodels_by_id(self, submodel_identifier: str, request_body: dict) -> bool:
89 """Updates a existing Submodel.
90
91 :param submodel_identifier: The Submodels unique id
92 :param request_body: Json data of the Submodel to update
93 :return: True if the update was successful, False otherwise
94 """
95 if not self._client.encoded_ids:
96 submodel_identifier = encode_base_64(submodel_identifier)
98 url = f"{self._client.base_url}/submodels/{submodel_identifier}"
99
100 self._client.set_token()
101
102 try:
103 response = self._session.put(url, json=request_body, timeout=self._client.time_out)
104 _logger.debug(f"Call REST API url '{response.url}'")
105
106 if response.status_code == STATUS_CODE_404:
107 if self._client.get_log_intensity() == LogIntensity.HIGH:
108 _logger.warning(f"Submodel with id '{submodel_identifier}' not found.")
109 elif self._client.get_log_intensity() == LogIntensity.STANDARD:
110 _logger.debug(f"Submodel with id '{submodel_identifier}' not found.")
111
112 _logger.debug(response.text)
113 return False
114
115 if response.status_code != STATUS_CODE_204:
116 log_response(response)
117 return False
118
119 except requests.exceptions.RequestException as e:
120 _logger.error(f"Error call REST API: {e}")
121 return False
122
123 return True
124
125 # DELETE /submodels/{submodelIdentifier}
126 def delete_submodel_by_id(self, submodel_identifier: str) -> bool:
127 """Deletes a Submodel.
128
129 :param submodel_identifier: The Submodels unique id
130 :return: True if the deletion was successful, False otherwise
131 """
132 if not self._client.encoded_ids:
133 submodel_identifier = encode_base_64(submodel_identifier)
135 url = f"{self._client.base_url}/submodels/{submodel_identifier}"
136
137 self._client.set_token()
138
139 try:
140 response = self._session.delete(url, timeout=self._client.time_out)
141 _logger.debug(f"Call REST API url '{response.url}'")
142
143 if response.status_code == STATUS_CODE_404:
144 if self._client.get_log_intensity() == LogIntensity.HIGH:
145 _logger.warning(f"Submodel with id '{submodel_identifier}' not found.")
146 elif self._client.get_log_intensity() == LogIntensity.STANDARD:
147 _logger.debug(f"Submodel with id '{submodel_identifier}' not found.")
148
149 _logger.debug(response.text)
150 return False
151
152 if response.status_code != STATUS_CODE_204:
153 log_response(response)
154 return False
155
156 except requests.exceptions.RequestException as e:
157 _logger.error(f"Error call REST API: {e}")
158 return False
159
160 return True
161
162 # GET /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}
164 self, submodel_identifier: str, id_short_path: str, level: str = "", extent: str = ""
165 ) -> dict | None:
166 """Returns a specific submodel element from the Submodel at a specified path.
167
168 :param submodel_identifier: The Submodels unique id
169 :param id_short_path: IdShort path to the submodel element (dot-separated)
170 :param level: Determines the structural depth of the respective resource content. Available values : deep, core
171 :param extent: Determines to which extent the resource is being serialized. Available values : withBlobValue, withoutBlobValue
172 :return: Submodel element data or None if an error occurred
173 """
174 if not self._client.encoded_ids:
175 submodel_identifier = encode_base_64(submodel_identifier)
176
177 url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements/{id_short_path}"
178
179 params: dict[str, str] = {}
180 if level:
181 params["level"] = level
182 if extent:
183 params["extent"] = extent
184
185 self._client.set_token()
186
187 try:
188 response = self._session.get(url, params=params, timeout=self._client.time_out)
189 _logger.debug(f"Call REST API url '{response.url}'")
190
191 if response.status_code == STATUS_CODE_404:
192 if self._client.get_log_intensity() == LogIntensity.HIGH:
193 _logger.warning(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
194 elif self._client.get_log_intensity() == LogIntensity.STANDARD:
195 _logger.debug(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
196
197 _logger.debug(response.text)
198 return None
199
200 if response.status_code != STATUS_CODE_200:
201 log_response(response)
202 return None
203
204 except requests.exceptions.RequestException as e:
205 _logger.error(f"Error call REST API: {e}")
206 return None
207
208 content = response.content.decode("utf-8")
209 return json.loads(content)
210
211 # PUT /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}
212 def put_submodel_element_by_path_submodel_repo(self, submodel_identifier: str, id_short_path: str, request_body: dict, level: str = "") -> bool:
213 """Creates or updates an existing submodel element at a specified path within submodel elements hierarchy.
214
215 :param submodel_identifier: The Submodels unique id
216 :param id_short_path: IdShort path to the submodel element (dot-separated)
217 :param request_body: Data for the submodel element
218 :param level: Determines the structural depth of the respective resource content. Available values : deep, core
219 :return: True if the operation was successful, False otherwise
220 """
221 if not self._client.encoded_ids:
222 submodel_identifier = encode_base_64(submodel_identifier)
224 url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements/{id_short_path}"
225
226 params: dict[str, str] = {}
227 if level:
228 params["level"] = level
229
230 self._client.set_token()
231
232 try:
233 response = self._session.put(url, json=request_body, params=params, timeout=self._client.time_out)
234 _logger.debug(f"Call REST API url '{response.url}'")
235
236 if response.status_code == STATUS_CODE_404:
237 if self._client.get_log_intensity() == LogIntensity.HIGH:
238 _logger.warning(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
239 elif self._client.get_log_intensity() == LogIntensity.STANDARD:
240 _logger.debug(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
241
242 _logger.debug(response.text)
243 return False
244
245 if response.status_code != STATUS_CODE_204:
246 log_response(response)
247 return False
248
249 except requests.exceptions.RequestException as e:
250 _logger.error(f"Error call REST API: {e}")
251 return False
252
253 return True
254
255 # POST /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}
257 self, submodel_identifier: str, id_short_path: str, request_body: dict, level: str = "", extent: str = ""
258 ) -> dict | None:
259 """Creates a new submodel element at a specified path within submodel elements hierarchy.
260
261 :param submodel_identifier: The Submodels unique id
262 :param id_short_path: IdShort path to the submodel element (dot-separated)
263 :param request_body: Data for the new Submodel element
264 :param level: Determines the structural depth of the respective resource content. Available values : deep, core
265 :param extent: Determines to which extent the resource is being serialized. Available values : withBlobValue, withoutBlobValue
266 :return: Submodel element data or None if an error occurred
267 """
268 if not self._client.encoded_ids:
269 submodel_identifier = encode_base_64(submodel_identifier)
270
271 url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements/{id_short_path}"
272
273 params: dict[str, str] = {}
274 if level:
275 params["level"] = level
276 if extent:
277 params["extent"] = extent
278
279 self._client.set_token()
280
281 try:
282 response = self._session.post(url, json=request_body, params=params, timeout=self._client.time_out)
283 _logger.debug(f"Call REST API url '{response.url}'")
284
285 if response.status_code == STATUS_CODE_404:
286 if self._client.get_log_intensity() == LogIntensity.HIGH:
287 _logger.warning(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
288 elif self._client.get_log_intensity() == LogIntensity.STANDARD:
289 _logger.debug(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
290
291 _logger.debug(response.text)
292 return None
293
294 if response.status_code != STATUS_CODE_201:
295 log_response(response)
296 return None
297
298 except requests.exceptions.RequestException as e:
299 _logger.error(f"Error call REST API: {e}")
300 return None
301
302 content = response.content.decode("utf-8")
303 return json.loads(content)
304
305 # DELETE /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}
306 def delete_submodel_element_by_path_submodel_repo(self, submodel_identifier: str, id_short_path: str) -> bool:
307 """Deletes a submodel element at a specified path within the submodel elements hierarchy.
308
309 :param submodel_identifier: The Submodels unique id
310 :param id_short_path: IdShort path to the submodel element (dot-separated)
311 :return: True if the deletion was successful, False otherwise
312 """
313 if not self._client.encoded_ids:
314 submodel_identifier = encode_base_64(submodel_identifier)
316 url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements/{id_short_path}"
317
318 self._client.set_token()
319 try:
320 response = self._session.delete(url, timeout=self._client.time_out)
321 _logger.debug(f"Call REST API url '{response.url}'")
322
323 if response.status_code == STATUS_CODE_404:
324 if self._client.get_log_intensity() == LogIntensity.HIGH:
325 _logger.warning(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
326 elif self._client.get_log_intensity() == LogIntensity.STANDARD:
327 _logger.debug(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
328
329 _logger.debug(response.text)
330 return False
331
332 if response.status_code != STATUS_CODE_204:
333 log_response(response)
334 return False
335
336 except requests.exceptions.RequestException as e:
337 _logger.error(f"Error call REST API: {e}")
338 return False
339
340 return True
341
342 # GET /submodels
344 self, semantic_id: str = "", id_short: str = "", limit: int = 0, cursor: str = "", level: str = "", extent: str = ""
345 ) -> dict | None:
346 """Returns all Submodels.
347
348 :param semantic_id: The value of the semantic id reference (UTF8-BASE64-URL-encoded)
349 :param id_short: The Submodels IdShort
350 :param limit: The maximum number of elements in the response array
351 :param cursor: A server-generated identifier retrieved from pagingMetadata that specifies from which position the result listing should continue
352 :param level: Determines the structural depth of the respective resource content. Available values : deep, core
353 :param extent: Determines to which extent the resource is being serialized. Available values : withBlobValue, withoutBlobValue
354 :return: List of Submodel data or None if an error occurred
355 """
356 url = f"{self._client.base_url}/submodels"
357
358 params: dict[str, str] = {}
359 if semantic_id:
360 params["semanticId"] = semantic_id
361 if id_short:
362 params["idShort"] = id_short
363 if limit:
364 params["limit"] = str(limit)
365 if cursor:
366 params["cursor"] = cursor
367 if level:
368 params["level"] = level
369 if extent:
370 params["extent"] = extent
371
372 self._client.set_token()
373
374 try:
375 response = self._session.get(url, params=params, timeout=self._client.time_out)
376 _logger.debug(f"Call REST API url '{response.url}'")
377
378 if response.status_code != STATUS_CODE_200:
379 log_response(response)
380 return None
381
382 except requests.exceptions.RequestException as e:
383 _logger.error(f"Error call REST API: {e}")
384 return None
385
386 content = response.content.decode("utf-8")
387 return json.loads(content)
388
389 # POST /submodels
390 def post_submodel(self, request_body: dict) -> dict | None:
391 """Creates a new Submodel.
392
393 :param request_body: Json data of the Submodel to post
394 :return: Submodel data or None if an error occurred
395 """
396 url = f"{self._client.base_url}/submodels"
397
398 self._client.set_token()
399
400 try:
401 response = self._session.post(url, json=request_body, timeout=self._client.time_out)
402 _logger.debug(f"Call REST API url '{response.url}'")
403
404 if response.status_code != STATUS_CODE_201:
405 log_response(response)
406 return None
407
408 except requests.exceptions.RequestException as e:
409 _logger.error(f"Error call REST API: {e}")
410 return None
411
412 content = response.content.decode("utf-8")
413 return json.loads(content)
414
415 # GET /submodels/{submodelIdentifier}/submodel-elements
417 self, submodel_identifier: str, limit: int = 100, cursor: str = "", level: str = "", extent: str = ""
418 ) -> dict | None:
419 """Returns all submodel elements including their hierarchy.
420
421 :param submodel_identifier: The Submodels unique id
422 :param limit: The maximum number of elements in the response array
423 :param cursor: A server-generated identifier retrieved from pagingMetadata that specifies from which position the result listing should continue
424 :param level: Determines the structural depth of the respective resource content. Available values : deep, core
425 :param extent: Determines to which extent the resource is being serialized. Available values : withBlobValue, withoutBlobValue
426 :return: List of Submodel element data or None if an error occurred
427 """
428 if not self._client.encoded_ids:
429 submodel_identifier = encode_base_64(submodel_identifier)
430
431 url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements"
432
433 params: dict[str, str] = {}
434 if limit:
435 params["limit"] = str(limit)
436 if cursor:
437 params["cursor"] = cursor
438 if level:
439 params["level"] = level
440 if extent:
441 params["extent"] = extent
442
443 self._client.set_token()
444
445 try:
446 response = self._session.get(url, params=params, timeout=self._client.time_out)
447 _logger.debug(f"Call REST API url '{response.url}'")
448
449 if response.status_code != STATUS_CODE_200:
450 log_response(response)
451 return None
452
453 except requests.exceptions.RequestException as e:
454 _logger.error(f"Error call REST API: {e}")
455 return None
456
457 content = response.content.decode("utf-8")
458 return json.loads(content)
459
460 # POST /submodels/{submodelIdentifier}/submodel-elements
461 def post_submodel_element_submodel_repo(self, submodel_identifier: str, request_body: dict) -> dict | None:
462 """Creates a new submodel element.
463
464 :param submodel_identifier: The Submodels unique id
465 :param request_body: Data for the new Submodel element
466 :return: Submodel element data or None if an error occurred
467 """
468 if not self._client.encoded_ids:
469 submodel_identifier = encode_base_64(submodel_identifier)
471 url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements"
472
473 self._client.set_token()
474
475 try:
476 response = self._session.post(url, json=request_body, timeout=self._client.time_out)
477 _logger.debug(f"Call REST API url '{response.url}'")
478
479 if response.status_code != STATUS_CODE_201:
480 log_response(response)
481 return None
482
483 except requests.exceptions.RequestException as e:
484 _logger.error(f"Error call REST API: {e}")
485 return None
486
487 content = response.content.decode("utf-8")
488 return json.loads(content)
489
490 # POST /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}/invoke
491 def invoke_operation_submodel_repo(self, submodel_identifier: str, id_short_path: str, request_body: dict, async_: str = "async") -> dict | None:
492 """Synchronously invokes an Operation at a specified path.
493
494 :param submodel_identifier: The Submodels unique id
495 :param id_short_path: IdShort path to the operation element (dot-separated)
496 :param request_body: Input parameters for the operation
497 :param async_: Determines whether an operation invocation is performed asynchronously or synchronously
498 :return: Operation result or None if an error occurred
499 """
500 if not self._client.encoded_ids:
501 submodel_identifier = encode_base_64(submodel_identifier)
503 url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements/{id_short_path}/invoke"
504
505 self._client.set_token()
506
507 try:
508 response = self._session.post(url, json=request_body, timeout=self._client.time_out)
509 _logger.debug(f"Call REST API url '{response.url}'")
510
511 if response.status_code != STATUS_CODE_200:
512 log_response(response)
513 return None
514
515 except requests.exceptions.RequestException as e:
516 _logger.error(f"Error call REST API: {e}")
517 return None
518
519 content = response.content.decode("utf-8")
520 return json.loads(content)
521
522 # GET /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}/$value
523 def get_submodel_element_by_path_value_only_submodel_repo(self, submodel_identifier: str, id_short_path: str) -> str | None:
524 """Retrieves the value of a specific SubmodelElement.
525
526 :param submodel_identifier: The Submodels unique id
527 :param id_short_path: IdShort path to the submodel element (dot-separated)
528 :return: Submodel element value or None if an error occurred
529 """
530 if not self._client.encoded_ids:
531 submodel_identifier = encode_base_64(submodel_identifier)
533 url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements/{id_short_path}/$value"
534
535 self._client.set_token()
536
537 try:
538 response = self._session.get(url, timeout=self._client.time_out)
539 _logger.debug(f"Call REST API url '{response.url}'")
540
541 if response.status_code != STATUS_CODE_200:
542 log_response(response)
543 return None
544
545 except requests.exceptions.RequestException as e:
546 _logger.error(f"Error call REST API: {e}")
547 return None
548
549 content = response.content.decode("utf-8")
550 return json.loads(content)
551
552 # PATCH /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}/$value
554 self, submodel_identifier: str, id_short_path: str, value: str, level: str = ""
555 ) -> bool:
556 """Updates the value of an existing SubmodelElement.
557
558 :param submodel_identifier: The Submodels unique id
559 :param id_short_path: IdShort path to the submodel element (dot-separated)
560 :param value: Submodel element value to update as string
561 :param level: Determines the structural depth of the respective resource content. Available values : deep, core
562 :return: True if the patch was successful, False otherwise
563 """
564 if not self._client.encoded_ids:
565 submodel_identifier = encode_base_64(submodel_identifier)
566
567 url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements/{id_short_path}/$value"
568
569 params: dict[str, str] = {}
570 if level:
571 params["level"] = level
572
573 self._client.set_token()
574
575 try:
576 response = self._session.patch(url, json=value, params=params, timeout=self._client.time_out)
577 _logger.debug(f"Call REST API url '{response.url}'")
578
579 if response.status_code == STATUS_CODE_404:
580 if self._client.get_log_intensity() == LogIntensity.HIGH:
581 _logger.warning(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
582 elif self._client.get_log_intensity() == LogIntensity.STANDARD:
583 _logger.debug(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
584
585 _logger.debug(response.text)
586 return False
587
588 if response.status_code != STATUS_CODE_204:
589 log_response(response)
590 return False
591
592 except requests.exceptions.RequestException as e:
593 _logger.error(f"Error call REST API: {e}")
594 return False
595
596 return True
597
598 # GET /submodels/{submodelIdentifier}/$value
599 def get_submodel_by_id_value_only(self, submodel_identifier: str, level: str = "", extent: str = "") -> dict | None:
600 """Returns a specific Submodel in the ValueOnly representation.
601
602 :param submodel_identifier: The Submodels unique id
603 :param level: Determines the structural depth of the respective resource content. Available values : deep, core
604 :param extent: Determines to which extent the resource is being serialized. Available values : withBlobValue, withoutBlobValue
605 :return: Submodel value as dict or None if an error occurred
606 """
607 params: dict[str, str] = {}
608 if level:
609 params["level"] = level
610 if extent:
611 params["extent"] = extent
612
613 if not self._client.encoded_ids:
614 submodel_identifier = encode_base_64(submodel_identifier)
615
616 url = f"{self._client.base_url}/submodels/{submodel_identifier}/$value"
617
618 self._client.set_token()
619
620 try:
621 response = self._session.get(url, params=params, timeout=self._client.time_out)
622 _logger.debug(f"Call REST API url '{response.url}'")
623
624 if response.status_code == STATUS_CODE_404:
625 if self._client.get_log_intensity() == LogIntensity.HIGH:
626 _logger.warning(f"Submodel with id '{submodel_identifier}' not found.")
627 elif self._client.get_log_intensity() == LogIntensity.STANDARD:
628 _logger.debug(f"Submodel with id '{submodel_identifier}' not found.")
629
630 _logger.debug(response.text)
631 return None
632
633 if response.status_code != STATUS_CODE_200:
634 log_response(response)
635 return None
636
637 except requests.exceptions.RequestException as e:
638 _logger.error(f"Error call REST API: {e}")
639 return None
640
641 content = response.content.decode("utf-8")
642 return json.loads(content)
643
644 # PATCH /submodels/{submodelIdentifier}/$value
645 def patch_submodel_by_id_value_only(self, submodel_identifier: str, request_body: dict, level: str = "") -> bool:
646 """Updates the values of an existing Submodel.
647
648 :param submodel_identifier: The Submodels unique id
649 :param request_body: Submodel values to update as dict
650 :param level: Determines the structural depth of the respective resource content. Available values : deep, core
651 :return: True if the patch was successful, False otherwise
652 """
653 params: dict[str, str] = {}
654 if level:
655 params["level"] = level
656
657 if not self._client.encoded_ids:
658 submodel_identifier = encode_base_64(submodel_identifier)
659
660 url = f"{self._client.base_url}/submodels/{submodel_identifier}/$value"
661
662 self._client.set_token()
663
664 try:
665 response = self._session.patch(url, json=request_body, params=params, timeout=self._client.time_out)
666 _logger.debug(f"Call REST API url '{response.url}'")
667
668 if response.status_code == STATUS_CODE_404:
669 if self._client.get_log_intensity() == LogIntensity.HIGH:
670 _logger.warning(f"Submodel with id '{submodel_identifier}' not found.")
671 elif self._client.get_log_intensity() == LogIntensity.STANDARD:
672 _logger.debug(f"Submodel with id '{submodel_identifier}' not found.")
673
674 _logger.debug(response.text)
675 return False
676
677 if response.status_code != STATUS_CODE_204:
678 log_response(response)
679 return False
680
681 except requests.exceptions.RequestException as e:
682 _logger.error(f"Error call REST API: {e}")
683 return False
684
685 return True
686
687 # GET /submodels/{submodelIdentifier}/$metadata
688 def get_submodel_by_id_metadata(self, submodel_identifier: str, level: str = "") -> dict | None:
689 """Returns the metadata attributes of a specific Submodel.
690
691 :param submodel_identifier: The Submodels unique id
692 :param level: Determines the structural depth of the respective resource content. Available values : deep, core
693 :return: Metadata attributes of the Submodel as dict or None if an error occurred
694 """
695 params: dict[str, str] = {}
696 if level:
697 params["level"] = level
698
699 if not self._client.encoded_ids:
700 submodel_identifier = encode_base_64(submodel_identifier)
701
702 url = f"{self._client.base_url}/submodels/{submodel_identifier}/$metadata"
703
704 self._client.set_token()
705
706 try:
707 response = self._session.get(url, params=params, timeout=self._client.time_out)
708 _logger.debug(f"Call REST API url '{response.url}'")
709
710 if response.status_code == STATUS_CODE_404:
711 if self._client.get_log_intensity() == LogIntensity.HIGH:
712 _logger.warning(f"Submodel with id '{submodel_identifier}' not found.")
713 elif self._client.get_log_intensity() == LogIntensity.STANDARD:
714 _logger.debug(f"Submodel with id '{submodel_identifier}' not found.")
715
716 _logger.debug(response.text)
717 return None
718
719 if response.status_code != STATUS_CODE_200:
720 log_response(response)
721 return None
722
723 except requests.exceptions.RequestException as e:
724 _logger.error(f"Error call REST API: {e}")
725 return None
726
727 content = response.content.decode("utf-8")
728 return json.loads(content)
729
730 # not supported by Java Server
731
732 # PATCH /submodels/{submodelIdentifier}
733 def patch_submodel_by_id(self, submodel_identifier: str, submodel_data: dict) -> bool:
734 """Updates an existing Submodel.
735
736 :param submodel_identifier: The Submodels unique id
737 :return: True if the patch was successful, False otherwise
738 """
739 if not self._client.encoded_ids:
740 submodel_identifier = encode_base_64(submodel_identifier)
742 url = f"{self._client.base_url}/submodels/{submodel_identifier}"
743
744 self._client.set_token()
745
746 try:
747 response = self._session.patch(url, json=submodel_data, timeout=self._client.time_out)
748 _logger.debug(f"Call REST API url '{response.url}'")
749
750 if response.status_code == STATUS_CODE_404:
751 if self._client.get_log_intensity() == LogIntensity.HIGH:
752 _logger.warning(f"Submodel with id '{submodel_identifier}' not found.")
753 elif self._client.get_log_intensity() == LogIntensity.STANDARD:
754 _logger.debug(f"Submodel with id '{submodel_identifier}' not found.")
755
756 _logger.debug(response.text)
757 return False
758
759 if response.status_code != STATUS_CODE_204:
760 log_response(response)
761 return False
762
763 except requests.exceptions.RequestException as e:
764 _logger.error(f"Error call REST API: {e}")
765 return False
766
767 return True
Implementation of Submodel related API calls.
dict|None get_submodel_by_id(self, str submodel_identifier, str level="", str extent="")
Returns a specific Submodel.
dict|None get_submodel_element_by_path_submodel_repo(self, str submodel_identifier, str id_short_path, str level="", str extent="")
Returns a specific submodel element from the Submodel at a specified path.
__init__(self, "AasHttpClient" client)
Initializes the SmImplementation with the given parameters.
str|None get_submodel_element_by_path_value_only_submodel_repo(self, str submodel_identifier, str id_short_path)
Retrieves the value of a specific SubmodelElement.
dict|None get_all_submodels(self, str semantic_id="", str id_short="", int limit=0, str cursor="", str level="", str extent="")
Returns all Submodels.
dict|None get_submodel_by_id_metadata(self, str submodel_identifier, str level="")
Returns the metadata attributes of a specific Submodel.
dict|None post_submodel(self, dict request_body)
Creates a new Submodel.
dict|None post_submodel_element_by_path_submodel_repo(self, str submodel_identifier, str id_short_path, dict request_body, str level="", str extent="")
Creates a new submodel element at a specified path within submodel elements hierarchy.
bool patch_submodel_element_by_path_value_only_submodel_repo(self, str submodel_identifier, str id_short_path, str value, str level="")
Updates the value of an existing SubmodelElement.
dict|None post_submodel_element_submodel_repo(self, str submodel_identifier, dict request_body)
Creates a new submodel element.
dict|None get_submodel_by_id_value_only(self, str submodel_identifier, str level="", str extent="")
Returns a specific Submodel in the ValueOnly representation.
bool put_submodel_element_by_path_submodel_repo(self, str submodel_identifier, str id_short_path, dict request_body, str level="")
Creates or updates an existing submodel element at a specified path within submodel elements hierarch...
bool patch_submodel_by_id(self, str submodel_identifier, dict submodel_data)
Updates an existing Submodel.
bool delete_submodel_by_id(self, str submodel_identifier)
Deletes a Submodel.
dict|None invoke_operation_submodel_repo(self, str submodel_identifier, str id_short_path, dict request_body, str async_="async")
Synchronously invokes an Operation at a specified path.
bool delete_submodel_element_by_path_submodel_repo(self, str submodel_identifier, str id_short_path)
Deletes a submodel element at a specified path within the submodel elements hierarchy.
bool put_submodels_by_id(self, str submodel_identifier, dict request_body)
Updates a existing Submodel.
bool patch_submodel_by_id_value_only(self, str submodel_identifier, dict request_body, str level="")
Updates the values of an existing Submodel.
dict|None get_all_submodel_elements_submodel_repository(self, str submodel_identifier, int limit=100, str cursor="", str level="", str extent="")
Returns all submodel elements including their hierarchy.