Skip to content

Latest commit

 

History

History
547 lines (404 loc) · 18.6 KB

File metadata and controls

547 lines (404 loc) · 18.6 KB

🚀 Getting Started

This guide will walk you through installing and using aas-http-client .


Installation

Prerequisites:

  • Python 3.10 or newer
  • Access to an AAS server endpoint

Install via pip:

pip install aas-http-client

For detailed configuration options, authentication methods and examples, see the Configuration Guide.


Usage

This chapter provides a practical, end-to-end overview of working with aas-http-client after installation. It starts with client and wrapper creation patterns, then walks through the most relevant repository and registry endpoint groups, including experimental attachment endpoints. Use the sections below as quick-start references and adapt the examples to your server setup, authentication mode, and ID encoding configuration.

For a complete list and detailed descriptions of all functions, see the API reference.

Notes

Most important notes before calling endpoint methods:

  • Logical groups map to attributes on the client instance:
    • client.shells -> Shell repository endpoints
    • client.submodels -> Submodel repository endpoints
    • client.shell_registry -> Shell descriptor registry endpoints
    • client.submodel_registry -> Submodel descriptor registry endpoints
    • client.experimental -> Experimental attachment endpoints
  • For the wrapper, most endpoint implementations are available directly on wrapper (SDK object responses). Registry endpoints are currently not exposed on wrapper.
  • encoded_ids behavior:
    • Keep encoded_ids aligned with your server expectations.
    • If encoded_ids is enabled (default), pass encoded IDs for endpoint path parameters when required by the API/server.
    • Example (encoded mode): submodel_identifier="dXJuOmV4YW1wbGU6c3VibW9kZWw6MDAx".
    • If encoded_ids is disabled, the client encodes IDs internally for endpoints that require encoded path segments.
    • Example (decoded mode): submodel_identifier="urn:example:submodel:001".
    • Inconsistent encoded_ids settings are a common reason for 404 responses.
  • Pagination:
    • List endpoints commonly support limit and cursor.
    • Start with a sensible limit (for example: 10 or 100), then continue with the returned cursor until no more results are available.
    • For client responses, read elements from result (for example: response.get("result", [])).
  • Return value conventions:
    • Read operations return dict, SDK objects, or None on failure/not found.
    • Write/delete operations usually return bool (True on success, False on failure).
    • Always handle None and False explicitly in production code.
  • Wrapper vs client:
    • Use client.* for raw dictionary responses and full registry coverage.
    • Use wrapper.* for SDK object-oriented handling.
    • Registry endpoints are currently not exposed on wrapper (use client.shell_registry and client.submodel_registry).
  • Experimental endpoints:
    • Attachment endpoints are experimental and may vary by server implementation.
    • Prefer feature checks or graceful fallbacks when targeting multiple server products/versions.

Creation Methods

There are three ways to create an AAS HTTP client or wrapper. Use either a client (dictionary-based API) or a wrapper (SDK object-based API), depending on your use case. Creation methods can return None if the configuration is invalid or the connection fails.

For production usage, avoid hardcoding secrets in source code. Load credentials from environment variables or a secret manager.

Create by URL

Create a client or wrapper by providing parameters directly using create_by_url :

import os

from aas_http_client.classes.client import aas_client
from aas_http_client.classes.wrapper import sdk_wrapper

client = aas_client.create_by_url(
    base_url="http://localhost:8080",
    basic_auth_username="admin",
    basic_auth_password=os.getenv("AAS_BASIC_AUTH_PASSWORD", ""),
    time_out=300,
    ssl_verify=True
)

wrapper = sdk_wrapper.create_by_url(
    base_url="http://localhost:8080",
    basic_auth_username="admin",
    basic_auth_password=os.getenv("AAS_BASIC_AUTH_PASSWORD", ""),
    time_out=300,
    ssl_verify=True
)

if client is None:
    raise RuntimeError("Client creation failed")

if wrapper is None:
    raise RuntimeError("Wrapper creation failed")

print("Client connectivity:", client.get_root() is not None)
print("Wrapper connectivity:", wrapper.get_client().get_root() is not None)

Create by Dictionary

Create a client or wrapper using a configuration dictionary with create_by_dict :

import os

from aas_http_client.classes.client import aas_client
from aas_http_client.classes.wrapper import sdk_wrapper

config = {
    "BaseUrl": "http://localhost:8080",
    "TimeOut": 300,
    "AuthenticationSettings": {
        "BasicAuth": {
            "Username": "admin"
        }
    }
}

client = aas_client.create_by_dict(
    configuration=config,
    basic_auth_password=os.getenv("AAS_BASIC_AUTH_PASSWORD", "")
)

wrapper = sdk_wrapper.create_by_dict(
    configuration=config,
    basic_auth_password=os.getenv("AAS_BASIC_AUTH_PASSWORD", "")
)

if client is None:
    raise RuntimeError("Client creation failed")

if wrapper is None:
    raise RuntimeError("Wrapper creation failed")

print("Client connectivity:", client.get_root() is not None)
print("Wrapper connectivity:", wrapper.get_client().get_root() is not None)

