API Reference

class AbstractDoipClient(options: Options | None = None, **kwargs: Unpack[OptionsTypedDict])

Bases: DoipClient

This class implements the DoipClient interface and provides common functionality for all DOIP client implementations.

Concrete implementations need to provide the perform_operation_via_request() method to handle the actual communication with DOIP services, as well as a close() method to handle resource cleanup. This class will define all convenience methods by means of perform_operation_via_request().

default_options: Options | None

Default options for all operations performed by this client.

These options can be overridden by providing options to individual method calls.

abstractmethod perform_operation_via_request(request: DoipRequest, options: Options | None = None) → DoipResponse

This method must be implemented by concrete subclasses to handle the actual communication with DOIP services. Users of subclasses of DoipClient should prefer DoipClient.perform_operation() or the various convenience methods of DoipClient.

Parameters:

request – The DOIP request to perform
options – Optional request options for this operation

Returns:

A promise that resolves to a DoipResponse

perform_operation(request: DoipRequest, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → DoipResponse

Performs a DOIP operation on the target object.

This method can be called by either sending a DoipRequest object as a parameter or by providing the target_id, operation_id, and (optional) input arguments to the method call.

Parameters:

request – The DOIP request to perform
target_id – The ID of the target digital object or service
operation_id – The ID of the operation to perform
input – Optional input for the operation
options – Optional request options for this operation
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

A DoipResponse. The caller is responsible for closing the DoipResponse.

Raises:

DoipException – raised if the operation fails for any reason

Examples:

request = DoipRequest("service", "0.DOIP/Op.Hello")
with client.perform_operation(request) as response:
    print(response.as_json())

with client.perform_operation("service", "0.DOIP/Op.Hello") as response:
    print(response.as_json())

abstractmethod close() → None: Closes the client and releases any resources.

target_id_for_service(options: Options | None = None) → str

Determines the ID for a service, based on the options given. This is the identifier of the service as a digital object, used as the target ID for a request which targets the service itself.

If options does not specify a service ID, there may be a client default. Defaults to service if no other info is specified.

Parameters:: options – Optional request options
Returns:: The ID for the service

augmented_request(request: DoipRequest, options: Options | None = None) → DoipRequest

This method creates a copy of the request and adds authentication information and attributes from the options or default options if not already present.

Parameters:

request – The DOIP request to augment
options – Optional request options

Returns:

The augmented DOIP request

hello(options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → DigitalObject

Sends a hello request to the service to get information about the service and verify connectivity.

Parameters:

options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

A digital object containing information about the service

Raises:

DoipException – raised if the operation fails for any reason

Example:

response = client.hello()
print(response)

retrieve(target_id: str, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → DigitalObject | None

Retrieves a digital object by ID.

Parameters:

target_id – ID of the digital object to retrieve
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

The retrieved digital object, or None if not found

Raises:

DoipException – raised if the operation fails for any reason

Example:

dobj = client.retrieve('20.500.123/1')
if dobj is not None:
    print(dobj)

delete(target_id: str, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → None

Deletes a digital object by ID.

Parameters:

target_id – The ID of the digital object to delete
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Raises:

DoipException – raised if the operation fails for any reason

Example:

client.delete('20.500.123/1')

list_operations(target_id: str, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → list[str]

Retrieves a list of operations that can be performed on the specified target.

Parameters:

target_id – The ID of the target digital object or service
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

A list of operation IDs

Raises:

DoipException – raised if the operation fails for any reason

Example:

# List operations for a specific object
operations = client.list_operations('20.500.123/1')
print(operations)

# List operations for the service using the default service ID
service_ops = client.list_operations('service')
print(service_ops)

search(query: str, params: QueryParams | None = None, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → SearchResults[DigitalObject]

Searches for digital objects matching the specified query and returns the complete digital objects.

Parameters:

query – The search query
params – Optional query parameters
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

A SearchResults which among other properties is an iterator of digital objects. The caller is responsible for closing the SearchResults.

Raises:

DoipException – raised if the operation fails for any reason

Example:

with client.search('type:Document') as results:
    print(f'Number of results: {results.size}')
    for dobj in results:
        print(f'{dobj.id}: {dobj.attributes["content"]["name"]}')

search_ids(query: str, params: QueryParams | None = None, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → SearchResults[str]

Searches for digital objects matching the specified query and returns only their IDs.

Parameters:

query – The search query
params – Optional query parameters
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

A SearchResults which among other properties is an iterator of digital object IDs. The caller is responsible for closing the SearchResults.

Raises:

DoipException – raised if the operation fails for any reason

Example:

with client.search_ids("type:Document") as results:
    print(f"Number of results: {results.size}")
    for id in results:
        print(id)

retrieve_element_bytes(target_id: str, element_id: str, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → BufferedIOBase

Retrieves the body of a specific element of a digital object by its ID.

Parameters:

target_id – The ID of the digital object
element_id – The ID of the element to retrieve
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

The element body

Raises:

DoipException – raised if the operation fails for any reason

Examples:

# Get the bytes of a binary element payload
element_body = client.retrieve_element_bytes('20.500.123/1', 'attachment-binary')
all_bytes = element_body.read()
print(f'length: {len(all_bytes)}')

# Get text from a text element body
element_body = client.retrieve_element_bytes('20.500.123/1', 'attachment-text')
all_bytes = element_body.read()
print(all_bytes.decode("utf-8"))

retrieve_partial_element_bytes(target_id: str, element_id: str, start: int | None = None, end: int | None = None, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → BufferedIOBase

Retrieves the body of a specific element of a digital object by its ID.

Parameters:

target_id – The ID of the digital object
element_id – The ID of the element to retrieve
start – The starting byte position (inclusive)
end – The ending byte position (inclusive)
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

The partial element body

Raises:

DoipException – raised if the operation fails for any reason

Examples:

# Retrieve the first 1000 bytes of a binary element body
element_body = client.retrieve_partial_element_bytes('20.500.123/1', 'attachment-binary', 0, 999)
all_bytes = element_body.read()
print(f'length: {len(all_bytes)}')

# Retrieve bytes from the middle of a text element body
element_body = client.retrieve_partial_element_bytes('20.500.123/1', 'attachment-text', 2, 5)
all_bytes = element_body.read()
print(all_bytes.decode("utf-8"))

create(dobj: DigitalObject, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → DigitalObject

Creates a new digital object.

Parameters:

dobj – The digital object to create
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

The created digital object

Raises:

DoipException – raised if the operation fails for any reason

Example:

dobj = DigitalObject(
    type='Document',
    attributes={
        'content': {
            'name': 'Example Document'
        }
    }
)

created_object = client.create(dobj)
print(created_object.id)

update(dobj: DigitalObject, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → DigitalObject

Updates an existing digital object.

Parameters:

dobj – The digital object to update
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

The updated digital object

Raises:

DoipException – raised if the operation fails for any reason

Example:

dobj = client.retrieve('20.500.123/1')
if dobj is not None:
    dobj.attributes['content']['name'] = 'Updated Example Document'
    updated_object = client.update(dobj)
    print(updated_object.attributes['content']['name'])

static digital_object_from_response(response: DoipResponse) → DigitalObject

Parses a DoipResponse into a DigitalObject, throwing an error if the DoipResponse does not have the expected structure.

Parameters:: response – The DOIP response to parse
Returns:: The DigitalObject given by the DOIP response

static search_results_from_response(response: DoipResponse) → SearchResults

Parses a DoipResponse into a SearchResults, throwing an error if the DoipResponse does not have the expected structure.

Parameters:: response – The DOIP response to parse
Returns:: The SearchResults given by the DOIP response

class AuthenticationInfo

Bases: ABC

Used by a DoipRequest to authenticate to a DOIP service.

See: PasswordAuthenticationInfo, PrivateKeyAuthenticationInfo, TokenAuthenticationInfo.

abstractmethod get_client_id() → str | None: Returns the clientId property to be used in DOIP requests made with this AuthenticationInfo.

abstractmethod get_authentication() → Json | None: Returns the authentication property to be used in DOIP requests made with this AuthenticationInfo.

class Bucket

Bases: TypedDict

Type representing a facet results bucket.

value: str: The facet value for this bucket.

count: int: The number of items that have this facet value.

filter_query: str: A query string that can be used to filter to just items with this facet value.

class DigitalObject(id: str | None = None, type: str | None = None, attributes: dict[str, Json] | None = None, elements: list[Element] | None = None)

Bases: Mapping

A digital object is the fundamental entity in DOIP, consisting of an identifier, type, attributes, and optional binary elements.

Example:

dobj = DigitalObject(
    id="20.500.123/1",
    type="Document",
    attributes={
        "content": {
            "name": "Example Document",
            "description": "This is an example digital object"
        }
    }
)

id: str | None = None

The unique identifier of the digital object.

This is generally optional when creating a new object (the service will assign an ID), but will be present when retrieving an existing object.

type: str | None = None

The type of the digital object.

This typically corresponds to a schema that defines the structure of the object’s attributes.

attributes: dict[str, Json] | None = None

The attributes (metadata) of the digital object.

This is a flexible record that can contain any JSON-serializable data.

elements: list[Element] | None = None

Optional binary elements associated with the digital object.

These can represent files, attachments, or other binary data.

to_dict() → dict[str, Any]

Converts a DigitalObject into a dict, for use in JSON APIs.

Returns:: The DigitalObject as a dict

static from_dict(input: dict[str, Any]) → DigitalObject

Converts a dict, such as parsed from JSON, into a DigitalObject.

Parameters:: input – The dict
Returns:: The DigitalObject represented by the dict

get_attribute(attr: str) → Json | None

Gets an attribute from a DigitalObject.

Parameters:: attr – The name of the attribute to retrieve
Returns:: The value of the attribute, or None if missing

set_attribute(attr: str, value: Json) → None

Sets a top-level attribute on the DigitalObject.

Parameters:

attr – The name of the attribute to set
value – The value to set the attribute to

get_element_by_id(id: str) → Element | None

Gets an Element from a DigitalObject by the element id.

Parameters:: id – The identifier of the Element
Returns:: The Element, or None if missing

class DoipClient(*args, **kwargs)

Bases: Protocol

Interface for a client used to communicate with a DOIP service.

The client handles operations like retrieving, creating, updating, and deleting digital objects, as well as searching and other DOIP operations. All operations are accessible through perform_operation(); other methods are provided for convenience.

perform_operation(request: DoipRequest, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → DoipResponse

Performs a DOIP operation on the target object.

This method can be called by either sending a DoipRequest object as a parameter or by providing the target_id, operation_id, and (optional) input arguments to the method call.

Parameters:

request – The DOIP request to perform
target_id – The ID of the target digital object or service
operation_id – The ID of the operation to perform
input – Optional input for the operation
options – Optional request options for this operation
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

A DoipResponse. The caller is responsible for closing the DoipResponse.

Raises:

DoipException – raised if the operation fails for any reason

Examples:

request = DoipRequest("service", "0.DOIP/Op.Hello")
with client.perform_operation(request) as response:
    print(response.as_json())

with client.perform_operation("service", "0.DOIP/Op.Hello") as response:
    print(response.as_json())

close() → None: Closes the client and releases any resources.

target_id_for_service(options: Options | None = None) → str

Determines the ID for a service, based on the options given. This is the identifier of the service as a digital object, used as the target ID for a request which targets the service itself.

If options does not specify a service ID, there may be a client default. Defaults to service if no other info is specified.

Parameters:: options – Optional request options
Returns:: The ID for the service

hello(options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → DigitalObject

Sends a hello request to the service to get information about the service and verify connectivity.

Parameters:

options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

A digital object containing information about the service

Raises:

DoipException – raised if the operation fails for any reason

Example:

response = client.hello()
print(response)

retrieve(target_id: str, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → DigitalObject | None

Retrieves a digital object by ID.

Parameters:

target_id – ID of the digital object to retrieve
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

The retrieved digital object, or None if not found

Raises:

DoipException – raised if the operation fails for any reason

Example:

dobj = client.retrieve('20.500.123/1')
if dobj is not None:
    print(dobj)

delete(target_id: str, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → None

Deletes a digital object by ID.

Parameters:

target_id – The ID of the digital object to delete
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Raises:

DoipException – raised if the operation fails for any reason

Example:

client.delete('20.500.123/1')

list_operations(target_id: str, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → list[str]

Retrieves a list of operations that can be performed on the specified target.

Parameters:

target_id – The ID of the target digital object or service
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

A list of operation IDs

Raises:

DoipException – raised if the operation fails for any reason

Example:

# List operations for a specific object
operations = client.list_operations('20.500.123/1')
print(operations)

# List operations for the service using the default service ID
service_ops = client.list_operations('service')
print(service_ops)

search(query: str, params: QueryParams | None = None, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → SearchResults[DigitalObject]

Searches for digital objects matching the specified query and returns the complete digital objects.

Parameters:

query – The search query
params – Optional query parameters
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

A SearchResults which among other properties is an iterator of digital objects. The caller is responsible for closing the SearchResults.

Raises:

DoipException – raised if the operation fails for any reason

Example:

with client.search('type:Document') as results:
    print(f'Number of results: {results.size}')
    for dobj in results:
        print(f'{dobj.id}: {dobj.attributes["content"]["name"]}')

search_ids(query: str, params: QueryParams | None = None, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → SearchResults[str]

Searches for digital objects matching the specified query and returns only their IDs.

Parameters:

query – The search query
params – Optional query parameters
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

A SearchResults which among other properties is an iterator of digital object IDs. The caller is responsible for closing the SearchResults.

Raises:

DoipException – raised if the operation fails for any reason

Example:

with client.search_ids("type:Document") as results:
    print(f"Number of results: {results.size}")
    for id in results:
        print(id)

retrieve_element_bytes(target_id: str, element_id: str, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → BufferedIOBase

Retrieves the body of a specific element of a digital object by its ID.

Parameters:

target_id – The ID of the digital object
element_id – The ID of the element to retrieve
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

The element body

Raises:

DoipException – raised if the operation fails for any reason

Examples:

# Get the bytes of a binary element payload
element_body = client.retrieve_element_bytes('20.500.123/1', 'attachment-binary')
all_bytes = element_body.read()
print(f'length: {len(all_bytes)}')

# Get text from a text element body
element_body = client.retrieve_element_bytes('20.500.123/1', 'attachment-text')
all_bytes = element_body.read()
print(all_bytes.decode("utf-8"))

retrieve_partial_element_bytes(target_id: str, element_id: str, start: int | None = None, end: int | None = None, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → BufferedIOBase

Retrieves the body of a specific element of a digital object by its ID.

Parameters:

target_id – The ID of the digital object
element_id – The ID of the element to retrieve
start – The starting byte position (inclusive)
end – The ending byte position (inclusive)
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

The partial element body

Raises:

DoipException – raised if the operation fails for any reason

Examples:

# Retrieve the first 1000 bytes of a binary element body
element_body = client.retrieve_partial_element_bytes('20.500.123/1', 'attachment-binary', 0, 999)
all_bytes = element_body.read()
print(f'length: {len(all_bytes)}')

# Retrieve bytes from the middle of a text element body
element_body = client.retrieve_partial_element_bytes('20.500.123/1', 'attachment-text', 2, 5)
all_bytes = element_body.read()
print(all_bytes.decode("utf-8"))

create(dobj: DigitalObject, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → DigitalObject

Creates a new digital object.

Parameters:

dobj – The digital object to create
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

The created digital object

Raises:

DoipException – raised if the operation fails for any reason

Example:

dobj = DigitalObject(
    type='Document',
    attributes={
        'content': {
            'name': 'Example Document'
        }
    }
)

created_object = client.create(dobj)
print(created_object.id)

update(dobj: DigitalObject, options: Options | None = None, **kwargs: Unpack[OptionsTypedDict]) → DigitalObject

Updates an existing digital object.

Parameters:

dobj – The digital object to update
options – Optional request options
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

Returns:

The updated digital object

Raises:

DoipException – raised if the operation fails for any reason

Example:

dobj = client.retrieve('20.500.123/1')
if dobj is not None:
    dobj.attributes['content']['name'] = 'Updated Example Document'
    updated_object = client.update(dobj)
    print(updated_object.attributes['content']['name'])

class DoipConstants

Bases: object

Constants representing DOIP statuses and default operations

STATUS_OK = '0.DOIP/Status.001': OK - The operation was successfully processed.

STATUS_BAD_REQUEST = '0.DOIP/Status.101': Bad Request - The request was invalid in some way.

STATUS_UNAUTHENTICATED = '0.DOIP/Status.102': Unauthenticated - The client did not successfully authenticate.

STATUS_FORBIDDEN = '0.DOIP/Status.103': Forbidden - The client successfully authenticated, but is unauthorized to invoke the operation.

STATUS_NOT_FOUND = '0.DOIP/Status.104': Not Found - The digital object is not known to the service to exist.

STATUS_CONFLICT = '0.DOIP/Status.105': Conflict - The client tried to create a new digital object with an identifier already in use by an existing digital object.

STATUS_DECLINED = '0.DOIP/Status.200': Declined - The service declines to execute the extended operation.

STATUS_ERROR = '0.DOIP/Status.500': System Error - Error other than the ones stated above occurred.

OP_HELLO = '0.DOIP/Op.Hello'

This operation allows a client to get information about the DOIP service.

Request attributes: none
Input: empty
Response attributes: none
Output: the default serialization of the DOIP Service Information as a digital object.

OP_LIST_OPERATIONS = '0.DOIP/Op.ListOperations'

This operation requests the list of operations that can be invoked on the target object.

Request attributes: none
Input: none
Response attributes: none
Output: a serialized list of strings based on the default serialization, each of which is an operation id that the target object supports.

OP_CREATE = '0.DOIP/Op.Create'

This operation creates a digital object within the DOIP service. The target of a creation operation is the DOIP service itself.

Request attributes: none
Input: a serialized digital object. The “id” can be omitted to ask the DOIP service to automatically choose the id.
Response attributes: none
Output: the default serialization of the created object omitting element data. Notably, includes the identifier of the object (even if chosen by the client) and any changes to the object automatically performed by the DOIP service.

OP_RETRIEVE = '0.DOIP/Op.Retrieve'

This operation retrieves (some parts of the) information represented by the target object.

Request attributes:
- element: if specified, retrieves the data for that element
- includeElementData: if present, returns the serialization of the object including all element data
Input: none
Response attributes: none
Output: the default output is the serialization of the object using the default serialization without element data. If element was specified, the output is a single bytes segment with the bytes of the specified element. If includeElementData was specified, the output is the full serialized digital object.

OP_UPDATE = '0.DOIP/Op.Update'

This operation updates (some parts of the) information represented by the target object.

Request attributes: none
Input: a serialized digital object. Elements which are not intended to be changed can be omitted from the input.
Response attributes: none
Output: the default serialization of the created object omitting element data. Notably, includes any changes to the object automatically performed by the DOIP service.

OP_DELETE = '0.DOIP/Op.Delete'

This operation removes the target digital object from the management of the DOIP service.

Request attributes: none
Input: none
Response attributes: none
Output: none

OP_SEARCH = '0.DOIP/Op.Search'

This operation is used to discover digital objects by searching metadata contained in the set of digital objects managed by the DOIP service.

Request attributes:
- query: the search query to be performed, in a textual representation
- pageNum: the page number to be returned, starting with 0
- pageSize: the page size to be returned; if missing or negative, all results will be returned; if zero, no results are returned, but the size is still returned
- sortFields: a comma-separated list of sort specifications, each of which is a field name optionally followed by ASC or DESC
- type: either “id”, to return just object ids, or “full”, to return full object data (omitting element data); defaults to “full”
Input: none
Response attributes: none
Output: an object based on the default serialization with top-level properties:
- size: the number of results across all pages
- results: a list of results, each of which is either a string (the object id) or the default serialization of an object omitting element data.

exception DoipException(status: str, response: Json)

Bases: Exception

Represents a DOIP error. This is an extension of the native Python Exception class.

status: The DOIP Status code (for example, ‘0.DOIP/Status.200’)

response

The full response message. Generally a JSON object containing a detailed message.

Example:

{ message: 'Operation not supported' }

Bases: object

A DOIP request is sent to a DOIP service to perform an operation on a digital object or service.

target_id: str: Identifier of the target digital object or service. This is the entity on which the operation will be performed.

operation_id: str: Identifier of the operation to perform.

request_id: str | None = None: Identifier for the request. If not provided, the client implementation will generate one if needed.

client_id: str | None = None: Identifier for the client making the request. If authentication is provided as an AuthenticationInfo, that can provide the client_id instead of it being provided directly here.

authentication: AuthenticationInfo | Json | None = None: Authentication information for the request. This can be arbitrary JSON, though it is frequently an AuthenticationInfo.

attributes: Json | None = None: Additional parameters for the request. The exact properties depend on the type of request.

Input data for the request.

The type of the input is very flexible. InDoipMessage provides the general case of a multi-segment DOIP message, but it is generally preferable to provide arbitrary JSON directly, or bytes as a BufferedIOBase, or str, bytes, bytearray, or file-like object.

class DoipResponse(initial_segment: InDoipSegment, message: InDoipMessage)

Bases: object

Represents a response from a DOIP operation.

This class encapsulates the response data from a DOIP service, including status, attributes, and output. It provides methods to access this data and convert the response to different formats (JSON, BufferedIOBase, or Exception).

Callers are generally responsible for calling close() to ensure the network connection behind this response is closed. The as_json(), as_byte_stream(), and as_exception() methods automatically close the response.

get_status() → str: Gets the DOIP status of the response. For example, 0.DOIP/Status.001

get_attributes() → dict[str, Any]

Gets all attributes from the response. The attributes that are available depend on the operation that was invoked.

Returns:: The JSON object representing the attributes for this response

get_attribute(attr: str) → Json | None

Gets a specific attribute from the response.

Parameters:: attr – The name of the attribute to retrieve
Returns:: The JSON associated with the given attribute, or None if not found

get_output() → InDoipMessage

Gets DOIP output as an InDoipMessage.

This is a lower-level api for advanced usage. In most cases, as_json() and as_byte_stream() should be preferred.

as_exception() → DoipException

Gets the response as a DoipException object. This allows easy access to any error information in the response.

Returns:: The resulting DoipException

as_json() → Json: Gets the response as a JSON object

as_byte_stream() → BufferedIOBase: Gets the response as a stream of bytes

close() → None: Close the response object and clean up resources, if needed

Bases: Mapping

Elements are used to store binary data (like files or attachments) associated with a digital object. Each element has an identifier and can optionally include metadata about the content. When making a request, ElementWithBody can be used to provide the element bytes using other Python types.

id: str | None: The identifier of the element within the digital object. This is used to reference the element when retrieving it.

type: str | None = None: The media type of the element’s content. For example, “application/pdf” or “image/jpeg”.

length: int | None = None: The length of the element’s content in bytes.

attributes: dict[str, Json] | None = None: Additional metadata about the element. This can include information like filename.

body: BufferedIOBase | None = None: The binary content of the element. This is typically present when retrieving an element or when creating/updating an element with streaming content.

to_dict() → dict[str, Any]

Converts an Element into a dict, for use in JSON APIs.

Returns:: The Element as a dict

classmethod from_dict(input: dict[str, Any]) → Self

Converts a dict, such as parsed from JSON, into an Element.

Parameters:: input – The dict
Returns:: The Element represented by the dict

get_attribute(attr: str) → Json | None

Gets an attribute from an Element.

Parameters:: attr – The name of the attribute to retrieve
Returns:: The value of the attribute, or None if missing

set_attribute(attr: str, value: Json) → None

Sets a top-level attribute on the Element.

Parameters:

attr – The name of the attribute to set
value – The value to set the attribute to

Bases: Element

This helper class can be used to create a new element with a body to attach to a Digital Object. The body can be provided using a number of different types besides the BufferedIOBase required by Element, include str, bytes, bytearray, and file-like objects.

class Facet(field: str)

Bases: object

Class representing a facet to use in a search request.

field: str: The field name to facet on.

class FacetResult

Bases: TypedDict

Type representing a facet in search results.

field: str: The field name that was faceted on.

buckets: list[Bucket]: The buckets of values found for this facet.

class InDoipMessage

Bases: Iterator[InDoipSegment], ABC

Class representing a DOIP message as a series of InDoipSegment objects, which is how messages are sent in DOIP. When the input of a DoipRequest or the output of a DoipResponse has multiple segments, an InDoipMessage can be used.

For advanced usage. In most cases, the input of DoipRequest can be provided, and the output of DoipResponse can be accessed, more simply as JSON or a byte stream.

abstractmethod close() → None

static get_first_segment(m: InDoipMessage) → InDoipSegment | None

static is_empty(m: InDoipMessage) → bool

class InDoipMessageFromIterable(iterable: Iterable[InDoipSegment])

Bases: InDoipMessage

A convenience class for creating an InDoipMessage from an iterable of InDoipSegment.

For advanced usage. In most cases, the input of DoipRequest can be provided, and the output of DoipResponse can be accessed, more simply as JSON or a byte stream.

close() → None

class InDoipSegment(is_json: bool | None = None, *, json: Json | None = None, body: BufferedIOBase | None = None)

Bases: object

Represents a segment of an InDoipMessage. Each segment may consist of JSON or arbitrary bytes.

For advanced usage. In most cases, the input of DoipRequest can be provided, and the output of DoipResponse can be accessed, more simply as JSON or a byte stream.

is_json: bool

static of_json(is_json: bool | None = None, json: Json | None = None) → InDoipSegment

static of_bytes(is_json: bool | None = None, body: BufferedIOBase | None = None) → InDoipSegment

json() → Json

property body: BufferedIOBase

close() → None

class Json: A type alias representing arbitrary JSON values (dicts, lists, strings, numbers, booleans, or None). The actual type used is Any, but Json appears in the generated documentation.

Bases: object

Class defining request options for DOIP client operations.

service_info: ServiceInfo | None = None

Information used to connect to a DOIP service to use for the operation.

Note: Service information provided via the options parameter of a single call overrides any client default options.

lookup_service_info: bool | None = None

Controls whether to look up service information based on the target ID.

If the service_info property is absent and lookup_service_info is True, the client will look up service information rather than using any client default service information.

authentication: AuthenticationInfo | Json | None = None

Authentication information to use for the operation.

Note: Authentication provided via DoipRequest overrides that provided via the options parameter of a single call, which overrides any client default options.

anonymous: bool | None = None

Indicates whether to make the request anonymously.

If the authentication property is absent and anonymous is True, the client will not use any default authentication information.

attributes: Json | None = None: Additional attributes to include in the request.

static from_args(options: Options | None, **kwargs: Unpack[OptionsTypedDict]) → Options | None

class OptionsTypedDict

Bases: TypedDict

Options manifested as a TypedDict; allows typing of keyword arguments for providing options.

service_info: ServiceInfo | None

lookup_service_info: bool | None

authentication: AuthenticationInfo | Json | None

anonymous: bool | None

attributes: Json | None

class PasswordAuthenticationInfo(username: str, password: str, as_user_id: str | None = None, client_id: str | None = None)

Bases: AuthenticationInfo

Used for password-based authentication.

Example:

import cnri_doip_client as doip_client

password_auth_info = doip_client.PasswordAuthenticationInfo("username", "password")
service_info = doip_client.ServiceInfo(ip_address="localhost", port=9000)
client = doip_client.StandardDoipClient(
    service_info=service_info, authentication=password_auth_info
)

__init__(username: str, password: str, as_user_id: str | None = None, client_id: str | None = None)

Constructs a PasswordAuthenticationInfo.

Parameters:

username – Username to use for authentication
password – Password to use for authentication
as_user_id – If present, instructs the DOIP service to perform operations as this user
client_id – Optional clientId to send in requests using this token

get_client_id() → str | None: Returns the clientId property to be used in DOIP requests made with this AuthenticationInfo.

get_authentication() → dict[str, Any]: Returns the authentication property to be used in DOIP requests made with this AuthenticationInfo.

class PrivateKeyAuthenticationInfo(client_id: str, private_key_jwk: dict[str, Any], as_user_id: str | None = None)

Bases: AuthenticationInfo

Used for authenticating using a private key.

Example:

import cnri_doip_client as doip_client

client_id = "20.500.123/2";
jwk = {
    "kty": "RSA",
    "n": "k31...",
    "e": "AQAB",
    "d": "NEY...",
    "p": "vR9...",
    "q": "x6I...",
    "dp": "plB...",
    "dq": "oqp...",
    "qi": "s7M...",
}
key_auth_info = doip_client.PrivateKeyAuthenticationInfo(client_id, jwk)
service_info = doip_client.ServiceInfo(ip_address="localhost", port=9000)
client = doip_client.StandardDoipClient(
    service_info=service_info, authentication=key_auth_info
)

__init__(client_id: str, private_key_jwk: dict[str, Any], as_user_id: str | None = None)

Constructs a PrivateKeyAuthenticationInfo.

Parameters:

client_id – Client ID for this private key
private_key_jwk – Private key to use to generate authentication for requests, in JSON Web Key format
as_user_id – If present, instructs the DOIP service to perform operations as this user

get_client_id() → str | None: Returns the clientId property to be used in DOIP requests made with this AuthenticationInfo.

get_authentication() → dict[str, Any]: Returns the authentication property to be used in DOIP requests made with this AuthenticationInfo.

class QueryParams(page_num: int = 0, page_size: int = -1, sort_fields: list[SortField] | None = None, filter_queries: list[str] | None = None, facets: list[Facet] | None = None)

Bases: object

Class defining parameters for a search query.

page_num: int = 0: The page number to retrieve (0-based).

page_size: int = -1: The number of items per page.

sort_fields: list[SortField] | None = None: Fields to sort the results by.

filter_queries: list[str] | None = None: Additional query strings to filter the results.

facets: list[Facet] | None = None: Facets to calculate for the search results.

class SearchResults(*args, **kwargs)

Bases: Iterator[T], Protocol[T]

Results of a search operation.

Contains facets and size properties, and is itself an iterator for accessing the results. The type of items in the search results is either string or DigitalObject.

Users should call close() to ensure that any underlying network operation is closed.

size: int: The total number of items in the search results.

facets: list[FacetResult] | None: Facet results, if faceting was requested.

property results: Self

to_dict() → dict[str, Any]

Converts a SearchResults into a dict, for use in JSON APIs. The dict will have a property “results” which is a list.

Returns:: The SearchResults as a dict

abstractmethod close() → None: Closes this SearchResults, freeing up any resources such as an open DOIP network connection.

Bases: ServiceInfoInterface

Information used to connect to a DOIP service.

This includes metadata about the service and information about the service’s available network interfaces. Note that because ServiceInfo extends ServiceInfoInterface, network interface information can be defined at the top level of this object or in the interfaces property.

Examples:

# A native-DOIP-only service
native_service_info = ServiceInfo(ip_address='localhost', port=9000)

# An DOIP-API-for-HTTP-Clients-only service
http_service_info = ServiceInfo(base_uri='https://localhost:8443/doip')

# A service with both TCP and HTTPS listeners
interfaces = [
    ServiceInfoInterface(protocol='TCP', ip_address='localhost', port=9000),
    ServiceInfoInterface(protocol='HTTPS', base_uri='https://localhost:8443/doip'),
]
multi_service_info = ServiceInfo(interfaces=interfaces)

service_id: str | None = None: The identifier for the service, typically used as the target ID for service-level operations.

protocol_version: str | None = None: The DOIP protocol version supported by the service.

public_key: Any = None: The public key of the service, used for verification.

interfaces: list[ServiceInfoInterface] | None = None: Interfaces through which the service can be accessed, in addition to any interface defined at the top level of the ServiceInfo object.

static from_dict(dict: dict) → ServiceInfo

Bases: object

Describes the basic connection details needed to communicate with a DOIP service via a specific endpoint.

protocol: Literal['TCP', 'HTTPS'] | None = None: The protocol used for communication.

ip_address: str | None = None: The IP address or hostname of the service. For example, 127.0.0.1 or example.org

port: int | None = None: The port number on which the service is listening.

base_uri: str | None = None: The base URI for connections using the DOIP API for HTTP Clients. See also Cordra’s documentation on this API. For example, https://example.org/doip/

path: str | None = None

Additional path information for the service endpoint.

For HTTPS interfaces without a baseUri, the path will be used to construct a URI, along with the ipAddress and port.

static from_dict(dict: dict) → ServiceInfoInterface

class SortField(name: str, reverse: bool = False)

Bases: object

Class defining a field to sort search results by.

name: str: The name of the field to sort by.

reverse: bool = False: Whether to sort in reverse (descending) order. If False or None, sorts in ascending order.

class StandardDoipClient(base_uri: str, options: Options | None = None, *, preferred_protocol: Literal['TCP', 'HTTPS'] | None = None, timeout: float | None = 60, **kwargs: Unpack[OptionsTypedDict])

class StandardDoipClient(options: Options | None = None, *, base_uri: str | None = None, preferred_protocol: Literal['TCP', 'HTTPS'] | None = None, timeout: float | None = 60, **kwargs: Unpack[OptionsTypedDict])

Bases: AbstractDoipClient

DOIP Client for TCP and HTTPS interfaces.

This class provides an implementation of the DOIP client that can use either HTTPS or native TCP for DOIP communication, depending on the available interfaces and configuration. It automatically selects the appropriate protocol (HTTPS or TCP) based on the service information and client capabilities.

Examples:

import cnri_doip_client as doip_client

authentication = doip_client.PasswordAuthenticationInfo(USERNAME, PASSWORD)
service_info = doip_client.ServiceInfo(ip_address="localhost", port=9000)
client = doip_client.StandardDoipClient(
    service_info=service_info, authentication=authentication
)

# Create a client with a base URI
client = doip_client.StandardDoipClient('https://localhost:8443/doip')

# Create a client with a base URI and authentication
authentication = doip_client.PasswordAuthenticationInfo(USERNAME, PASSWORD)
base_uri = 'https://localhost:8443/doip'
client = doip_client.StandardDoipClient(base_uri=base_uri, authentication=authentication)

# Create a client with multiple interfaces and a preferred protocol
interfaces = [
    doip_client.ServiceInfoInterface(ip_address="localhost", port=9000),
    doip_client.ServiceInfoInterface(base_uri="https://localhost:8443/doip"),
]
service_info = doip_client.ServiceInfo(interfaces=interfaces)
client = doip_client.StandardDoipClient(
    service_info=service_info, preferred_protocol='TCP'
)

__init__(base_uri: str, options: Options | None = None, *, preferred_protocol: Literal['TCP', 'HTTPS'] | None = None, timeout: float | None = 60, **kwargs: Unpack[OptionsTypedDict])

__init__(options: Options | None = None, *, base_uri: str | None = None, preferred_protocol: Literal['TCP', 'HTTPS'] | None = None, timeout: float | None = 60, **kwargs: Unpack[OptionsTypedDict])

Constructs a StandardDoipClient.

The primary way to call this constructor is with options as the first positional argument. However, as a convenience, a base_uri can instead be supplied as the first argument when constructing a StandardDoipClient which uses the DOIP API for HTTP Clients, with options as the second argument.

Parameters:

options – Optional default request options. The serviceInfo and authentication property can be used to provide a default serviceInfo and authentication for all requests.
base_uri – The base HTTP endpoint of the DOIP API for HTTP Clients service
preferred_protocol – Either “TCP” for native DOIP or “HTTPS” for the DOIP API for HTTP Clients
timeout – Default timeout for requests
**kwargs – Optional request options provided as keyword arguments; these override any given in the options parameter

perform_operation_via_request(request: DoipRequest, options: Options | None = None) → DoipResponse

This method must be implemented by concrete subclasses to handle the actual communication with DOIP services. Users of subclasses of DoipClient should prefer DoipClient.perform_operation() or the various convenience methods of DoipClient.

Parameters:

request – The DOIP request to perform
options – Optional request options for this operation

Returns:

A promise that resolves to a DoipResponse

close() → None: Closes the client and releases any resources.

class SupportsRead(*args, **kwargs)

Bases: Protocol

A type for a readable file-like object.

read(length: int = Ellipsis, /) → bytes

class TokenAuthenticationInfo(token: str, client_id: str | None = None, as_user_id: str | None = None)

Bases: AuthenticationInfo

Used for token-based authentication.

Example:

import cnri_doip_client as doip_client

token_auth_info = doip_client.TokenAuthenticationInfo("ACCESS_TOKEN")
service_info = doip_client.ServiceInfo(ip_address="localhost", port=9000)
client = doip_client.StandardDoipClient(
    service_info=service_info, authentication=token_auth_info
)

__init__(token: str, client_id: str | None = None, as_user_id: str | None = None)

Constructs a TokenAuthenticationInfo.

Parameters:

token – Authentication token. This should be acquired from the DOIP service using whatever method is provided by that service.
clientId – Optional clientId to send in requests using this token
asUserId – If present, instructs the DOIP service to perform operations as this user

get_client_id() → str | None: Returns the clientId property to be used in DOIP requests made with this AuthenticationInfo.

get_authentication() → dict[str, Any]: Returns the authentication property to be used in DOIP requests made with this AuthenticationInfo.