Copyright (C) 2020-2023 HERE Global B.V. and its affiliate(s).

All rights reserved.

This software and other materials contain proprietary information

controlled by HERE and are protected by applicable copyright legislation.

Any use and utilization of this software and other materials and

disclosure to any third parties is conditional upon having a separate

agreement with HERE for the access, use, utilization or disclosure of this

software. In the absence of such agreement, the use of the software is not

allowed.

"""
HERE Platform Schema module

A schema contains the protobuf parsed schema and provides helper methods for
working with the schema and decoding data to a protobuf Message or Python Dict.
"""

import webbrowser
from typing import List, Union

from google.protobuf.message import Message
from here.platform.config import PlatformConfig
from here.platform.schema.parser import Parser
from here.platform.schema.protobuf_parser import ProtobufParser

[docs]
class Schema:
"""A schema registered on the HERE platform."""

def init(self, schema_hrn: str, parser: Parser, platform_config: PlatformConfig):
"""Instantiate a Schema with HRN and parser."""
self.schema_hrn = schema_hrn
self._parser = parser
self._platform_config = platform_config

@property
def supported_content_types(self) -> List[str]:
"""The content types this schema describes."""
return [
"application/protobuf",
"application/x-protobuf",
"application/json",
"application/geo+json",
"application/vnd.geo+json",
]

@property
def parser(self) -> Parser:
"""To return parser object."""
return self._parser

[docs]
def decode_blob(self, data: bytes) -> Union[Message, dict]:
"""
Decode an encoded blob and returns
the corresponding partition main message.

:param data: encoded content
:return: decoded message
"""
return self._parser.parse_message(data)

[docs]
def decode_blob_by_fully_qualified_message_name(
self, data: bytes, fully_qualified_message_name: str
) -> Union[Message, dict]:
"""
Decode a protobuf-encoded blob by its fully_qualified_message_name class
and returns the corresponding partition main message.

:param data: encoded protobuf content
:param fully_qualified_message_name: class name and its fully qualified message name
:return: decoded protobuf message
"""
assert isinstance(self._parser, ProtobufParser)
return self._parser.parse_message_by_class(data, fully_qualified_message_name)

[docs]
def encode_blob(self, data: Union[Message, dict]) -> bytes:
"""
Encode Message and returns the corresponding bytes.

:param data: encoded content
:return: encoded bytes
"""
return self._parser.serialize_bytes(data)

[docs]
def open_in_portal(self):
"""Open the schema page on the HERE platform portal."""
portal_url = self._platform_config.portal_url
if portal_url is None:
raise ValueError("here_platform_portal_url is not present in configuration.")
webbrowser.open_new(f"{portal_url}/data/schemas/{self.schema_hrn}")