Skip to content

ETP v1.2 messages and objects

This section documents the implemented protocols and data types from ETP v1.2. We aim to support the same subset of ETP v1.2 as the OSDU open-etp-server. Documentation for ETP v1.2 is found here: https://publications.opengroup.org/standards/energistics-standards/energistics-transfer-protocol/v234.

Protocol 0: Core

energistics.etp.v12.protocol.core

__all__ module-attribute 

__all__ = [
    "Acknowledge",
    "Authorize",
    "AuthorizeResponse",
    "CloseSession",
    "OpenSession",
    "Ping",
    "Pong",
    "ProtocolException",
    "RequestSession",
]

Acknowledge

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)

Authorize

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
authorization instance-attribute 
authorization: str
supplemental_authorization class-attribute instance-attribute 
supplemental_authorization: Mapping[str, str] = Field(
    alias="supplementalAuthorization", default_factory=dict
)

AuthorizeResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
success instance-attribute 
success: bool
challenges class-attribute instance-attribute 
challenges: list[str] = Field(default_factory=list)
model_post_init 
model_post_init(__context: Any) -> None
Source code in src/energistics/etp/v12/protocol/core/authorize_response.py 
30
31
32
33
34
35
36
37
38
def model_post_init(self, __context: typing.Any) -> None:
    if self.success and self.challenges:
        # This is not permitted according to the ETP v1.2 spec, but it
        # might be ignored in the wild and so possibly this should be a
        # warning or ignored instead.
        raise ValueError(
            f"Authorization was successful, but challenges {self.challenges} is "
            "non-empty"
        )

CloseSession

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
reason class-attribute instance-attribute 
reason: str = ''

OpenSession

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
application_name class-attribute instance-attribute 
application_name: str = Field(alias='applicationName')
application_version class-attribute instance-attribute 
application_version: str = Field(alias='applicationVersion')
server_instance_id class-attribute instance-attribute 
server_instance_id: Uuid = Field(alias='serverInstanceId')
supported_protocols class-attribute instance-attribute 
supported_protocols: list[SupportedProtocol] = Field(alias="supportedProtocols")
supported_data_objects class-attribute instance-attribute 
supported_data_objects: list[SupportedDataObject] = Field(
    alias="supportedDataObjects"
)
supported_compression class-attribute instance-attribute 
supported_compression: str = Field(alias='supportedCompression', default='')
supported_formats class-attribute instance-attribute 
supported_formats: list[str] = Field(
    alias="supportedFormats", default_factory=lambda: ["xml"]
)
current_date_time class-attribute instance-attribute 
current_date_time: int = Field(
    alias="currentDateTime",
    default_factory=lambda: int(timestamp() * 1000000.0),
)
earliest_retained_change_time class-attribute instance-attribute 
earliest_retained_change_time: int = Field(alias='earliestRetainedChangeTime')
session_id class-attribute instance-attribute 
session_id: Uuid = Field(alias='sessionId')
endpoint_capabilities class-attribute instance-attribute 
endpoint_capabilities: Mapping[EndpointCapabilityKind, DataValue] = Field(
    alias="endpointCapabilities", default_factory=dict
)

Ping

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
current_date_time class-attribute instance-attribute 
current_date_time: int = Field(
    alias="currentDateTime",
    default_factory=lambda: int(timestamp() * 1000000.0),
)

Pong

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
current_date_time class-attribute instance-attribute 
current_date_time: int = Field(
    alias="currentDateTime",
    default_factory=lambda: int(timestamp() * 1000000.0),
)

ProtocolException

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
error instance-attribute 
error: ErrorInfo | None
errors class-attribute instance-attribute 
errors: Mapping[str, ErrorInfo] = Field(default_factory=dict)
model_post_init 
model_post_init(__context: Any) -> None
Source code in src/energistics/etp/v12/protocol/core/protocol_exception.py 
44
45
46
47
48
def model_post_init(self, __context: typing.Any) -> None:
    if self.error is None and not self.errors:
        raise ValueError("No error-fields were populated")
    if self.error is not None and self.errors:
        raise ValueError("Both error-fields were populated")

RequestSession

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
application_name class-attribute instance-attribute 
application_name: str = Field(alias='applicationName')
application_version class-attribute instance-attribute 
application_version: str = Field(alias='applicationVersion')
client_instance_id class-attribute instance-attribute 
client_instance_id: Uuid = Field(alias='clientInstanceId')
requested_protocols class-attribute instance-attribute 
requested_protocols: list[SupportedProtocol] = Field(alias="requestedProtocols")
supported_data_objects class-attribute instance-attribute 
supported_data_objects: list[SupportedDataObject] = Field(
    alias="supportedDataObjects"
)
supported_compression class-attribute instance-attribute 
supported_compression: list[str] = Field(
    alias="supportedCompression", default_factory=list
)
supported_formats class-attribute instance-attribute 
supported_formats: list[str] = Field(
    alias="supportedFormats", default_factory=lambda: ["xml"]
)
current_date_time class-attribute instance-attribute 
current_date_time: int = Field(
    alias="currentDateTime",
    default_factory=lambda: int(timestamp() * 1000000.0),
)
earliest_retained_change_time class-attribute instance-attribute 
earliest_retained_change_time: int = Field(
    alias="earliestRetainedChangeTime", default=0
)
server_authorization_required class-attribute instance-attribute 
server_authorization_required: bool = Field(
    alias="serverAuthorizationRequired", default=False
)
endpoint_capabilities class-attribute instance-attribute 
endpoint_capabilities: Mapping[EndpointCapabilityKind, DataValue] = Field(
    alias="endpointCapabilities", default_factory=dict
)

Protocol 3: Discovery

energistics.etp.v12.protocol.discovery

__all__ module-attribute 

__all__ = [
    "GetDeletedResources",
    "GetDeletedResourcesResponse",
    "GetResources",
    "GetResourcesEdgesResponse",
    "GetResourcesResponse",
]

GetDeletedResources

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
dataspace_uri instance-attribute 
dataspace_uri: str
delete_time_filter class-attribute instance-attribute 
delete_time_filter: int | None = Field(alias='deleteTimeFilter')
data_object_types instance-attribute 
data_object_types: list[str]

GetDeletedResourcesResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
deleted_resources class-attribute instance-attribute 
deleted_resources: list[DeletedResource] = Field(
    alias="deletedResources", default_factory=list
)

GetResources

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
context instance-attribute 
context: ContextInfo
scope instance-attribute 
scope: ContextScopeKind
count_objects class-attribute instance-attribute 
count_objects: bool = Field(alias='countObjects', default=False)
store_last_write_filter class-attribute instance-attribute 
store_last_write_filter: int | None = Field(
    alias="storeLastWriteFilter", default=None
)
active_status_filter class-attribute instance-attribute 
active_status_filter: ActiveStatusKind | None = Field(
    alias="activeStatusFilter", default=None
)
include_edges class-attribute instance-attribute 
include_edges: bool = Field(alias='includeEdges', default=False)

GetResourcesEdgesResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
edges instance-attribute 
edges: list[Edge]

GetResourcesResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
resources class-attribute instance-attribute 
resources: list[Resource] = Field(default_factory=list)

Protocol 4: Store

energistics.etp.v12.protocol.store

__all__ module-attribute 

__all__ = [
    "Chunk",
    "DeleteDataObjects",
    "DeleteDataObjectsResponse",
    "GetDataObjects",
    "GetDataObjectsResponse",
    "PutDataObjects",
    "PutDataObjectsResponse",
]

Chunk

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
blob_id class-attribute instance-attribute 
blob_id: Uuid = Field(alias='blobId')
data instance-attribute 
data: bytes
final instance-attribute 
final: bool

DeleteDataObjects

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
uris instance-attribute 
uris: Mapping[str, str]
prune_contained_objects class-attribute instance-attribute 
prune_contained_objects: bool = Field(
    alias="pruneContainedObjects", default=False
)

DeleteDataObjectsResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
deleted_uris instance-attribute 
deleted_uris: Mapping[str, ArrayOfString]

GetDataObjects

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
uris instance-attribute 
uris: Mapping[str, str]
format class-attribute instance-attribute 
format: str = 'xml'

GetDataObjectsResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
data_objects class-attribute instance-attribute 
data_objects: Mapping[str, DataObject] = Field(
    alias="dataObjects", default_factory=dict
)

PutDataObjects

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
data_objects class-attribute instance-attribute 
data_objects: Mapping[str, DataObject] = Field(alias='dataObjects')
prune_contained_objects class-attribute instance-attribute 
prune_contained_objects: bool = Field(
    alias="pruneContainedObjects", default=False
)

PutDataObjectsResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
success instance-attribute 
success: Mapping[str, PutResponse]

Protocol 9: DataArray

energistics.etp.v12.protocol.data_array

__all__ module-attribute 

__all__ = [
    "GetDataArrayMetadata",
    "GetDataArrayMetadataResponse",
    "GetDataArrays",
    "GetDataArraysResponse",
    "GetDataSubarrays",
    "GetDataSubarraysResponse",
    "PutDataArrays",
    "PutDataArraysResponse",
    "PutDataSubarrays",
    "PutDataSubarraysResponse",
    "PutUninitializedDataArrays",
    "PutUninitializedDataArraysResponse",
]

GetDataArrayMetadata

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
data_arrays class-attribute instance-attribute 
data_arrays: Mapping[str, DataArrayIdentifier] = Field(alias='dataArrays')

GetDataArrayMetadataResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
array_metadata class-attribute instance-attribute 
array_metadata: Mapping[str, DataArrayMetadata] = Field(
    alias="arrayMetadata", default_factory=dict
)

GetDataArrays

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
data_arrays class-attribute instance-attribute 
data_arrays: Mapping[str, DataArrayIdentifier] = Field(alias='dataArrays')

GetDataArraysResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
data_arrays class-attribute instance-attribute 
data_arrays: Mapping[str, DataArray] = Field(
    alias="dataArrays", default_factory=dict
)

GetDataSubarrays

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
data_subarrays class-attribute instance-attribute 
data_subarrays: Mapping[str, GetDataSubarraysType] = Field(
    alias="dataSubarrays"
)

GetDataSubarraysResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
data_subarrays class-attribute instance-attribute 
data_subarrays: Mapping[str, DataArray] = Field(
    alias="dataSubarrays", default_factory=dict
)

PutDataArrays

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
data_arrays class-attribute instance-attribute 
data_arrays: Mapping[str, PutDataArraysType] = Field(alias='dataArrays')

PutDataArraysResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
success instance-attribute 
success: Mapping[str, str]

PutDataSubarrays

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
data_subarrays class-attribute instance-attribute 
data_subarrays: Mapping[str, PutDataSubarraysType] = Field(
    alias="dataSubarrays"
)

PutDataSubarraysResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
success instance-attribute 
success: Mapping[str, str]

PutUninitializedDataArrays

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
data_arrays class-attribute instance-attribute 
data_arrays: Mapping[str, PutUninitializedDataArrayType] = Field(
    alias="dataArrays"
)

PutUninitializedDataArraysResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
success instance-attribute 
success: Mapping[str, str]

Protocol 18: Transaction

energistics.etp.v12.protocol.transaction

__all__ module-attribute 

__all__ = [
    "CommitTransaction",
    "CommitTransactionResponse",
    "RollbackTransaction",
    "RollbackTransactionResponse",
    "StartTransaction",
    "StartTransactionResponse",
]

CommitTransaction

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
transaction_uuid class-attribute instance-attribute 
transaction_uuid: Uuid = Field(alias='transactionUuid')

CommitTransactionResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
transaction_uuid class-attribute instance-attribute 
transaction_uuid: Uuid = Field(alias='transactionUuid')
successful class-attribute instance-attribute 
successful: bool = True
failure_reason class-attribute instance-attribute 
failure_reason: str = Field(alias='failureReason', default='')

RollbackTransaction

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
transaction_uuid class-attribute instance-attribute 
transaction_uuid: Uuid = Field(alias='transactionUuid')

RollbackTransactionResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
transaction_uuid class-attribute instance-attribute 
transaction_uuid: Uuid = Field(alias='transactionUuid')
successful class-attribute instance-attribute 
successful: bool = True
failure_reason class-attribute instance-attribute 
failure_reason: str = Field(alias='failureReason', default='')

StartTransaction

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
read_only class-attribute instance-attribute 
read_only: bool = Field(alias='readOnly')
message class-attribute instance-attribute 
message: str = ''
dataspace_uris instance-attribute 
dataspace_uris: list[str]

StartTransactionResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
transaction_uuid class-attribute instance-attribute 
transaction_uuid: Uuid = Field(alias='transactionUuid')
successful class-attribute instance-attribute 
successful: bool = True
failure_reason class-attribute instance-attribute 
failure_reason: str = Field(alias='failureReason', default='')

Protocol 24: Dataspace

energistics.etp.v12.protocol.dataspace

__all__ module-attribute 

__all__ = [
    "DeleteDataspaces",
    "DeleteDataspacesResponse",
    "GetDataspaces",
    "GetDataspacesResponse",
    "PutDataspaces",
    "PutDataspacesResponse",
]

