Source code for tdw_catalog.metadata.editor
from datetime import date, datetime
import mimetypes
import os
from typing import List, Optional, Union
import tdw_catalog.dataset as dataset
from tdw_catalog.errors import CatalogFailedPreconditionException, CatalogInvalidArgumentException, CatalogNotFoundException, CatalogPermissionDeniedException, _convert_error
from tdw_catalog.team import Team
import tdw_catalog.metadata.contract_owner as contract_owner
import tdw_catalog.metadata.currency as currency_field
import tdw_catalog.metadata.data_cost as data_cost
import tdw_catalog.metadata.date_field as date_field
import tdw_catalog.metadata.decimal as decimal
import tdw_catalog.metadata.license_expiry as license_expiry
import tdw_catalog.metadata.link as link
import tdw_catalog.metadata.list_field as list_field
import tdw_catalog.metadata.field as field
import tdw_catalog.metadata.number as number
import tdw_catalog.metadata.organization_member as organization_member
import tdw_catalog.metadata.organization_team as organization_team
import tdw_catalog.metadata.point_in_time as point_in_time
import tdw_catalog.metadata.point_of_contact as point_of_contact
import tdw_catalog.metadata.text as text
import tdw_catalog.metadata_template as metadata_template
import tdw_catalog.metadata.alias as alias
from tdw_catalog.organization_member import OrganizationMember
from tdw_catalog.team import Team
[docs]class TemplatedMetadataEditor():
"""
A :class:`.TemplatedMetadataEditor` assists with the alteration and updating of :class:`.MetadataField` values in a :class:`.Dataset`
"""
fields: List['field.MetadataField']
_dataset: 'dataset.Dataset'
template: 'metadata_template.MetadataTemplate'
def __init__(
self,
fields: List['field.MetadataField'],
dataset: 'dataset.Dataset',
template: Optional['metadata_template.MetadataTemplate'] = None
) -> None:
"""
Initializes a :class:`.TemplatedMetadataEditor` for the alteration and updating of metadata fields in a :class:`.Dataset`
Parameters
----------
fields : List[MetadataField] | List[MetadataTemplateField]
The list of metadata fields to be added or updated
"""
self.fields = fields if fields is not None else []
self._dataset = dataset
self.template = template
[docs] def get_field(
self,
key: str,
cls: 'field.MetadataField[field.T]' = None
) -> 'Union[field.MetadataField[field.T],field.TemplatedMetadataField[field.T]]':
"""
Get a specific metadata field belonging to this :class:`.Dataset`. The field's properties can then be changed directly
Parameters
----------
key : str
The key of the desired field
Returns
---------
MetadataField[T] | TemplatedMetadataField[T]
The specificed field whose key matches the provided key
Raises
------
CatalogNotFoundException:
If no :class:`.MetadataField` with a matching key can be found
"""
f = next((field for field in self.fields if field.key == key), None)
if f is None:
raise CatalogNotFoundException(
message="No metadata field with that key was found")
return field.TemplatedMetadataField(
f) if self.template is not None else f
[docs] def list_fields(self) -> 'List[field.MetadataField]':
"""
Return the list of fields currently being edited by this :class:`.MetadataEditor`
Parameters
----------
None
Returns
-------
List[MetadataField]
The list of :class:`.MetadataField`\\ s currently being edited by this :class:`.MetadataEditor`
"""
return self.fields
[docs] def keys(self) -> List[str]:
"""
Return a list of keys for each field in this :class:`.Dataset`'s custom or templated fields list
Parameters
----------
None
Returns
--------
List[str]
A list of strings representing the keys for each field in this list
"""
return list(map(lambda field: field["key"], self.fields))
[docs] def save(self) -> None:
"""
Save and persist all changes made to the metadata of the :class:`.Dataset`
Parameters
----------
None
Returns
-------
None
Raises
------
CatalogInternalException
If the call to the :class:`.Catalog` server fails
CatalogPermissionDeniedException
If the user does not have permission to make changes to this :class:`.Dataset`
"""
if self.template is not None:
untouched_fields = self._dataset.custom_metadata if self._dataset.custom_metadata else []
res = self.fields + untouched_fields
self._dataset._custom_fields = res
else:
untouched_fields = self._dataset.templated_metadata if self._dataset.templated_metadata else []
res = untouched_fields + self.fields
self._dataset._custom_fields = res
self._dataset.save()
[docs]class MetadataEditor(TemplatedMetadataEditor):
"""
A :class:`.MetadataEditor` assists with the alteration and updating of :class:`.MetadataField`\\ s in a :class:`.Dataset`
"""
[docs] def add_text_field(
self,
key: str,
value: Optional[str],
) -> 'MetadataEditor':
"""
Any additional text-based information you want to add to the :class:`.Dataset`
Parameters
----------
key : str
An identifying key for the this field
value : Optional[str]
An optional `str` value for the text this field represents
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
"""
f = text.Field(key, value=value)
self.fields.append(f)
return self
[docs] def add_attachment_field(
self,
key: str,
value: str,
) -> 'MetadataEditor':
"""
Attach any metadata files directly to the :class:`.Dataset`
Parameters
----------
key : str
A path for the file to be attached with this field
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
Raises
------
CatalogInternalException
If the call to the :class:`.Catalog` server fails when uploading the file to be attached
"""
try:
file_size = os.path.getsize(value)
file_name = os.path.basename(value)
(mime_type, encoding) = mimetypes.guess_type(value)
if mime_type is None:
mime_type = 'text/csv'
res = self._dataset._client._create_upload(
resource_type="dataset",
resource_id=self._dataset.id,
content_length=file_size,
content_type="csv",
filename=file_name)
import tdw_catalog.metadata.attachment as attachment
f = attachment.Field(
key=key,
value=res["upload"]["id"],
)
self.fields.append(f)
return self
except Exception as e:
raise _convert_error(e)
[docs] def add_linked_dataset_field(
self,
key: str,
value: Optional['dataset.Dataset'],
) -> 'MetadataEditor':
"""
Connect :class:`.Dataset`\\ s together to maintain lineage or increase discoverability
Parameters
----------
key : str
An identifying key for this field
value : Optional[Dataset]
An optional object representing a :class:`.Dataset` that has been semantically linked to this field
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
"""
import tdw_catalog.metadata.linked_dataset as linked_dataset
f = linked_dataset.Field(key, value=value)
self.fields.append(f)
return self
[docs] def add_link_field(
self,
key: str,
value: Optional[str],
) -> 'MetadataEditor':
"""
Any links, urls, or websites associated with the :class:`.Dataset`
Parameters
----------
key : str
An identifying key for this field
value : Optional[str]
An optional `str` value for the url this field represents
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
"""
f = link.Field(key, value=value)
self.fields.append(f)
return self
[docs] def add_organization_member_field(
self,
key: str,
value: Optional[OrganizationMember],
) -> 'MetadataEditor':
"""
Someone who is associated with the data
Parameters
----------
key : str
An identifying key for this field
value : Optional[OrganizationMember]
An optional :class:`.OrganizationMember` value representing a member of this :class:`.Dataset`'s :class:`.Organization`
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
"""
f = organization_member.Field(
key,
value=value,
)
self.fields.append(f)
return self
[docs] def add_organization_team_field(
self,
key: str,
value: Optional[Team],
) -> 'MetadataEditor':
"""
A :class:`.Team` who is associated with the data
Parameters
----------
key : str
An identifying key for this field
value : Optional[Team]
An optional :class:`.Team` value representing a team in this :class:`.Dataset`'s :class:`.Organization`
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
"""
f = organization_team.Field(key, value=value)
self.fields.append(f)
return self
[docs] def add_date_field(
self,
key: str,
value: Optional[date],
) -> 'MetadataEditor':
"""
Add dates to the :class:`.Dataset` to keep track of timelines or events
Parameters
----------
key : str
An identifying key for this field
value : Optional[date]
An optional `date` object for the date this field represents
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
"""
f = date_field.Field(key, value=value)
self.fields.append(f)
return self
[docs] def add_point_in_time_field(
self,
key: str,
value: Optional[datetime],
) -> 'MetadataEditor':
"""
A specific time and date associated with the :class:`.Dataset`
Parameters
----------
key : str
An identifying key for this field
value : Optional[datetime]
An optional `datetime` value for the datetime this field represents
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
"""
f = point_in_time.Field(key, value=value)
self.fields.append(f)
return self
[docs] def add_number_field(
self,
key: str,
value: Optional[int],
) -> 'MetadataEditor':
"""
Track numbers associated with the :class:`.Dataset`, like total number of allowed users
Parameters
----------
key : str
An identifying key for this field
value : Optional[int]
An optional `int` value for the number this field represents
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
"""
f = number.Field(key, value=value)
self.fields.append(f)
return self
[docs] def add_currency_field(
self,
key: str,
currency: Optional[str],
amount: Optional[float],
) -> 'MetadataEditor':
"""
A field for currency values
Parameters
----------
key : str
An identifying key for this field
currency : Optional[str]
An optional three character string representation for this currency type (e.g. USD)
amount: Optional[float]
An optional fractional amount of currency for this currency field
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
"""
f = currency_field.Field(key, currency, amount)
self.fields.append(f)
return self
[docs] def add_decimal_field(
self,
key: str,
value: Optional[float],
) -> 'MetadataEditor':
"""
A field for confidence values or other fractional information
Parameters
----------
key : str
An identifying key for this field
value : Optional[float]
An optional `float` value for the decimal this field represents
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
"""
f = decimal.Field(key, value=value)
self.fields.append(f)
return self
[docs] def add_alias_field(
self,
key: str,
value: Optional[str],
) -> 'MetadataEditor':
"""
An alternative unique identifier for this :class:`.Dataset`\\ , for integrating with external systems
Parameters
----------
key : str
An identifying key for the this field
value : Optional[str]
An optional `str` value for the alias this field represents
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
"""
f = alias.Field(key, value=value)
self.fields.append(f)
return self
[docs] def add_data_cost_recommended_field(
self,
currency: Optional[str],
amount: Optional[float],
) -> 'MetadataEditor':
"""
Track and report on the data license cost. The key of this field will automatically be set
Parameters
----------
currency : Optional[str]
An optional three character string representation for this currency type (e.g. USD)
amount: Optional[float]
An optional fractional amount of currency for this currency field
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
Raises
------
CatalogFailedPreconditionException
If the :class:`.Dataset` already has an attached data cost field, templated or not
"""
if self._dataset._custom_fields is not None:
for field in self._dataset._custom_fields:
if isinstance(field, data_cost.Field):
raise CatalogFailedPreconditionException(
message=
"Only one data cost recommended field may be applied per template"
)
f = data_cost.Field(currency=currency, amount=amount, key="cost__0")
self.fields.append(f)
return self
[docs] def add_point_of_contact_recommended_field(
self,
value: Optional[OrganizationMember],
) -> 'MetadataEditor':
"""
Primary person who manages the data license. The key of this field will automatically be set
Parameters
----------
value: Optional[OrganizationMember]
An optional :class:`.OrganizationMember` object representing a member of this :class:`.Organization`
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
Raises
------
CatalogFailedPreconditionException
If the :class:`.Dataset` already has an attached point of contact field, templated or not
"""
if self._dataset._custom_fields is not None:
for field in self._dataset._custom_fields:
if isinstance(field, point_of_contact.Field):
raise CatalogFailedPreconditionException(
message=
"Only one point of contact recommended field may be applied per template"
)
f = point_of_contact.Field(
key="poc__0",
value=value,
)
self.fields.append(f)
return self
[docs] def add_license_expiry_recommended_field(
self,
value: Optional[date],
) -> 'MetadataEditor':
"""
Monitor data license term limits. The key of this field will automatically be set
Parameters
----------
value : Optional[date]
An optional `date` value for the date that this field represents
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
Raises
------
CatalogFailedPreconditionException
If the :class:`.Dataset` already has an attached license expiry field, templated or not
"""
if self._dataset._custom_fields is not None:
for field in self._dataset._custom_fields:
if isinstance(field, license_expiry.Field):
raise CatalogFailedPreconditionException(
message=
"Only one license expiry recommended field may be applied per template"
)
f = license_expiry.Field(
key="expiryDate__0",
value=value,
)
self.fields.append(f)
return self
[docs] def add_contract_owner_recommended_field(
self,
value: Optional[Team],
) -> 'MetadataEditor':
"""
Team or department that owns the data license. The key of this field will automatically be set
Parameters
----------
value : Optional[Team]
An optional :class:`.Team` object representing a team in this :class:`.Organization`
Returns
-------
MetadataEditor
Returns this :class:`.MetadataEditor` for further metadata editing
Raises
------
CatalogFailedPreconditionException
If the :class:`.Dataset` already has an attached contract owner field, templated or not
"""
if self._dataset._custom_fields is not None:
for field in self._dataset._custom_fields:
if isinstance(field, contract_owner.Field):
raise CatalogFailedPreconditionException(
message=
"Only one contract owner recommended field may be applied per template"
)
f = contract_owner.Field(
key="contractOwner__0",
value=value,
)
self.fields.append(f)
return self
[docs] def remove_field(self, key: str) -> 'MetadataEditor':
"""
Remove a metadata field from the :class:`.Dataset`
Parameters
----------
key : str
The key of the field to be removed
Returns
-------
MetadataEditor
This :class:`.MetadataEditor` with the given field removed from its fields list
"""
self.fields = list(
filter(lambda field: field["key"] != key, self.fields))
return self
[docs] def reorder(
self,
new_field_order: 'List[field.MetadataField]') -> 'MetadataEditor':
"""
Reorder the fields list of this :class:`.MetadataEditor`
Parameters
----------
new_field_order : List[MetadataField]
The newly ordered list of :class:`.MetadataTemplateField`\\ s to replace the old list on this :class:`.MetadataTemplate`. Must contain the same fields as the original list, with the only allowed change being the list's order
Returns
-------
MetadataEditor
This :class:`.MetadataEditor` with a reordered fields list
Raises
------
CatalogInvalidArgumentException
If the user supplied fields list contains a field with a key that does not exist on this :class:`.Dataset`, the number of fields in the list provided does not match the number of fields on the :class:`.Dataset`, or if a provided field matches the key of a field in the original list but is of a different type
"""
if len(self.fields) != len(new_field_order):
raise CatalogInvalidArgumentException(
message=
"The number of fields provided must equal the number of custom fields on the Dataset"
)
if set(new_field_order).issubset(self.fields) == False:
raise CatalogInvalidArgumentException(
message=
"One or more of the fields provided does not match the original custom fields list of this Dataset"
)
keys = self.keys()
for f in new_field_order:
if f.key not in keys:
raise CatalogInvalidArgumentException(
message=
f"No custom field with key {f.key} exists on this Dataset")
self.fields = new_field_order
return self