Create by Configuration File

Create a client or wrapper using a JSON configuration file with create_by_config :

import os

from pathlib import Path
from aas_http_client.classes.client import aas_client
from aas_http_client.classes.wrapper import sdk_wrapper

config_file = Path("config.json")
client = aas_client.create_by_config(
    config_file=config_file,
    basic_auth_password=os.getenv("AAS_BASIC_AUTH_PASSWORD", "")
)

wrapper = sdk_wrapper.create_by_config(
    config_file=config_file,
    basic_auth_password=os.getenv("AAS_BASIC_AUTH_PASSWORD", "")
)

if client is None:
    raise RuntimeError("Client creation failed")

if wrapper is None:
    raise RuntimeError("Wrapper creation failed")

print("Client connectivity:", client.get_root() is not None)
print("Wrapper connectivity:", wrapper.get_client().get_root() is not None)

/Shell/ Endpoints

This section shows how to work with the most common Shell repository operations after client or wrapper creation.

Most important points:

  • Shell endpoints are available via client.shells (dictionary responses) or directly on wrapper (SDK object responses).
  • List endpoints are paginated. Use limit and cursor when iterating through larger result sets.
  • Always handle None results to detect connectivity, authorization, or server-side issues.

For the full list of available methods and signatures, see:

Example: List Asset Administration Shells (client)

# Assumes `client` was created successfully in one of the sections above.
result = client.shells.get_all_asset_administration_shells(limit=10)

if result is None:
    raise RuntimeError("Failed to fetch shells")

shells = result.get("result", [])
print(f"Received {len(shells)} shell(s)")

Example: Fetch one shell by ID (wrapper)

# Assumes `wrapper` was created successfully in one of the sections above.
aas_id = "urn:example:aas:001"
aas = wrapper.get_asset_administration_shell_by_id(aas_id)

if aas is None:
    print("Shell not found or request failed")
else:
    print("Found shell with id:", aas.id)

/Submodel/ Endpoints

This section shows how to work with common Submodel repository operations after client or wrapper creation.

Most important points:

  • Submodel endpoints are available via client.submodels (dictionary responses) or directly on wrapper (SDK object responses).
  • List endpoints are paginated. Use limit and cursor when retrieving larger result sets.
  • level and extent can be used to control response depth and blob behavior.
  • Always handle None results to detect connectivity, authorization, or server-side issues.

For the full list of available methods and signatures, see:

Example: List Submodels (client)

# Assumes `client` was created successfully in one of the sections above.
result = client.submodels.get_all_submodels(limit=10)

if result is None:
    raise RuntimeError("Failed to fetch submodels")

submodels = result.get("result", [])
print(f"Received {len(submodels)} submodel(s)")

Example: Fetch one submodel by ID (wrapper)

# Assumes `wrapper` was created successfully in one of the sections above.
submodel_id = "urn:example:submodel:001"
submodel = wrapper.get_submodel_by_id(submodel_id)

if submodel is None:
    print("Submodel not found or request failed")
else:
    print("Found submodel with id:", submodel.id)

/shell-descriptors/ Endpoints

This section shows how to work with Asset Administration Shell registry endpoints.

Most important points:

  • Shell descriptor endpoints are available via client.shell_registry.
  • Registry endpoints currently are not exposed on wrapper.
  • List endpoints are paginated. Use limit and cursor when iterating through larger result sets.
  • Depending on server settings, IDs may need Base64 URL-safe encoding. Keep encoded_ids consistent with your server setup.

For the full list of available methods and signatures, see:

Example: List shell descriptors

# Assumes `client` was created successfully in one of the sections above.
result = client.shell_registry.get_all_asset_administration_shell_descriptors(limit=10)

if result is None:
    raise RuntimeError("Failed to fetch shell descriptors")

descriptors = result.get("result", [])
print(f"Received {len(descriptors)} shell descriptor(s)")

Example: Fetch one shell descriptor by ID

# Assumes `client` was created successfully in one of the sections above.
aas_id = "urn:example:aas:001"
descriptor = client.shell_registry.get_asset_administration_shell_descriptor_by_id(aas_id)

if descriptor is None:
    print("Shell descriptor not found or request failed")
else:
    print("Found descriptor with id:", descriptor.get("id"))

Example: Register a shell descriptor

# Assumes `client` was created successfully in one of the sections above.
payload = {
    "id": "urn:example:aas:001",
    "idShort": "aas_example_001",
    "assetKind": "Instance",
    "endpoints": [
        {
            "interface": "AAS-3.0",
            "protocolInformation": {
                "href": "http://localhost:8080/shells/urn:example:aas:001",
                "endpointProtocol": "http",
                "endpointProtocolVersion": ["1.1"],
                "subprotocol": "",
                "subprotocolBody": "",
                "subprotocolBodyEncoding": "plain",
                "securityAttributes": []
            }
        }
    ],
    "submodelDescriptors": []
}

created = client.shell_registry.post_asset_administration_shell_descriptor(payload)

