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.

"""
A registry for parsed protobuf schemas, so that they can be managed, cached, etc.
"""
import io
from typing import Dict, Optional

from here.platform.api.artifact_api import ArtifactApi
from here.platform.config import PlatformConfig
from here.platform.schema import Schema
from here.platform.schema.ParserFactory import ParserFactory

[docs]
class SchemaRegistry:
"""HERE platform schema registry."""

def init(self, platform_config: PlatformConfig, artifact_api: ArtifactApi):
"""Instantiate HERE platform schema registry."""
self._registry: Dict[str, Schema] =
self._platform_config = platform_config
self._artifact_api = artifact_api

[docs]
def has_schema(self, schema_hrn: str) -> bool:
"""
Check whether a given schema has already been registered.

:param schema_hrn: Schema HRN
:return: a bool flag indicating if schema is already registered.
"""
return schema_hrn in self._registry

[docs]
def get_schema(self, schema_hrn: str) -> Schema:
"""
Return the schema by the provided schema HRN identifier.
The schema must be already available in the schema registry
by registering it or by obtaining it and registering it.

:param schema_hrn: Schema HRN
:return: Schema instance for the given hrn
:raises ValueError: in case the requested schema is not available
"""
if self.has_schema(schema_hrn):
return self._registry[schema_hrn]
else:
raise ValueError(
f"Schema {schema_hrn} not available in the schema registry" # noqa: E713
)

[docs]
def register_schema(self, schema_hrn: str, schema: Schema):
"""
Register a new schema in the registry.

If a schema with the same HRN is present, it's discarded.
This does not upload the schema to the Artifact service.

:param schema_hrn: the HRN of the schema
:param schema: the schema to be registered
"""
self._registry[schema_hrn] = schema

[docs]
def obtain_and_register_schema(
self, schema_hrn: str, content_type: Optional[str] = None
) -> Schema:
"""
Obtain a schema from the Artifact service and register it
in the schema registry. Return the registered schema.
If the schema is already available, it's simply returned.

:param schema_hrn: the HRN of the schema
:param content_type: The MIME type of the blobs stored in the layer
:return: the schema just obtained or already available
"""

if schema_hrn in self._registry:
return self._registry[schema_hrn]

art_info = self._artifact_api.get_schema(schema_hrn)
addr = [var for var in art_info["variants"] if var["id"] == "ds"][0]["url"]

art_components = addr.split("/")
art_hrn = "/".join(art_components[2:-1])
file_name = art_components[-1]
schema_package = self._artifact_api.get_artifact_file(
artifact_hrn=art_hrn, file_name=file_name
)
schema_pkg_file = io.BytesIO(schema_package)
parser = ParserFactory.get_parser(schema_pkg_file, content_type)

schema = Schema(schema_hrn, parser, self._platform_config)
self._registry[schema_hrn] = schema
return schema

[docs]
def load_and_register_schema(
self, schema_file, schema_hrn, content_type: Optional[str] = None
) -> Schema:
"""
Register Schema object by given zip artifact and schema hrn .

:param schema_file: zip artifact
:param schema_hrn: Schema HRN
:param content_type: The MIME type of the blobs stored in the layer
:return: Schema instance for the given hrn
"""

parser = ParserFactory.get_parser(schema_file, content_type)
schema = Schema(schema_hrn, parser, self._platform_config)
self.register_schema(schema_hrn, schema)
return schema