Source code for tdw_catalog.metadata.field

from datetime import date, datetime
import json
from typing import Generic, List, Optional, TypeVar
import uuid
from tdw_catalog.errors import CatalogFailedPreconditionException, CatalogPermissionDeniedException
import tdw_catalog.metadata_template.contract_owner as template_contract_owner
import tdw_catalog.metadata_template.currency as template_currency
import tdw_catalog.metadata_template.data_cost as template_data_cost
import tdw_catalog.metadata_template.date_field as template_date
import tdw_catalog.metadata_template.decimal as template_decimal
import tdw_catalog.metadata_template.license_expiry as template_license_expiry
import tdw_catalog.metadata_template.link as template_link
import tdw_catalog.metadata_template.field as field
import tdw_catalog.metadata_template.number as template_number
import tdw_catalog.metadata_template.organization_member as template_organization_member
import tdw_catalog.metadata_template.organization_team as template_organization_team
import tdw_catalog.metadata_template.point_in_time as template_point_in_time
import tdw_catalog.metadata_template.point_of_contact as template_point_of_contact
import tdw_catalog.metadata_template.text as template_text
import tdw_catalog.metadata_template.alias as template_alias
import tdw_catalog.metadata_template.list_field as template_list

from tdw_catalog.utils import MetadataFieldType

T = TypeVar('T')

from typing import TYPE_CHECKING
if TYPE_CHECKING:
    import tdw_catalog.dataset as dataset


class MetadataField(Generic[T]):
    """
    The base type for all custom metadata fields which can be added to a :class:`.Dataset`

    Attributes
    ----------
    key : str
        A key for this field, which must be unique among all metadata fields in a :class:`.Dataset`
    value : Optional[T]
        An optional value for :class:`.MetadataField`\\ s produced from this :class:`.MetadataField`.
    """
    _key: str
    _value: Optional[T]

    def __init__(
        self,
        key: str,
        value: Optional[T],
    ) -> None:
        self._key = key
        self._value = value

    @property
    def key(self):
        return self._key

    @key.setter
    def key(self, key: str):
        self._key = key

    @property
    def value(self):
        return self._value

    @value.setter
    def value(self, value: Optional[T]):
        self._value = value

    @property
    def is_recommended(self) -> bool:
        return False

    def serialize(self) -> dict:
        res = {
            "key": self._key,
        }
        return res

    def __getitem__(self, item):
        return getattr(self, item)

    @classmethod
    def deserialize(cls, data: dict, dataset: 'dataset.Dataset') -> None:

        # if field_type is missing, assume string
        if "field_type" not in data:
            data["field_type"] = MetadataFieldType.FT_STRING

        if "value" not in data:
            data["value"] = ""

        # if field_type comes back as a string, convert to int enum
        if isinstance(data["field_type"], str):
            data["field_type"] = MetadataFieldType[data["field_type"]]

        if "recommended_type" in data and data["recommended_type"] == "cost":
            currencyDict = json.loads(
                data["value"]) if "value" in data else {}
            import tdw_catalog.metadata.data_cost as data_cost
            return data_cost.Field(
                key=data["key"],
                currency=currencyDict["currency"] if "currency" in currencyDict else "",
                amount=currencyDict["value"] if "value" in currencyDict else 0,
            )

        elif "recommended_type" in data and data["recommended_type"] == "poc":
            import tdw_catalog.metadata.point_of_contact as point_of_contact
            try:
                return point_of_contact.Field(
                    key=data["key"],
                    value=dataset.organization.get_member(
                        user_id=data["value"]),
                )
            except:
                return point_of_contact.Field(
                    key=data["key"],
                    value=None,
                )

        elif "recommended_type" in data and data[
                "recommended_type"] == "expiryDate":
            import tdw_catalog.metadata.license_expiry as license_expiry

            return license_expiry.Field(
                key=data["key"],
                value=datetime.strptime(data["value"][0:10], "%Y-%m-%d").date() if "value" in data and len(data["value"]) == 10 else None,
            )

        elif "recommended_type" in data and data[
                "recommended_type"] == "contractOwner":
            import tdw_catalog.metadata.contract_owner as contract_owner
            try:
                return contract_owner.Field(
                    key=data["key"],
                    value=dataset.organization.get_team(team_id=data["value"]),
                )
            except:
                return contract_owner.Field(
                    key=data["key"],
                    value=None,
                )

        # determine field type and instantiate appropriate class
        elif data["field_type"] == MetadataFieldType.FT_STRING:
            import tdw_catalog.metadata.text as text
            return text.Field(
                key=data["key"],
                value=data["value"],
            )

        elif data["field_type"] == MetadataFieldType.FT_INTEGER:
            import tdw_catalog.metadata.number as number
            return number.Field(
                key=data["key"],
                value=int(data["value"]),
            )

        elif data["field_type"] == MetadataFieldType.FT_DECIMAL:
            import tdw_catalog.metadata.decimal as decimal
            return decimal.Field(
                key=data["key"],
                value=float(data["value"]),
            )

        elif data["field_type"] == MetadataFieldType.FT_DATE:
            import tdw_catalog.metadata.date_field as date_field
            try:
                d = datetime.strptime(data["value"][0:10], "%Y-%m-%d").date()
                return date_field.Field(
                    key=data["key"],
                    value=d,
                )
            except ValueError:
                return date_field.Field(
                    key=data["key"],
                    value=None,
                )
        elif data["field_type"] == MetadataFieldType.FT_DATETIME:
            import tdw_catalog.metadata.point_in_time as point_in_time
            try:
                dt = datetime.fromisoformat(data["value"])
                return point_in_time.Field(
                    key=data["key"],
                    value=dt,
                )
            except ValueError:
                return point_in_time.Field(
                    key=data["key"],
                    value=None,
                )
        elif data["field_type"] == MetadataFieldType.FT_DATASET:
            import tdw_catalog.metadata.linked_dataset as linked_dataset
            try:
                return linked_dataset.Field(
                    key=data["key"],
                    value=dataset.organization.get_dataset(id=data["value"]),
                )
            except:
                return linked_dataset.Field(
                    key=data["key"],
                    value=None,
                )
        elif data["field_type"] == MetadataFieldType.FT_URL:
            import tdw_catalog.metadata.link as link
            return link.Field(
                key=data["key"],
                value=data["value"],
            )

        elif data["field_type"] == MetadataFieldType.FT_USER:
            import tdw_catalog.metadata.organization_member as organization_member
            try:
                return organization_member.Field(
                    key=data["key"],
                    value=dataset.organization.get_member(
                        user_id=data["value"]),
                )
            except:
                return organization_member.Field(
                    key=data["key"],
                    value=None,
                )
        elif data["field_type"] == MetadataFieldType.FT_ATTACHMENT:
            import tdw_catalog.metadata.attachment as attachment
            return attachment.Field(key=data["key"], value=data["value"])

        elif data["field_type"] == MetadataFieldType.FT_LIST:
            import tdw_catalog.metadata.list_field as list_field
            return list_field.Field(
                key=data["key"],
                value=data["value"],
                list_items=data["listValueList"]
                if "listValueList" in data else None,
            )

        elif data["field_type"] == MetadataFieldType.FT_CURRENCY:
            currencyDict = json.loads(data["value"])
            import tdw_catalog.metadata.currency as currency
            return currency.Field(
                key=data["key"],
                currency=currencyDict["currency"],
                amount=currencyDict["value"],
            )

        elif data["field_type"] == MetadataFieldType.FT_TEAM:
            import tdw_catalog.metadata.organization_team as organization_team
            try:
                return organization_team.Field(
                    key=data["key"],
                    value=dataset.organization.get_team(team_id=data["value"]),
                )
            except:
                return organization_team.Field(
                    key=data["key"],
                    value=None,
                )

        elif data["field_type"] == MetadataFieldType.FT_ALIAS:
            import tdw_catalog.metadata.alias as alias
            return alias.Field(
                key=data["key"],
                value=data["value"],
            )


class TemplatedMetadataField(Generic[T]):
    """
    A wrapper class for :class:`.MetadataField`\\ s that come from an attached :class:`.MetadataTemplate`. :class:`.TemplatedMetadataField`\\ s `key`\\ s can not be changed on a :class:`.Dataset`.
    """
    _field: MetadataField[T]

    def __init__(self, field: MetadataField[T]):
        self._field = field

    @property
    def key(self):
        return self._field._key

    @key.setter
    def key(self, key: str):
        raise CatalogPermissionDeniedException(
            message=
            "Keys of fields on a MetadataTemplate can only be changed on the MetadataTemplate directly"
        )

    @property
    def value(self):
        return self._field._value

    @value.setter
    def value(self, value: Optional[T]):
        self._field._value = value


class RecommendedMetadataField(MetadataField[T]):
    """
    A subclass of :class:`.MetadataField` representing one of `license expiry`\\ , `point of contact`\\ , `data cost`\\ , or `contract owner`. The `key` of a :class:`.RecommendedMetadataField` cannot be changed
    """

    def __init__(self, key: str, value: T):
        super().__init__(key, value)

    @property
    def key(self):
        return super().key

    @key.setter
    def key(self, key: str):
        raise CatalogPermissionDeniedException(
            message="Keys of recommended metadata fields may not be changed")

    @property
    def is_recommended() -> bool:
        return True


def _deserialize_metadata_fields(
        data: List[dict],
        dataset: 'dataset.Dataset') -> Optional[List[MetadataField]]:
    if data is None:
        return None
    return list(map(lambda d: MetadataField.deserialize(d, dataset), data))


