Copyright (C) 2020-2022 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 catalog abstraction."""

import enum
from typing import Any, Dict, List, Optional

from here.platform.utils import JsonDictDocument

[docs]
class Coverage(JsonDictDocument):
"""
The geographic area that this catalog or layer covers
"""

@property
def admin_areas(self) -> List[str]:
"""A list of ISO 3166 two-letter codes for countries and regions"""
return [str(t) for t in self.json.get("adminAreas", [])]

[docs]
class Creator(JsonDictDocument):
"""
Details of the user or the application that initially created the catalog.
"""

@property
def id(self) -> str:
"""The unique ID of the user or application that initially created the catalog"""
return str(self.json["id"])

[docs]
class Organisation(JsonDictDocument):
"""
Organisation of the customer that created the catalog.
"""

@property
def id(self) -> str:
"""The ID of the customer organisation relating to this catalog"""
return str(self.json["id"])

[docs]
class Owner(JsonDictDocument):
"""
Details of the owner of the catalog.
"""

@property
def creator(self) -> Creator:
"""The user or application that initially created the catalog"""
return Creator(self.json["creator"])

@property
def organisation(self) -> Organisation:
"""The organisation of the customer that created the catalog"""
return Organisation(self.json["organisation"])

[docs]
class Region(JsonDictDocument):
"""
The replication region of the catalog, including each region's role
"""

@property
def id(self) -> str:
"""The ID of the region"""
return str(self.json["id"])

@property
def role(self) -> str:
"""Indicates whether the region is a primary or failover region"""
return str(self.json["role"])

[docs]
class Replication(JsonDictDocument):
"""
The replication set for the catalog.
"""

@property
def regions(self) -> List[Region]:
"""A list of the catalog's replication regions and each region's role"""
return [Region.from_dict(t) for t in self.json["regions"]]

[docs]
class AutomaticVersionDeletion(JsonDictDocument):
"""
Specifies the number of versions to keep for the catalog.
"""

@property
def number_of_versions_to_keep(self) -> int:
"""The number of versions to keep."""
return int(self.json["numberOfVersionsToKeep"])

[docs]
class Notifications(JsonDictDocument):
"""Indicates whether or not to notify each time the version of the catalog changes"""

@property
def enabled(self) -> bool:
"""Determines if the notifications are enabled for the catalog"""
return bool(self.json["enabled"])

[docs]
class PartitioningScheme(enum.Enum):
"""Enum class for the name of the partitioning scheme for the layer."""

GENERIC = "generic"
HERE_TILE = "heretile"
NO_PARTITIONING = "nopartitioning"

[docs]
class Partitioning(JsonDictDocument):
"""Describes the way in which data is partitioned within the layer"""

@property
def scheme(self) -> PartitioningScheme:
"""The name of the partitioning scheme for the layer"""
return PartitioningScheme(self.json["scheme"])

@property
def tile_levels(self) -> Optional[int]:
"""Quadtree tile levels which contain data partitions."""
return self.json.get("tileLevels")

[docs]
class VolumeType(enum.Enum):
"""Enum class for the volume type to be used for storing the layer's data content.."""

DURABLE = "durable"
VOLATILE = "volatile"

[docs]
class MaxMemoryPolicy(enum.Enum):
"""Enum for the keys eviction policy when the memory limit for volatile layer is reached."""

FAIL_ON_WRITE = "failOnWrite"
REPLACE_LESS_RECENTLY_USED_KEY = "replaceLessRecentlyUsedKey"

[docs]
class Volume(JsonDictDocument):
"""Describes the volume to be used for storing the layer's data content"""

def init(
self,
volume_type: str,
max_memory_policy: Optional[str] = None,
):
"""
Create a new volume.
"""
json_dict = (
{"volumeType": volume_type}
if volume_type == VolumeType.DURABLE.value
else {"volumeType": volume_type, "maxMemoryPolicy": max_memory_policy if max_memory_policy is not None else MaxMemoryPolicy.FAIL_ON_WRITE.value,}
)
super().init(json_dict)

[docs]
class DurableVolume(Volume):
"""Describes a catalog durable volume to be used for storing the layer's data content"""

def init(self, volume: Dict[str, Any]):
"""
Create a new durable volume.
"""
Volume.init(self, volume["volumeType"])

@property
def volume_type(self) -> VolumeType:
"""The type of volume used for storing the layer's data content"""
return VolumeType(self.json["volumeType"])

[docs]
class VolatileVolume(Volume):
"""Describes a catalog volatile volume to be used for storing the layer's data content."""

def init(self, volume: Dict[str, Any]):
"""
Create a new volatile volume.
"""
Volume.init(self, volume["volumeType"], volume["maxMemoryPolicy"])

@property
def volume_type(self) -> VolumeType:
"""The type of volume used for storing the layer's data content"""
return VolumeType(self.json["volumeType"])

@property
def max_memory_policy(self) -> MaxMemoryPolicy:
"""Defines a keys eviction policy when the memory limit for volatile layer is reached"""
return MaxMemoryPolicy(self.json["maxMemoryPolicy"])