DeleteDataspaces

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
uris instance-attribute 
uris: Mapping[str, str]
model_post_init 
model_post_init(__context: Any) -> None
Source code in src/energistics/etp/v12/protocol/dataspace/delete_dataspaces.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def model_post_init(self, __context: typing.Any) -> None:
    errors = []
    for k, uri in self.uris.items():
        m = re.match(energistics.uris.DATASPACE_URI_PATTERN, uri)

        if m is None:
            error = ValueError(f"Uri({k}) '{uri}' is not a valid ETP dataspace uri")
            errors.append(error)

    if len(errors) > 0:
        # NOTE: This does not raise pydantic.ValidationError, but instead an
        # ExceptionGroup of ValueError's.
        raise ExceptionGroup(
            f"There were {len(errors)} invalid ETP dataspace uris",
            errors,
        )

DeleteDataspacesResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
success instance-attribute 
success: Mapping[str, str]

GetDataspaces

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
store_last_write_filter class-attribute instance-attribute 
store_last_write_filter: int | None = Field(
    alias="storeLastWriteFilter", default=None
)

GetDataspacesResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
dataspaces class-attribute instance-attribute 
dataspaces: list[Dataspace] = Field(default_factory=list)

PutDataspaces

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
dataspaces instance-attribute 
dataspaces: Mapping[str, Dataspace]

PutDataspacesResponse

Bases: ETPBaseProtocolModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
success instance-attribute 
success: Mapping[str, str]

ETP Datatypes

energistics.etp.v12.datatypes

__all__ module-attribute 

__all__ = [
    "AnyArray",
    "AnyArrayType",
    "AnyLogicalArrayType",
    "AnySparseArray",
    "AnySubarray",
    "ArrayOfBoolean",
    "ArrayOfBytes",
    "ArrayOfDouble",
    "ArrayOfFloat",
    "ArrayOfInt",
    "ArrayOfLong",
    "ArrayOfNullableBoolean",
    "ArrayOfNullableInt",
    "ArrayOfNullableLong",
    "ArrayOfString",
    "Contact",
    "DataObjectCapabilityKind",
    "DataValue",
    "ErrorCode",
    "ErrorInfo",
    "MessageHeader",
    "ProtocolCapabilityKind",
    "ServerCapabilities",
    "SupportedDataObject",
    "SupportedProtocol",
    "Uuid",
    "Version",
    "EndpointCapabilityKind",
]

AnyArray

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
item instance-attribute 
item: ETPBasicArrayType | bytes

AnyArrayType

Bases: StrEnum

ARRAY_OF_BOOLEAN class-attribute instance-attribute 
ARRAY_OF_BOOLEAN = 'arrayOfBoolean'
ARRAY_OF_INT class-attribute instance-attribute 
ARRAY_OF_INT = 'arrayOfInt'
ARRAY_OF_LONG class-attribute instance-attribute 
ARRAY_OF_LONG = 'arrayOfLong'
ARRAY_OF_FLOAT class-attribute instance-attribute 
ARRAY_OF_FLOAT = 'arrayOfFloat'
ARRAY_OF_DOUBLE class-attribute instance-attribute 
ARRAY_OF_DOUBLE = 'arrayOfDouble'
ARRAY_OF_STRING class-attribute instance-attribute 
ARRAY_OF_STRING = 'arrayOfString'
BYTES class-attribute instance-attribute 
BYTES = 'bytes'

AnyLogicalArrayType

Bases: StrEnum

ARRAY_OF_BOOLEAN class-attribute instance-attribute 
ARRAY_OF_BOOLEAN = 'arrayOfBoolean'
ARRAY_OF_INT8 class-attribute instance-attribute 
ARRAY_OF_INT8 = 'arrayOfInt8'
ARRAY_OF_UINT8 class-attribute instance-attribute 
ARRAY_OF_UINT8 = 'arrayOfUInt8'
ARRAY_OF_INT16_LE class-attribute instance-attribute 
ARRAY_OF_INT16_LE = 'arrayOfInt16LE'
ARRAY_OF_INT32_LE class-attribute instance-attribute 
ARRAY_OF_INT32_LE = 'arrayOfInt32LE'
ARRAY_OF_INT64_LE class-attribute instance-attribute 
ARRAY_OF_INT64_LE = 'arrayOfInt64LE'
ARRAY_OF_UINT16_LE class-attribute instance-attribute 
ARRAY_OF_UINT16_LE = 'arrayOfUInt16LE'
ARRAY_OF_UINT32_LE class-attribute instance-attribute 
ARRAY_OF_UINT32_LE = 'arrayOfUInt32LE'
ARRAY_OF_UINT64_LE class-attribute instance-attribute 
ARRAY_OF_UINT64_LE = 'arrayOfUInt64LE'
ARRAY_OF_FLOAT32_LE class-attribute instance-attribute 
ARRAY_OF_FLOAT32_LE = 'arrayOfFloat32LE'
ARRAY_OF_DOUBLE64_LE class-attribute instance-attribute 
ARRAY_OF_DOUBLE64_LE = 'arrayOfDouble64LE'
ARRAY_OF_INT16_BE class-attribute instance-attribute 
ARRAY_OF_INT16_BE = 'arrayOfInt16BE'
ARRAY_OF_INT32_BE class-attribute instance-attribute 
ARRAY_OF_INT32_BE = 'arrayOfInt32BE'
ARRAY_OF_INT64_BE class-attribute instance-attribute 
ARRAY_OF_INT64_BE = 'arrayOfInt64BE'
ARRAY_OF_UINT16_BE class-attribute instance-attribute 
ARRAY_OF_UINT16_BE = 'arrayOfUInt16BE'
ARRAY_OF_UINT32_BE class-attribute instance-attribute 
ARRAY_OF_UINT32_BE = 'arrayOfUInt32BE'
ARRAY_OF_UINT64_BE class-attribute instance-attribute 
ARRAY_OF_UINT64_BE = 'arrayOfUInt64BE'
ARRAY_OF_FLOAT32_BE class-attribute instance-attribute 
ARRAY_OF_FLOAT32_BE = 'arrayOfFloat32BE'
ARRAY_OF_DOUBLE64_BE class-attribute instance-attribute 
ARRAY_OF_DOUBLE64_BE = 'arrayOfDouble64BE'
ARRAY_OF_STRING class-attribute instance-attribute 
ARRAY_OF_STRING = 'arrayOfString'
ARRAY_OF_CUSTOM class-attribute instance-attribute 
ARRAY_OF_CUSTOM = 'arrayOfCustom'

AnySparseArray

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
slices instance-attribute 
slices: list[AnySubarray]

AnySubarray

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
start instance-attribute 
start: int
slice instance-attribute 
slice: AnyArray

ArrayOfBoolean

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
values instance-attribute 
values: NDArray[bool_]

ArrayOfBytes

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
values instance-attribute 
values: NDArray[bytes_]