if created is None:
    raise RuntimeError("Failed to create shell descriptor")

print("Registered descriptor:", created.get("id"))

/submodel-descriptors/ Endpoints

This section shows how to work with Submodel registry endpoints.

Most important points:

  • Submodel descriptor endpoints are available via client.submodel_registry.
  • Registry endpoints currently are not exposed on wrapper.
  • List endpoints are paginated. Use limit and cursor when iterating through larger result sets.

For the full list of available methods and signatures, see:

Example: List submodel descriptors

# Assumes `client` was created successfully in one of the sections above.
result = client.submodel_registry.get_all_submodel_descriptors(limit=10)

if result is None:
    raise RuntimeError("Failed to fetch submodel descriptors")

descriptors = result.get("result", [])
print(f"Received {len(descriptors)} submodel descriptor(s)")

Example: Fetch one submodel descriptor by ID

# Assumes `client` was created successfully in one of the sections above.
submodel_id = "urn:example:submodel:001"
descriptor = client.submodel_registry.get_submodel_descriptor_by_id(submodel_id)

if descriptor is None:
    print("Submodel descriptor not found or request failed")
else:
    print("Found descriptor with id:", descriptor.get("id"))

Example: Register a submodel descriptor

# Assumes `client` was created successfully in one of the sections above.
payload = {
    "id": "urn:example:submodel:001",
    "idShort": "sm_example_001",
    "endpoints": [
        {
            "interface": "SUBMODEL-3.0",
            "protocolInformation": {
                "href": "http://localhost:8080/submodels/urn:example:submodel:001",
                "endpointProtocol": "http",
                "endpointProtocolVersion": ["1.1"],
                "subprotocol": "",
                "subprotocolBody": "",
                "subprotocolBodyEncoding": "plain",
                "securityAttributes": []
            }
        }
    ],
    "semanticId": {
        "type": "ExternalReference",
        "keys": [
            {
                "type": "GlobalReference",
                "value": "urn:example:semantic:submodel:001"
            }
        ]
    }
}

created = client.submodel_registry.post_submodel_descriptor(payload)

if created is None:
    raise RuntimeError("Failed to create submodel descriptor")

print("Registered descriptor:", created.get("id"))

Experimental Endpoint Implementations

This section shows the experimental attachment endpoints for file content on Submodel File elements.

Most important points:

  • Experimental endpoints are available via client.experimental and via wrapper methods prefixed with experimental_.
  • Experimental endpoints may not be supported by every server implementation.

For the full list of available methods and signatures, see:

Example: Download file content (wrapper)

from pathlib import Path

# Assumes `wrapper` was created successfully in one of the sections above.
attachment = wrapper.experimental_get_file_by_path_submodel_repo(
    submodel_identifier="urn:example:submodel:001",
    id_short_path="Documents.UserManual"
)

if attachment is None:
    print("No attachment found or request failed")
else:
    output = Path("downloaded_" + attachment.filename)
    output.write_bytes(attachment.content)
    print("Downloaded attachment to:", output)
    print("Detected content type:", attachment.content_type)

Example: Upload or replace file content (client)

from pathlib import Path

# Assumes `client` was created successfully in one of the sections above.
submodel_id = "urn:example:submodel:001"
id_short_path = "Documents.UserManual"
file_path = Path("manual.pdf")

created = client.experimental.post_file_by_path_submodel_repo(
    submodel_identifier=submodel_id,
    id_short_path=id_short_path,
    file=file_path
)

if not created:
    raise RuntimeError("Failed to upload attachment with POST")

updated = client.experimental.put_file_by_path_submodel_repo(
    submodel_identifier=submodel_id,
    id_short_path=id_short_path,
    file=file_path
)

if not updated:
    raise RuntimeError("Failed to replace attachment with PUT")

print("Attachment upload/update successful")

Generic Endpoint Implementations

This section covers the low-level generic endpoint helpers on AasHttpClient .

Most important points:

  • Generic endpoint helpers are available on client only (not on wrapper).
  • They allow direct HTTP calls to arbitrary endpoint URLs when no specialized helper exists.

Example: Generic GET call

# Assumes `client` was created successfully in one of the sections above.
url = f"{client.base_url}/description"
result = client.get_endpoint(url)

if result is None:
    print("GET request failed or returned non-200")
else:
    print("Service description keys:", list(result.keys()))

Example: Generic POST/PUT/PATCH/DELETE calls

# Assumes `client` was created successfully in one of the sections above.
url = f"{client.base_url}/some-custom-endpoint"
payload = {"example": "value"}

post_error = client.post_endpoint(url, payload)
if post_error is not None:
    print("POST returned non-success:", post_error)

put_error = client.put_endpoint(url, payload)
if put_error is not None:
    print("PUT returned non-success:", put_error)

patch_error = client.patch_endpoint(url, {"example": "updated"})
if patch_error is not None:
    print("PATCH returned non-success:", patch_error)

delete_error = client.delete_endpoint(url)
if delete_error is not None:
    print("DELETE returned non-success:", delete_error)

For the full list of available methods and signatures, see: