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 Json Parser module
"""

import io
import json
import os
import zipfile
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Union

import jsonschema
from google.protobuf.message import Message
from here.platform.exceptions import SchemaException
from here.platform.schema.parser import Parser
from referencing import Registry, Resource
from referencing.jsonschema import DRAFT7

[docs]
class JsonParser(Parser):
"""A class for parsing Json data of a catalog layer partition."""

_SCHEMA_ARTIFACTS_PATH = TemporaryDirectory(
prefix="pysdk-schema-", ignore_cleanup_errors=True
).name

def init(self, schema_pkg_file: Union[str, Path, io.BytesIO]):
"""
Initialize a JsonParser instance.

:param schema_pkg_file: the schema package zip-file as a str, Path or file-like object
:raises SchemaException: if the required meta information is missing in schema package
"""
layermanifest, = Parser.extract_layer_info(schema_pkg_file)
if not layer_manifest:
raise SchemaException(
f"Invalid schema. Layer manifest file {Parser.LAYER_MANIFEST} "
"not found in schema package file" # noqa: E713
)

self.main_partition_message = layer_manifest["main"]["message"]
self.main_partition_message_source = layer_manifest["main"]["source"]

if not self.main_partition_message:
raise SchemaException(
f"Unable to determine the main message defined in the "
f"{Parser.LAYER_MANIFEST} for schema artifact"
)

if not self.main_partition_message_source:
raise SchemaException(
f"Unable to determine the main message source defined in the "
f"{Parser.LAYER_MANIFEST} for schema artifact"
)

self._schema_pgk = schema_pkg_file
self._schema: dict =
self._schema_store: dict =
self._registry: Registry = None
self._extract_json_schema()

def _extract_json_schema(self) -> dict:
"""
Unpack input zipfile, locate the main message source
and initialize the json-schema object.

:raises SchemaException: if main schema could not be loaded
:returns: json-schema object
"""

Unpack schema zip to temporary directory

if not self._schema:
input_zipfile = zipfile.ZipFile(self._schema_pgk)
input_zipfile.extractall(path=self._SCHEMA_ARTIFACTS_PATH)

main_message_path = os.path.join(
self._SCHEMA_ARTIFACTS_PATH, self.main_partition_message_source
)
schema_dir = os.path.dirname(main_message_path)

if not self._schema_store:

Load schemas from disk

for root, _, files in os.walk(schema_dir):
for file in files:
if file.endswith(".json"):
full_path = os.path.join(root, file)
rel_path = Path(os.path.relpath(full_path, schema_dir)).as_posix()
with open(full_path, "r", encoding="utf-8") as f:
schema = json.load(f)
if Path(full_path).as_posix() == Path(main_message_path).as_posix():
main_schema = schema

Adjust the start level of the main schema

message_path = self.main_partition_message.split("/")
for i in range(0, len(message_path)):
if message_path[i] == "#":
continue
main_schema = main_schema[message_path[i]]
self._schema_store[rel_path] = Resource.from_contents(
contents=schema, default_specification=DRAFT7
)
self._schema = main_schema

if not self._schema:
raise SchemaException(f"Unable to load the main schema {main_message_path}")

Build a registry of sub-schemas

if not self._registry:
self._registry = Registry(self._schema_store)

return self._schema

[docs]
def validate_json_schema(self, data: dict):
"""
Validates the given json instance against the defined schema.

:param data: json instance to be validated
"""
jsonschema.validate(instance=data, schema=self._schema, registry=self._registry)

[docs]
def parse_message(self, message: bytes) -> Union[Message, dict]:
"""
Parse a message and returns the corresponding partition class instance.

:param message: message received as blob
:return: parsed message
"""
decoded = json.loads(message)
assert isinstance(decoded, dict)
self.validate_json_schema(decoded)
return decoded

[docs]
def serialize_bytes(self, data: Union[Message, dict]) -> bytes:
"""
Parse a Json message and returns the bytes.

:param data: Json message
:return: bytes
"""
assert isinstance(data, dict)
self.validate_json_schema(data)
message_bytes = json.dumps(data).encode("utf-8")
assert isinstance(message_bytes, bytes)
return message_bytes