ArrayOfDouble

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
values instance-attribute 
values: NDArray[float64]

ArrayOfFloat

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
values instance-attribute 
values: NDArray[float32]

ArrayOfInt

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
values instance-attribute 
values: NDArray[int32]

ArrayOfLong

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
values instance-attribute 
values: NDArray[int64]

ArrayOfNullableBoolean

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
values instance-attribute 
values: MaskedArray[tuple[int], dtype[bool_]]

ArrayOfNullableInt

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
values instance-attribute 
values: MaskedArray[tuple[int], dtype[int32]]

ArrayOfNullableLong

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
values instance-attribute 
values: MaskedArray[tuple[int], dtype[int64]]

ArrayOfString

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
values instance-attribute 
values: NDArray[str_]

Contact

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
organization_name class-attribute instance-attribute 
organization_name: str = Field(alias='organizationName', default='')
contact_name class-attribute instance-attribute 
contact_name: str = Field(alias='contactName', default='')
contact_phone class-attribute instance-attribute 
contact_phone: str = Field(alias='contactPhone', default='')
contact_email class-attribute instance-attribute 
contact_email: str = Field(alias='contactEmail', default='')

DataObjectCapabilityKind

Bases: StrEnum

ACTIVE_TIMEOUT_PERIOD class-attribute instance-attribute 
ACTIVE_TIMEOUT_PERIOD = 'ActiveTimeoutPeriod'
MAX_CONTAINED_DATA_OBJECT_COUNT class-attribute instance-attribute 
MAX_CONTAINED_DATA_OBJECT_COUNT = 'MaxContainedDataObjectCount'
MAX_DATA_OBJECT_SIZE class-attribute instance-attribute 
MAX_DATA_OBJECT_SIZE = 'MaxDataObjectSize'
ORPHANED_CHILDREN_PRUNED_ON_DELETE class-attribute instance-attribute 
ORPHANED_CHILDREN_PRUNED_ON_DELETE = 'OrphanedChildrenPrunedOnDelete'
SUPPORTS_GET class-attribute instance-attribute 
SUPPORTS_GET = 'SupportsGet'
SUPPORTS_PUT class-attribute instance-attribute 
SUPPORTS_PUT = 'SupportsPut'
SUPPORTS_DELETE class-attribute instance-attribute 
SUPPORTS_DELETE = 'SupportsDelete'
MAX_SECONDARY_INDEX_COUNT class-attribute instance-attribute 
MAX_SECONDARY_INDEX_COUNT = 'MaxSecondaryIndexCount'
get_valid_type 
get_valid_type() -> Type[bool | int]
Source code in src/energistics/etp/v12/datatypes/data_object_capability_kind.py 
35
36
37
38
39
40
41
42
43
def get_valid_type(self) -> typing.Type[bool | int]:
    if self in [
        DataObjectCapabilityKind.ORPHANED_CHILDREN_PRUNED_ON_DELETE,
        DataObjectCapabilityKind.SUPPORTS_GET,
        DataObjectCapabilityKind.SUPPORTS_PUT,
        DataObjectCapabilityKind.SUPPORTS_DELETE,
    ]:
        return bool
    return int

DataValue

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
item instance-attribute 
item: ItemType

EndpointCapabilityKind

Bases: StrEnum

ACTIVE_TIMEOUT_PERIOD class-attribute instance-attribute 
ACTIVE_TIMEOUT_PERIOD = 'ActiveTimeoutPeriod'
AUTHORIZATION_DETAILS class-attribute instance-attribute 
AUTHORIZATION_DETAILS = 'AuthorizationDetails'
CHANGE_PROPAGATION_PERIOD class-attribute instance-attribute 
CHANGE_PROPAGATION_PERIOD = 'ChangePropagationPeriod'
CHANGE_RETENTION_PERIOD class-attribute instance-attribute 
CHANGE_RETENTION_PERIOD = 'ChangeRetentionPeriod'
MAX_CONCURRENT_MULTIPART class-attribute instance-attribute 
MAX_CONCURRENT_MULTIPART = 'MaxConcurrentMultipart'
MAX_DATA_OBJECT_SIZE class-attribute instance-attribute 
MAX_DATA_OBJECT_SIZE = 'MaxDataObjectSize'
MAX_PART_SIZE class-attribute instance-attribute 
MAX_PART_SIZE = 'MaxPartSize'
MAX_SESSION_CLIENT_COUNT class-attribute instance-attribute 
MAX_SESSION_CLIENT_COUNT = 'MaxSessionClientCount'
MAX_SESSION_GLOBAL_COUNT class-attribute instance-attribute 
MAX_SESSION_GLOBAL_COUNT = 'MaxSessionGlobalCount'
MAX_WEB_SOCKET_FRAME_PAYLOAD_SIZE class-attribute instance-attribute 
MAX_WEB_SOCKET_FRAME_PAYLOAD_SIZE = 'MaxWebSocketFramePayloadSize'
MAX_WEB_SOCKET_MESSAGE_PAYLOAD_SIZE class-attribute instance-attribute 
MAX_WEB_SOCKET_MESSAGE_PAYLOAD_SIZE = 'MaxWebSocketMessagePayloadSize'
MULTIPART_MESSAGE_TIMEOUT_PERIOD class-attribute instance-attribute 
MULTIPART_MESSAGE_TIMEOUT_PERIOD = 'MultipartMessageTimeoutPeriod'
RESPONSE_TIMEOUT_PERIOD class-attribute instance-attribute 
RESPONSE_TIMEOUT_PERIOD = 'ResponseTimeoutPeriod'
REQUEST_SESSION_TIMEOUT_PERIOD class-attribute instance-attribute 
REQUEST_SESSION_TIMEOUT_PERIOD = 'RequestSessionTimeoutPeriod'
SESSION_ESTABLISHMENT_TIMEOUT_PERIOD class-attribute instance-attribute 
SESSION_ESTABLISHMENT_TIMEOUT_PERIOD = 'SessionEstablishmentTimeoutPeriod'
SUPPORTS_ALTERNATE_REQUEST_URIS class-attribute instance-attribute 
SUPPORTS_ALTERNATE_REQUEST_URIS = 'SupportsAlternateRequestUris'
SUPPORTS_MESSAGE_HEADER_EXTENSIONS class-attribute instance-attribute 
SUPPORTS_MESSAGE_HEADER_EXTENSIONS = 'SupportsMessageHeaderExtensions'
MAX_MESSAGE_PAYLOAD_UNCOMPRESSED_SIZE class-attribute instance-attribute 
MAX_MESSAGE_PAYLOAD_UNCOMPRESSED_SIZE = 'MaxMessagePayloadUncompressedSize'
get_valid_type 
get_valid_type() -> Type[bool | int | ArrayOfString]
Source code in src/energistics/etp/v12/datatypes/endpoint_capability_kind.py 
58
59
60
61
62
63
64
65
66
67
68
def get_valid_type(self) -> typing.Type[bool | int | ArrayOfString]:
    match self:
        case EndpointCapabilityKind.AUTHORIZATION_DETAILS:
            return ArrayOfString
        case (
            EndpointCapabilityKind.SUPPORTS_ALTERNATE_REQUEST_URIS
            | EndpointCapabilityKind.SUPPORTS_MESSAGE_HEADER_EXTENSIONS
        ):
            return bool

    return int

