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.

"""
This module defines the model and functions related to a publication.
"""

import enum
import logging
import time
import uuid
from typing import TYPE_CHECKING, List, Optional, Tuple, cast

from here.platform.exceptions import PublicationException
from here.platform.model.version_dependency import VersionDependency

if TYPE_CHECKING:
from here.platform.layer import Layer

logger = logging.getLogger(name)

[docs]
class PublicationState(enum.Enum):
"""Enum class for publication states."""

INITIALIZED = "initialized"
SUCCEEDED = "succeeded"
FAILED = "failed"
CANCELLED = "cancelled"
EXPIRED = "expired"
SUBMITTED = "submitted"

[docs]
class Publication:
"""
A publication is needed to write to a catalog.

Create a publication via the Catalog.init_publication function,
specifying to which layers you want to write to. The request
is validated and a Publication is returned when it can be satisfied.

Use the publication in the set_* and write_* method of Layer classes,
for layers that need a publication to operate.

When writing is complete, close the publication by either
calling the complete function, or cancel. Note however that
not all the layer types support cancelling and rolling back a publication.
"""

def init(
self,
catalog,
layers: List["Layer"],
dependencies: Optional[List[VersionDependency]] = None,
publication_info: Optional[Tuple[str, Optional[int]]] = None,
):
"""
Create a new publication.

:param catalog: the catalog.
:param layers: the list of layers affected.
:param dependencies: the dependencies of the new version.
:param publication_info: ID and catalog version for an existing publication.
"""
self.catalog = catalog
self.layers = layers
self.dependencies = dependencies
self.billing_tag = self.catalog.billing_tag
self._validate_layer_combinations()
self._validate_layer_exist()
if publication_info:
self.publication_id = publication_info[0]
self.catalog_version = publication_info[1]
else:
publication_details = self._initialize_publication()
self.publication_id = publication_details.get("id") or str(uuid.uuid4())

version is unset in case no versioned layers are involved

self.catalog_version = cast(Optional[int], publication_details.get("catalogVersion"))
self.is_active = True
self._polling_wait = self.catalog.platform.application_config.polling_wait

def enter(self):
"""
Entry point for publication context manager
"""
return self

def exit(self, exc_type, exc_value, exc_tb):
"""
Exit for publication context manager
"""
if exc_type is not None:
try:
self.cancel()
except PublicationException as e:
logger.error(
f"Error occured when canceling publication with"
f"Publication id: {e.publication_id} state: {e.publication_state}"
)
else:
self.complete()

def _validate_layer_combinations(self):
"""
Validate layer combinations.
"""
iseq = iter(self.layers)
first = next(iseq)
first_type = type(first)
if not (first.is_versioned() or first.is_volatile() or first.is_stream()):
raise ValueError(
f"Initialize publication failed: Publication does not support {first_type}"
)
validation_status = all((type(x) is first_type) for x in iseq)
if validation_status is False:
raise ValueError(
"Initialize publication failed: Layer types for publication are different."
"A publication can only consist of layer IDs that have the same layer type."
)

def _validate_layer_exist(self):
"""
Validate layer exists.
"""
validation_status = all(self.catalog.has_layer(x.id) for x in self.layers)
if validation_status is False:
raise ValueError(
"Initialize publication failed: "
"Some layers do not exist or do not belong to the catalog."
)

def _initialize_publication(self) -> dict:
"""
Initialize the publication.

:return: dict with publication response
"""
deps = (
[{"hrn": d.hrn, "version": d.version, "direct": d.direct} for d in self.dependencies]
if self.dependencies
else None
)
layer_ids = [layer.id for layer in self.layers]
body = {"layerIds": layer_ids, "versionDependencies": deps}
return cast(
dict,
self.catalog._data_publish_api.init_publication(
body=body, billing_tag=self.billing_tag
),
)

def _poll_publication_status(self, expected_state: PublicationState):
"""
Poll the publication status until the expected state has reached.

:param expected_state: the publication state to match.
:raises PublicationException: is raised in case the publication is not
finalized successfully.
"""
while True:
logging.debug(f"status polling wait {self._polling_wait} sec.")
time.sleep(self._polling_wait)
status_resp = self.catalog._data_publish_api.get_publication(
publication_id=self.publication_id, billing_tag=self.billing_tag
)
current_state = PublicationState(status_resp["details"]["state"])
logger.debug(f"Publication id: {self.publication_id} state: {current_state.value}")
if current_state in [
PublicationState.SUCCEEDED,
PublicationState.FAILED,
PublicationState.CANCELLED,
PublicationState.EXPIRED,
]:
logger.info(
f"Publication for id: {self.publication_id}"
f"finished with state: {current_state}"
)
if current_state == expected_state:
self.is_active = False
return
else:
raise PublicationException(str(self.publication_id), current_state)

[docs]
def complete(self):
"""
Complete the publication succesfully.

After calling this function, the object can't be used anymore.
Pending transactions are committed.
"""
if self.is_active:
self.catalog._data_publish_api.submit_publication(
publication_id=self.publication_id, billing_tag=self.billing_tag
)
self._poll_publication_status(PublicationState.SUCCEEDED)

[docs]
def cancel(self, strict: bool = False):
"""
Cancel the publication, usually as a result of errors.

After calling this function, the object can't be used anymore.
Pending transactions are rolled back, but data written
to APIs that are not transaction-based stay written.
:param strict: True to require that the publication exists, False to allow it to have
already been cancelled.
"""
if self.is_active:
if not self.catalog._data_publish_api.cancel_publication(
publication_id=self.publication_id, billing_tag=self.billing_tag, strict=strict
):
return
self._poll_publication_status(PublicationState.CANCELLED)