def _convert_template_field(
        field: field.MetadataTemplateField[T]) -> MetadataField[T]:
    if isinstance(field, template_contract_owner.Field):
        import tdw_catalog.metadata.contract_owner as contract_owner
        return contract_owner.Field(key=field.key, value=field.default_value)

    if isinstance(field, template_point_of_contact.Field):
        import tdw_catalog.metadata.point_of_contact as point_of_contact
        return point_of_contact.Field(key=field.key, value=field.default_value)

    if isinstance(field, template_license_expiry.Field):
        import tdw_catalog.metadata.license_expiry as license_expiry
        return license_expiry.Field(key=field.key, value=field.default_value)

    if isinstance(field, template_data_cost.Field):
        currencyDict = field["default_value"] if field[
            "default_value"] is not None else None
        import tdw_catalog.metadata.currency as currency
        import tdw_catalog.metadata.data_cost as data_cost
        currency = currencyDict.currency if currencyDict is not None else ""
        amount = currencyDict.value if currencyDict is not None else 0
        return data_cost.Field(key=field.key, currency=currency, amount=amount)

    if isinstance(field, template_text.Field):
        import tdw_catalog.metadata.text as text
        return text.Field(key=field.key, value=field.default_value)

    if isinstance(field, template_number.Field):
        import tdw_catalog.metadata.number as number
        return number.Field(key=field.key, value=field.default_value)

    if isinstance(field, template_decimal.Field):
        import tdw_catalog.metadata.decimal as decimal
        return decimal.Field(key=field.key, value=field.default_value)

    if isinstance(field, template_link.Field):
        import tdw_catalog.metadata.link as link
        return link.Field(key=field.key, value=field.default_value)

    import tdw_catalog.metadata_template.linked_dataset as template_linked_dataset
    if isinstance(field, template_linked_dataset.Field):
        import tdw_catalog.metadata.linked_dataset as linked_dataset
        return linked_dataset.Field(key=field.key, value=field.default_value)

    if isinstance(field, template_date.Field):
        import tdw_catalog.metadata.date_field as date_field
        return date_field.Field(key=field.key, value=field.default_value)

    if isinstance(field, template_point_in_time.Field):
        import tdw_catalog.metadata.point_in_time as point_in_time
        return point_in_time.Field(key=field.key, value=field.default_value)

    if isinstance(field, template_organization_member.Field):
        import tdw_catalog.metadata.organization_member as organization_member
        return organization_member.Field(key=field.key,
                                         value=field.default_value)

    if isinstance(field, template_organization_team.Field):
        import tdw_catalog.metadata.organization_team as organization_team
        return organization_team.Field(key=field.key,
                                       value=field.default_value)

    if isinstance(field, template_alias.Field):
        import tdw_catalog.metadata.alias as alias
        return alias.Field(key=field.key, value=field.default_value)

    if isinstance(field, template_list.Field):
        import tdw_catalog.metadata.list_field as list_field
        return list_field.Field(key=field.key,
                                value=field.default_value,
                                list_items=field.list_items)


def _check_field_match(field: MetadataField,
                       templated_field: 'field.MetadataTemplateField') -> bool:
    if isinstance(templated_field, template_contract_owner.Field):
        import tdw_catalog.metadata.contract_owner as contract_owner
        return isinstance(field, contract_owner.Field)

    if isinstance(field, template_point_of_contact.Field):
        import tdw_catalog.metadata.point_of_contact as point_of_contact
        return isinstance(field, point_of_contact.Field)

    if isinstance(field, template_license_expiry.Field):
        import tdw_catalog.metadata.license_expiry as license_expiry
        return isinstance(field, license_expiry.Field)

    if isinstance(field, template_data_cost.Field):
        import tdw_catalog.metadata.data_cost as data_cost
        return isinstance(field, data_cost.Field)

    if isinstance(field, template_text.Field):
        import tdw_catalog.metadata.text as text
        return isinstance(field, text.Field)

    if isinstance(field, template_number.Field):
        import tdw_catalog.metadata.number as number
        return isinstance(field, number.Field)

    if isinstance(field, template_decimal.Field):
        import tdw_catalog.metadata.decimal as decimal
        return isinstance(field, decimal.Field)

    if isinstance(field, template_link.Field):
        import tdw_catalog.metadata.link as link
        return isinstance(field, link.Field)

    import tdw_catalog.metadata_template.linked_dataset as template_linked_dataset
    if isinstance(field, template_linked_dataset.Field):
        import tdw_catalog.metadata.linked_dataset as linked_dataset
        return isinstance(field, linked_dataset.Field)

    if isinstance(field, template_date.Field):
        import tdw_catalog.metadata.date_field as date_field
        return isinstance(field, date_field.Field)

    if isinstance(field, template_point_in_time.Field):
        import tdw_catalog.metadata.point_in_time as point_in_time
        return isinstance(field, point_in_time.Field)

    if isinstance(field, template_organization_member.Field):
        import tdw_catalog.metadata.organization_member as organization_member
        return isinstance(field, organization_member.Field)

    if isinstance(field, template_organization_team.Field):
        import tdw_catalog.metadata.organization_team as organization_team
        return isinstance(field, organization_team.Field)

    if isinstance(field, template_currency.Field):
        import tdw_catalog.metadata.currency as currency
        return isinstance(field, currency.Field)

    if isinstance(field, template_alias.Field):
        import tdw_catalog.metadata.alias as alias
        return isinstance(field, alias.Field)

    if isinstance(field, template_list.Field):
        import tdw_catalog.metadata.list_field as list_field
        return isinstance(field, list_field.Field)