ErrorCode

Bases: IntEnum

ENOROLE class-attribute instance-attribute 
ENOROLE = 1
ENOSUPPORTEDPROTOCOLS class-attribute instance-attribute 
ENOSUPPORTEDPROTOCOLS = 2
EINVALID_MESSAGETYPE class-attribute instance-attribute 
EINVALID_MESSAGETYPE = 3
EUNSUPPORTED_PROTOCOL class-attribute instance-attribute 
EUNSUPPORTED_PROTOCOL = 4
EINVALID_ARGUMENT class-attribute instance-attribute 
EINVALID_ARGUMENT = 5
EREQUEST_DENIED class-attribute instance-attribute 
EREQUEST_DENIED = 6
ENOTSUPPORTED class-attribute instance-attribute 
ENOTSUPPORTED = 7
EINVALID_STATE class-attribute instance-attribute 
EINVALID_STATE = 8
EINVALID_URI class-attribute instance-attribute 
EINVALID_URI = 9
EAUTHORIZATION_EXPIRED class-attribute instance-attribute 
EAUTHORIZATION_EXPIRED = 10
ENOT_FOUND class-attribute instance-attribute 
ENOT_FOUND = 11
ELIMIT_EXCEEDED class-attribute instance-attribute 
ELIMIT_EXCEEDED = 12
ECOMPRESSION_NOTSUPPORTED class-attribute instance-attribute 
ECOMPRESSION_NOTSUPPORTED = 13
EINVALID_OBJECT class-attribute instance-attribute 
EINVALID_OBJECT = 14
EMAX_TRANSACTIONS_EXCEEDED class-attribute instance-attribute 
EMAX_TRANSACTIONS_EXCEEDED = 15
EDATAOBJECTTYPE_NOTSUPPORTED class-attribute instance-attribute 
EDATAOBJECTTYPE_NOTSUPPORTED = 16
EMAXSIZE_EXCEEDED class-attribute instance-attribute 
EMAXSIZE_EXCEEDED = 17
EMULTIPART_CANCELLED class-attribute instance-attribute 
EMULTIPART_CANCELLED = 18
EINVALID_MESSAGE class-attribute instance-attribute 
EINVALID_MESSAGE = 19
EINVALID_INDEXKIND class-attribute instance-attribute 
EINVALID_INDEXKIND = 20
ENOSUPPORTEDFORMATS class-attribute instance-attribute 
ENOSUPPORTEDFORMATS = 21
EREQUESTUUID_REJECTED class-attribute instance-attribute 
EREQUESTUUID_REJECTED = 22
EUPDATEGROWINGOBJECT_DENIED class-attribute instance-attribute 
EUPDATEGROWINGOBJECT_DENIED = 23
EBACKPRESSURE_LIMIT_EXCEEDED class-attribute instance-attribute 
EBACKPRESSURE_LIMIT_EXCEEDED = 24
EBACKPRESSURE_WARNING class-attribute instance-attribute 
EBACKPRESSURE_WARNING = 25
ETIMED_OUT class-attribute instance-attribute 
ETIMED_OUT = 26
EAUTHORIZATION_REQUIRED class-attribute instance-attribute 
EAUTHORIZATION_REQUIRED = 27
EAUTHORIZATION_EXPIRING class-attribute instance-attribute 
EAUTHORIZATION_EXPIRING = 28
ENOSUPPORTEDDATAOBJECTTYPES class-attribute instance-attribute 
ENOSUPPORTEDDATAOBJECTTYPES = 29
ERESPONSECOUNT_EXCEEDED class-attribute instance-attribute 
ERESPONSECOUNT_EXCEEDED = 30
EINVALID_APPEND class-attribute instance-attribute 
EINVALID_APPEND = 31
EINVALID_OPERATION class-attribute instance-attribute 
EINVALID_OPERATION = 32
EINVALID_CHANNELID class-attribute instance-attribute 
EINVALID_CHANNELID = 1002
ENOCASCADE_DELETE class-attribute instance-attribute 
ENOCASCADE_DELETE = 4003
EPLURAL_OBJECT class-attribute instance-attribute 
EPLURAL_OBJECT = 4004
ERETENTION_PERIOD_EXCEEDED class-attribute instance-attribute 
ERETENTION_PERIOD_EXCEEDED = 5001
ENOTGROWINGOBJECT class-attribute instance-attribute 
ENOTGROWINGOBJECT = 6001

ErrorInfo

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
message instance-attribute 
message: str
code instance-attribute 
code: ErrorCode

MessageHeader

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
protocol instance-attribute 
protocol: Protocol
message_type class-attribute instance-attribute 
message_type: int = Field(alias='messageType')
correlation_id class-attribute instance-attribute 
correlation_id: int = Field(alias='correlationId')
message_id class-attribute instance-attribute 
message_id: int = Field(alias='messageId')
message_flags class-attribute instance-attribute 
message_flags: MessageHeaderFlags = Field(alias='messageFlags')
check_allowed_compression 
check_allowed_compression() -> Self
Source code in src/energistics/etp/v12/datatypes/message_header.py 
40
41
42
43
44
45
46
47
48
@model_validator(mode="after")
def check_allowed_compression(self) -> typing.Self:
    if (
        self.protocol == energistics.base.Protocol.CORE
        or self.message_type in [1000, 1001]
    ) and self.is_compressed():
        raise ValueError("Messages from core must not be compressed")

    return self
validate_message_id classmethod 
validate_message_id(value: int) -> int
Source code in src/energistics/etp/v12/datatypes/message_header.py 
50
51
52
53
54
55
56
@field_validator("message_id", mode="after")
@classmethod
def validate_message_id(cls, value: int) -> int:
    if value == 0:
        raise ValueError("A message id of 0 is invalid")

    return value
from_etp_protocol_body classmethod 
from_etp_protocol_body(
    body: Type[ETPBaseProtocolModel],
    message_id: int,
    message_flags: MessageHeaderFlags,
    correlation_id: int = 0,
) -> Self
Source code in src/energistics/etp/v12/datatypes/message_header.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@classmethod
def from_etp_protocol_body(
    cls,
    body: typing.Type[energistics.base.ETPBaseProtocolModel],
    message_id: int,
    message_flags: MessageHeaderFlags,
    correlation_id: int = 0,
) -> typing.Self:
    return cls(
        protocol=body._protocol,
        message_type=body._message_type,
        correlation_id=correlation_id,
        message_id=message_id,
        message_flags=message_flags,
    )
