Skip to content

Commit

Permalink
Merge branch 'main' into wip/xmlupload-make-serialise-rdf-values-into…
Browse files Browse the repository at this point in the history
…-classes
  • Loading branch information
Nora-Olivia-Ammann authored Nov 14, 2024
2 parents 5c554f6 + 0279615 commit d091bb3
Show file tree
Hide file tree
Showing 2 changed files with 152 additions and 72 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -2,20 +2,22 @@

from abc import ABC
from abc import abstractmethod
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any


@dataclass(frozen=True)
class SerialiseProperty:
property_name: str
values: list[SerialiseValue]
values: Sequence[SerialiseValue]

def serialise(self) -> dict[str, Any]:
"""Serialise the property and all its values."""
return {self.property_name: [value.serialise() for value in self.values]}


@dataclass(frozen=True)
class SerialiseValue(ABC):
"""A value to be serialised."""

Expand All @@ -34,14 +36,9 @@ def _get_optionals(self) -> dict[str, str]:
return optionals


@dataclass(frozen=True)
class SerialiseColor(SerialiseValue):
"""A ColorValue to be serialised."""

value: str
permissions: str | None
comment: str | None

def serialise(self) -> dict[str, Any]:
serialised = {
"@type": "knora-api:ColorValue",
Expand All @@ -51,14 +48,90 @@ def serialise(self) -> dict[str, Any]:
return serialised


@dataclass(frozen=True)
class SerialiseDecimal(SerialiseValue):
"""A DecimalValue to be serialised."""

def serialise(self) -> dict[str, Any]:
serialised = {
"@type": "knora-api:DecimalValue",
"knora-api:decimalValueAsDecimal": {
"@type": "xsd:decimal",
"@value": self.value,
},
}
serialised.update(self._get_optionals())
return serialised


class SerialiseGeometry(SerialiseValue):
"""A GeomValue to be serialised."""

def serialise(self) -> dict[str, Any]:
serialised = {
"@type": "knora-api:GeomValue",
"knora-api:geometryValueAsGeometry": self.value,
}
serialised.update(self._get_optionals())
return serialised


class SerialiseGeoname(SerialiseValue):
"""A GeonameValue to be serialised."""

def serialise(self) -> dict[str, Any]:
serialised = {
"@type": "knora-api:GeonameValue",
"knora-api:geonameValueAsGeonameCode": self.value,
}
serialised.update(self._get_optionals())
return serialised


class SerialiseSimpletext(SerialiseValue):
"""A Simpletext to be serialised."""

def serialise(self) -> dict[str, Any]:
serialised = {
"@type": "knora-api:TextValue",
"knora-api:valueAsString": self.value,
}
serialised.update(self._get_optionals())
return serialised


class SerialiseRichtext(SerialiseValue):
"""A Richtext to be serialised."""

def serialise(self) -> dict[str, Any]:
serialised = {
"@type": "knora-api:TextValue",
"knora-api:textValueAsXml": self.value,
"knora-api:textValueHasMapping": {
"@id": "http://rdfh.ch/standoff/mappings/StandardMapping",
},
}
serialised.update(self._get_optionals())
return serialised


class SerialiseTime(SerialiseValue):
"""A TimeValue to be serialised."""

def serialise(self) -> dict[str, Any]:
serialised = {
"@type": "knora-api:TimeValue",
"knora-api:timeValueAsTimeStamp": {
"@type": "xsd:dateTimeStamp",
"@value": self.value,
},
}
serialised.update(self._get_optionals())
return serialised


class SerialiseURI(SerialiseValue):
"""A UriValue to be serialised."""

value: str
permissions: str | None
comment: str | None

def serialise(self) -> dict[str, Any]:
serialised = {
"@type": "knora-api:UriValue",
Expand Down
129 changes: 68 additions & 61 deletions src/dsp_tools/commands/xmlupload/resource_create_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,13 @@
from dsp_tools.commands.xmlupload.models.serialise.serialise_rdf_value import BooleanValueRDF
from dsp_tools.commands.xmlupload.models.serialise.serialise_rdf_value import IntValueRDF
from dsp_tools.commands.xmlupload.models.serialise.serialise_value import SerialiseColor
from dsp_tools.commands.xmlupload.models.serialise.serialise_value import SerialiseDecimal
from dsp_tools.commands.xmlupload.models.serialise.serialise_value import SerialiseGeometry
from dsp_tools.commands.xmlupload.models.serialise.serialise_value import SerialiseGeoname
from dsp_tools.commands.xmlupload.models.serialise.serialise_value import SerialiseProperty
from dsp_tools.commands.xmlupload.models.serialise.serialise_value import SerialiseRichtext
from dsp_tools.commands.xmlupload.models.serialise.serialise_value import SerialiseSimpletext
from dsp_tools.commands.xmlupload.models.serialise.serialise_value import SerialiseTime
from dsp_tools.commands.xmlupload.models.serialise.serialise_value import SerialiseURI
from dsp_tools.commands.xmlupload.models.serialise.serialise_value import SerialiseValue
from dsp_tools.models.exceptions import BaseError
Expand Down Expand Up @@ -138,20 +144,39 @@ def make_values(p: XMLProperty) -> list[dict[str, Any]]:
last_prop_name = None

str_value_to_serialiser_mapper = {
"uri": SerialiseURI,
"color": SerialiseColor,
"geoname": SerialiseGeoname,
"time": SerialiseTime,
"uri": SerialiseURI,
}

for prop in resource.properties:
match prop.valtype:
# serialised as dict
case "uri" | "color" as val_type:
case "uri" | "color" | "geoname" | "time" as val_type:
transformed_prop = _transform_into_serialise_prop(
prop=prop,
permissions_lookup=self.permissions_lookup,
seraliser=str_value_to_serialiser_mapper[val_type],
)
properties_serialised.update(transformed_prop.serialise())
case "decimal":
transformed_prop = _transform_into_decimal_prop(
prop=prop, permissions_lookup=self.permissions_lookup
)
properties_serialised.update(transformed_prop.serialise())
case "geometry":
transformed_prop = _transform_into_geometry_prop(
prop=prop, permissions_lookup=self.permissions_lookup
)
properties_serialised.update(transformed_prop.serialise())
case "text":
transformed_prop = _transform_text_prop(
prop=prop,
permissions_lookup=self.permissions_lookup,
iri_resolver=self.iri_resolver,
)
properties_serialised.update(transformed_prop.serialise())
# serialised with rdflib
case "integer":
int_prop_name = self._get_absolute_prop_iri(prop.name, namespaces)
Expand Down Expand Up @@ -180,26 +205,16 @@ def _get_absolute_prop_iri(self, prefixed_prop: str, namespaces: dict[str, Names
raise InputError(f"Could not find namespace for prefix: {prefix}")
return namespace[prop]

def _make_value(self, value: XMLValue, value_type: str) -> dict[str, Any]: # noqa: PLR0912 (too many branches)
def _make_value(self, value: XMLValue, value_type: str) -> dict[str, Any]:
match value_type:
case "date":
res = _make_date_value(value)
case "decimal":
res = _make_decimal_value(value)
case "geometry":
res = _make_geometry_value(value)
case "geoname":
res = _make_geoname_value(value)
case "interval":
res = _make_interval_value(value)
case "resptr":
res = _make_link_value(value, self.iri_resolver)
case "list":
res = _make_list_value(value, self.listnode_lookup)
case "text":
res = _make_text_value(value, self.iri_resolver)
case "time":
res = _make_time_value(value)
case _:
raise UserError(f"Unknown value type: {value_type}")
if value.comment:
Expand Down Expand Up @@ -301,32 +316,29 @@ def _make_date_value(value: XMLValue) -> dict[str, Any]:
return res


def _make_decimal_value(value: XMLValue) -> dict[str, Any]:
def _transform_into_decimal_prop(prop: XMLProperty, permissions_lookup: dict[str, Permissions]) -> SerialiseProperty:
vals = [_transform_into_decimal_value(v, permissions_lookup) for v in prop.values]
return SerialiseProperty(property_name=prop.name, values=vals)


def _transform_into_decimal_value(value: XMLValue, permissions_lookup: dict[str, Permissions]) -> SerialiseDecimal:
s = _assert_is_string(value.value)
return {
"@type": "knora-api:DecimalValue",
"knora-api:decimalValueAsDecimal": {
"@type": "xsd:decimal",
"@value": str(float(s)),
},
}
val = str(float(s))
permission_str = _get_permission_str(value.permissions, permissions_lookup)
return SerialiseDecimal(value=val, permissions=permission_str, comment=value.comment)


def _transform_into_geometry_prop(prop: XMLProperty, permissions_lookup: dict[str, Permissions]) -> SerialiseProperty:
vals = [_transform_into_geometry_value(v, permissions_lookup) for v in prop.values]
return SerialiseProperty(property_name=prop.name, values=vals)


def _make_geometry_value(value: XMLValue) -> dict[str, Any]:
def _transform_into_geometry_value(value: XMLValue, permissions_lookup: dict[str, Permissions]) -> SerialiseGeometry:
s = _assert_is_string(value.value)
# this removes all whitespaces from the embedded json string
encoded_value = json.dumps(json.loads(s))
return {
"@type": "knora-api:GeomValue",
"knora-api:geometryValueAsGeometry": encoded_value,
}


def _make_geoname_value(value: XMLValue) -> dict[str, Any]:
return {
"@type": "knora-api:GeonameValue",
"knora-api:geonameValueAsGeonameCode": value.value,
}
permission_str = _get_permission_str(value.permissions, permissions_lookup)
return SerialiseGeometry(value=encoded_value, permissions=permission_str, comment=value.comment)


def _make_boolean_prop(
Expand Down Expand Up @@ -440,34 +452,29 @@ def _make_list_value(value: XMLValue, iri_lookup: dict[str, str]) -> dict[str, A
raise BaseError(msg)


def _make_text_value(value: XMLValue, iri_resolver: IriResolver) -> dict[str, Any]:
match value.value:
case str() as s:
return {
"@type": "knora-api:TextValue",
"knora-api:valueAsString": s,
}
case FormattedTextValue() as xml:
xml_with_iris = xml.with_iris(iri_resolver)
return {
"@type": "knora-api:TextValue",
"knora-api:textValueAsXml": xml_with_iris.as_xml(),
"knora-api:textValueHasMapping": {
"@id": "http://rdfh.ch/standoff/mappings/StandardMapping",
},
}
case _:
assert_never(value.value)


def _make_time_value(value: XMLValue) -> dict[str, Any]:
return {
"@type": "knora-api:TimeValue",
"knora-api:timeValueAsTimeStamp": {
"@type": "xsd:dateTimeStamp",
"@value": value.value,
},
}
def _transform_text_prop(
prop: XMLProperty, permissions_lookup: dict[str, Permissions], iri_resolver: IriResolver
) -> SerialiseProperty:
values = []
for val in prop.values:
match val.value:
case str():
values.append(_transform_into_serialise_value(val, permissions_lookup, SerialiseSimpletext))
case FormattedTextValue():
values.append(_transform_into_richtext_value(val, permissions_lookup, iri_resolver))
case _:
assert_never(val.value)
return SerialiseProperty(property_name=prop.name, values=values)


def _transform_into_richtext_value(
val: XMLValue, permissions_lookup: dict[str, Permissions], iri_resolver: IriResolver
) -> SerialiseRichtext:
xml_val = cast(FormattedTextValue, val.value)
xml_with_iris = xml_val.with_iris(iri_resolver)
val_str = xml_with_iris.as_xml()
permission_str = _get_permission_str(val.permissions, permissions_lookup)
return SerialiseRichtext(value=val_str, permissions=permission_str, comment=val.comment)


def _transform_into_serialise_prop(
Expand Down

0 comments on commit d091bb3

Please sign in to comment.