is_final_message 
is_final_message() -> bool
Source code in src/energistics/etp/v12/datatypes/message_header.py 
74
75
def is_final_message(self) -> bool:
    return (self.message_flags & MessageHeaderFlags.FIN) != 0
is_compressed 
is_compressed() -> bool
Source code in src/energistics/etp/v12/datatypes/message_header.py 
77
78
def is_compressed(self) -> bool:
    return (self.message_flags & MessageHeaderFlags.COMPRESSED) != 0
requests_acknowledgement 
requests_acknowledgement() -> bool
Source code in src/energistics/etp/v12/datatypes/message_header.py 
80
81
def requests_acknowledgement(self) -> bool:
    return (self.message_flags & MessageHeaderFlags.ACK) != 0
uses_extension_header 
uses_extension_header() -> bool
Source code in src/energistics/etp/v12/datatypes/message_header.py 
83
84
def uses_extension_header(self) -> bool:
    return (self.message_flags & MessageHeaderFlags.EXTENSION) != 0

ProtocolCapabilityKind

Bases: StrEnum

FRAME_CHANGE_DETECTION_PERIOD class-attribute instance-attribute 
FRAME_CHANGE_DETECTION_PERIOD = 'FrameChangeDetectionPeriod'
MAX_DATA_ARRAY_SIZE class-attribute instance-attribute 
MAX_DATA_ARRAY_SIZE = 'MaxDataArraySize'
MAX_DATA_OBJECT_SIZE class-attribute instance-attribute 
MAX_DATA_OBJECT_SIZE = 'MaxDataObjectSize'
MAX_FRAME_RESPONSE_ROW_COUNT class-attribute instance-attribute 
MAX_FRAME_RESPONSE_ROW_COUNT = 'MaxFrameResponseRowCount'
MAX_INDEX_COUNT class-attribute instance-attribute 
MAX_INDEX_COUNT = 'MaxIndexCount'
MAX_RANGE_CHANNEL_COUNT class-attribute instance-attribute 
MAX_RANGE_CHANNEL_COUNT = 'MaxRangeChannelCount'
MAX_RANGE_DATA_ITEM_COUNT class-attribute instance-attribute 
MAX_RANGE_DATA_ITEM_COUNT = 'MaxRangeDataItemCount'
MAX_RESPONSE_COUNT class-attribute instance-attribute 
MAX_RESPONSE_COUNT = 'MaxResponseCount'
MAX_STREAMING_CHANNELS_SESSION_COUNT class-attribute instance-attribute 
MAX_STREAMING_CHANNELS_SESSION_COUNT = 'MaxStreamingChannelsSessionCount'
MAX_SUBSCRIPTION_SESSION_COUNT class-attribute instance-attribute 
MAX_SUBSCRIPTION_SESSION_COUNT = 'MaxSubscriptionSessionCount'
MAX_TRANSACTION_COUNT class-attribute instance-attribute 
MAX_TRANSACTION_COUNT = 'MaxTransactionCount'
SUPPORTS_SECONDARY_INDEX_FILTERING class-attribute instance-attribute 
SUPPORTS_SECONDARY_INDEX_FILTERING = 'SupportsSecondaryIndexFiltering'
TRANSACTION_TIMOUT_PERIOD class-attribute instance-attribute 
TRANSACTION_TIMOUT_PERIOD = 'TransactionTimeoutPeriod'
get_valid_type 
get_valid_type() -> Type[bool | int]
Source code in src/energistics/etp/v12/datatypes/protocol_capability_kind.py 
45
46
47
48
def get_valid_type(self) -> typing.Type[bool | int]:
    if self == ProtocolCapabilityKind.SUPPORTS_SECONDARY_INDEX_FILTERING:
        return bool
    return int

ServerCapabilities

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
application_name class-attribute instance-attribute 
application_name: str = Field(alias='applicationName')
application_version class-attribute instance-attribute 
application_version: str = Field(alias='applicationVersion')
contact_information class-attribute instance-attribute 
contact_information: Contact = Field(alias='contactInformation')
supported_compression class-attribute instance-attribute 
supported_compression: list[str] = Field(
    alias="supportedCompression", default_factory=list
)
supported_encodings class-attribute instance-attribute 
supported_encodings: list[str] = Field(
    alias="supportedEncodings", default_factory=lambda: ["binary"]
)
supported_formats class-attribute instance-attribute 
supported_formats: list[str] = Field(
    alias="supportedFormats", default_factory=lambda: ["xml"]
)
supported_data_objects class-attribute instance-attribute 
supported_data_objects: list[SupportedDataObject] = Field(
    alias="supportedDataObjects"
)
supported_protocols class-attribute instance-attribute 
supported_protocols: list[SupportedProtocol] = Field(alias="supportedProtocols")
endpoint_capabilities class-attribute instance-attribute 
endpoint_capabilities: Mapping[EndpointCapabilityKind, DataValue] = Field(
    alias="endpointCapabilities", default_factory=dict
)

SupportedDataObject

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
qualified_type class-attribute instance-attribute 
qualified_type: str = Field(alias='qualifiedType')
data_object_capabilities class-attribute instance-attribute 
data_object_capabilities: Mapping[DataObjectCapabilityKind, DataValue] = Field(
    alias="dataObjectCapabilities", validate_default=True, default={}
)
check_qualified_type_and_capabilities 
check_qualified_type_and_capabilities() -> Self
Source code in src/energistics/etp/v12/datatypes/supported_data_object.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@model_validator(mode="after")
def check_qualified_type_and_capabilities(self) -> typing.Self:
    # These qualified types correspond to the ones supported by the
    # open-etp-server from OSDU.
    valid_qualified_types = [
        "eml20.",
        "resqml20.",
        "resqml22.",
        "eml23.",
        "witsml21.",
        "prodml22.",
    ]
    for vqt in valid_qualified_types:
        if self.qualified_type.startswith(vqt):
            break
    else:
        raise TypeError(
            "Valid qualified types must start with one of: "
            f"{valid_qualified_types}. Append '*' for all types, or specific "
            "objects under each category."
        )

    for k, v in self.data_object_capabilities.items():
        if not isinstance(v.item, k.get_valid_type()):
            raise TypeError(
                f"The '{k}' DataObjectCapabilityKind must be of type "
                f"{k.get_valid_type()}"
            )

    return self

SupportedProtocol

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
protocol instance-attribute 
protocol: Protocol
protocol_version class-attribute instance-attribute 
protocol_version: Version = Field(
    alias="protocolVersion",
    validate_default=True,
    default=Version(major=1, minor=2, revision=0, patch=0),
)
role instance-attribute 
role: Role
protocol_capabilities class-attribute instance-attribute 
protocol_capabilities: Mapping[ProtocolCapabilityKind, DataValue] = Field(
    alias="protocolCapabilities", validate_default=True, default_factory=dict
)
check_protocol_and_role 
check_protocol_and_role() -> Self
Source code in src/energistics/etp/v12/datatypes/supported_protocol.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@model_validator(mode="after")
def check_protocol_and_role(self) -> typing.Self:
    if self.protocol == Protocol.CORE and self.role not in [
        Role.CLIENT,
        Role.SERVER,
    ]:
        raise ValueError(
            f"Protocol {self.protocol} must have a role of either '{Role.CLIENT}' "
            f"or '{Role.SERVER}'"
        )

    elif self.protocol == Protocol.CHANNEL_STREAMING and self.role not in [
        Role.PRODUCER,
        Role.CONSUMER,
    ]:
        raise ValueError(
            f"Protocol {self.protocol} must have a role of either "
            f"'{Role.CONSUMER}' or '{Role.PRODUCER}'"
        )

    elif self.protocol in [
        Protocol.CHANNEL_DATA_FRAME,
        Protocol.DISCOVERY,
        Protocol.STORE,
        Protocol.STORE_NOTIFICATION,
        Protocol.GROWING_OBJECT,
        Protocol.GROWING_OBJECT_NOTIFICATION,
        Protocol.DATA_ARRAY,
        Protocol.DISCOVERY_QUERY,
        Protocol.STORE_QUERY,
        Protocol.GROWING_OBJECT_QUERY,
        Protocol.TRANSACTION,
        Protocol.CHANNEL_SUBSCRIBE,
        Protocol.CHANNEL_DATA_LOAD,
        Protocol.DATASPACE,
        Protocol.SUPPORTED_TYPES,
    ] and self.role not in [
        Role.STORE,
        Role.CUSTOMER,
    ]:
        raise ValueError(
            f"Protocol {self.protocol} must have a role of either "
            f"'{Role.CUSTOMER}' or '{Role.STORE}'"
        )

    for k, v in self.protocol_capabilities.items():
        if not isinstance(v.item, k.get_valid_type()):
            raise TypeError(
                f"The '{k}' ProtocolCapabilityKind must be of type "
                f"{k.get_valid_type()}"
            )

    return self

Uuid

Bases: RootModel[UUIDType], ETPMetaData

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
root instance-attribute 
root: UUIDType

Version

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
major class-attribute instance-attribute 
major: int = 0
minor class-attribute instance-attribute 
minor: int = 0
revision class-attribute instance-attribute 
revision: int = 0
patch class-attribute instance-attribute 
patch: int = 0

Data array types

energistics.etp.v12.datatypes.data_array_types

__all__ module-attribute 

__all__ = [
    "DataArray",
    "DataArrayIdentifier",
    "DataArrayMetadata",
    "GetDataSubarraysType",
    "PutDataArraysType",
    "PutDataSubarraysType",
    "PutUninitializedDataArrayType",
]

DataArray

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
dimensions instance-attribute 
dimensions: list[int]
data instance-attribute 
data: AnyArray
to_numpy_array 
to_numpy_array() -> NDArray[
    bool_ | int32 | int64 | float32 | float64 | str_ | int8
]
Source code in src/energistics/etp/v12/datatypes/data_array_types/data_array.py 
27
28
29
30
31
32
33
34
35
36
def to_numpy_array(
    self,
) -> npt.NDArray[
    np.bool_ | np.int32 | np.int64 | np.float32 | np.float64 | np.str_ | np.int8
]:
    if isinstance(self.data.item, bytes):
        return np.array(np.frombuffer(self.data.item, dtype=np.int8)).reshape(
            self.dimensions
        )
    return self.data.item.values.reshape(self.dimensions)

DataArrayIdentifier

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
uri instance-attribute 
uri: str
path_in_resource class-attribute instance-attribute 
path_in_resource: str = Field(alias='pathInResource')

DataArrayMetadata

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
dimensions instance-attribute 
dimensions: list[int]
preferred_subarray_dimensions class-attribute instance-attribute 
preferred_subarray_dimensions: list[int] = Field(
    alias="preferredSubarrayDimensions", default_factory=list
)
transport_array_type class-attribute instance-attribute 
transport_array_type: AnyArrayType = Field(alias='transportArrayType')
logical_array_type class-attribute instance-attribute 
logical_array_type: AnyLogicalArrayType = Field(alias='logicalArrayType')
store_last_write class-attribute instance-attribute 
store_last_write: int = Field(alias='storeLastWrite')
store_created class-attribute instance-attribute 
store_created: int = Field(alias='storeCreated')
custom_data class-attribute instance-attribute 
custom_data: Mapping[str, DataValue] = Field(
    alias="customData", default_factory=dict
)

GetDataSubarraysType

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
uid instance-attribute 
uid: DataArrayIdentifier
starts class-attribute instance-attribute 
starts: list[int] = Field(default_factory=list)
counts class-attribute instance-attribute 
counts: list[int] = Field(default_factory=list)

PutDataArraysType

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
uid instance-attribute 
uid: DataArrayIdentifier
array instance-attribute 
array: DataArray
custom_data class-attribute instance-attribute 
custom_data: Mapping[str, DataValue] = Field(
    alias="customData", default_factory=dict
)

PutDataSubarraysType

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
uid instance-attribute 
uid: DataArrayIdentifier
data instance-attribute 
data: AnyArray
starts instance-attribute 
starts: list[int]
counts instance-attribute 
counts: list[int]

PutUninitializedDataArrayType

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
uid instance-attribute 
uid: DataArrayIdentifier
metadata instance-attribute 
metadata: DataArrayMetadata

Object

energistics.etp.v12.datatypes.object

__all__ module-attribute 

__all__ = [
    "ActiveStatusKind",
    "ContextInfo",
    "ContextScopeKind",
    "DataObject",
    "Dataspace",
    "DeletedResource",
    "Edge",
    "PutResponse",
    "RelationshipKind",
    "Resource",
]

ActiveStatusKind

Bases: StrEnum

ACTIVE class-attribute instance-attribute 
ACTIVE = 'Active'
INACTIVE class-attribute instance-attribute 
INACTIVE = 'Inactive'

ContextInfo

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
uri instance-attribute 
uri: str
depth instance-attribute 
depth: PositiveInt
data_object_types instance-attribute 
data_object_types: list[str]
navigable_edges instance-attribute 
navigable_edges: RelationshipKind
include_secondary_targets class-attribute instance-attribute 
include_secondary_targets: bool = Field(
    alias="includeSecondaryTargets", default=False
)
include_secondary_sources class-attribute instance-attribute 
include_secondary_sources: bool = Field(
    alias="includeSecondarySources", default=False
)

ContextScopeKind

Bases: StrEnum

SELF class-attribute instance-attribute 
SELF = 'self'
SOURCES class-attribute instance-attribute 
SOURCES = 'sources'
TARGETS class-attribute instance-attribute 
TARGETS = 'targets'
SOURCES_OR_SELF class-attribute instance-attribute 
SOURCES_OR_SELF = 'sourcesOrSelf'
TARGETS_OR_SELF class-attribute instance-attribute 
TARGETS_OR_SELF = 'targetsOrSelf'

DataObject

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
resource instance-attribute 
resource: Resource
format class-attribute instance-attribute 
format: str = 'xml'
blob_id class-attribute instance-attribute 
blob_id: Uuid | None = Field(alias='blobId', default=None)
data class-attribute instance-attribute 
data: bytes = b''
check_blob_and_data 
check_blob_and_data() -> Self
Source code in src/energistics/etp/v12/datatypes/object/data_object.py 
37
38
39
40
41
42
43
44
45
@model_validator(mode="after")
def check_blob_and_data(self) -> typing.Self:
    if len(self.data) == 0 and self.blob_id is None:
        raise ValueError("Either the data field or the blob-id must be populated")

    if self.blob_id is not None and len(self.data) > 0:
        raise ValueError("If the blob-id is set, then data must be empty")

    return self

Dataspace

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
uri instance-attribute 
uri: str
path class-attribute instance-attribute 
path: str = Field(default='')
store_last_write class-attribute instance-attribute 
store_last_write: int = Field(alias='storeLastWrite')
store_created class-attribute instance-attribute 
store_created: int = Field(alias='storeCreated')
custom_data class-attribute instance-attribute 
custom_data: Mapping[str, DataValue] = Field(
    alias="customData", default_factory=dict
)
check_uri_and_path 
check_uri_and_path() -> Self
Source code in src/energistics/etp/v12/datatypes/object/dataspace.py 
42
43
44
45
46
47
48
49
50
51
52
53
@model_validator(mode="after")
def check_uri_and_path(self) -> typing.Self:
    dataspace_uri = energistics.uris.DataspaceURI.from_uri(self.uri)
    path = dataspace_uri.dataspace or ""

    if path != self.path:
        raise ValueError(
            f"The path ({path}) in the dataspace uri ({self.uri}) does not match "
            f"the provided path ({self.path})"
        )

    return self

DeletedResource

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
uri instance-attribute 
uri: str
deleted_time class-attribute instance-attribute 
deleted_time: int = Field(alias='deletedTime')
custom_data class-attribute instance-attribute 
custom_data: Mapping[str, DataValue] = Field(
    alias="customData", default_factory=dict
)

Edge

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
source_uri class-attribute instance-attribute 
source_uri: str = Field(alias='sourceUri')
target_uri class-attribute instance-attribute 
target_uri: str = Field(alias='targetUri')
relationship_kind class-attribute instance-attribute 
relationship_kind: RelationshipKind = Field(alias='relationshipKind')
custom_data class-attribute instance-attribute 
custom_data: Mapping[str, DataValue] = Field(
    alias="customData", default_factory=dict
)
check_uri 
check_uri() -> Self
Source code in src/energistics/etp/v12/datatypes/object/edge.py 
47
48
49
50
51
52
@model_validator(mode="after")
def check_uri(self) -> typing.Self:
    energistics.uris.DataObjectURI.validate_uri(self.source_uri)
    energistics.uris.DataObjectURI.validate_uri(self.target_uri)

    return self
check_relationship_kind 
check_relationship_kind() -> Self
Source code in src/energistics/etp/v12/datatypes/object/edge.py 
54
55
56
57
58
59
60
61
62
@model_validator(mode="after")
def check_relationship_kind(self) -> typing.Self:
    if self.relationship_kind == RelationshipKind.BOTH:
        raise ValueError(
            f"Relationship kind can only be {RelationshipKind.PRIMARY} or "
            f"{RelationshipKind.SECONDARY} in Edge"
        )

    return self

PutResponse

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
created_contained_object_uris class-attribute instance-attribute 
created_contained_object_uris: list[str] = Field(
    alias="createdContainedObjectUris", default_factory=list
)
deleted_contained_object_uris class-attribute instance-attribute 
deleted_contained_object_uris: list[str] = Field(
    alias="deletedContainedObjectUris", default_factory=list
)
joined_contained_object_uris class-attribute instance-attribute 
joined_contained_object_uris: list[str] = Field(
    alias="joinedContainedObjectUris", default_factory=list
)
unjoined_contained_object_uris class-attribute instance-attribute 
unjoined_contained_object_uris: list[str] = Field(
    alias="unjoinedContainedObjectUris", default_factory=list
)

RelationshipKind

Bases: StrEnum

PRIMARY class-attribute instance-attribute 
PRIMARY = 'Primary'
SECONDARY class-attribute instance-attribute 
SECONDARY = 'Secondary'
BOTH class-attribute instance-attribute 
BOTH = 'Both'

Resource

Bases: ETPBaseModel

avro_schema class-attribute 
avro_schema: AvroSchemaType
full_name class-attribute 
full_name: str
model_config class-attribute instance-attribute 
model_config = ConfigDict(populate_by_name=True)
uri instance-attribute 
uri: str
alternate_uris class-attribute instance-attribute 
alternate_uris: list[str] = Field(alias='alternateUris', default_factory=list)
name instance-attribute 
name: str
source_count class-attribute instance-attribute 
source_count: int | None = Field(alias='sourceCount', default=None)
target_count class-attribute instance-attribute 
target_count: int | None = Field(alias='targetCount', default=None)
last_changed class-attribute instance-attribute 
last_changed: int = Field(alias='lastChanged')
store_last_write class-attribute instance-attribute 
store_last_write: int = Field(alias='storeLastWrite')
store_created class-attribute instance-attribute 
store_created: int = Field(alias='storeCreated')
active_status class-attribute instance-attribute 
active_status: ActiveStatusKind = Field(alias='activeStatus')
custom_data class-attribute instance-attribute 
custom_data: Mapping[str, DataValue] = Field(
    alias="customData", default_factory=dict
)
check_uri 
check_uri() -> Self
Source code in src/energistics/etp/v12/datatypes/object/resource.py 
63
64
65
66
67
@model_validator(mode="after")
def check_uri(self) -> typing.Self:
    energistics.uris.DataObjectURI.validate_uri(self.uri)

    return self