Skip to content
Snippets Groups Projects
Commit d8ebb2ee authored by Frank Lange's avatar Frank Lange
Browse files

generalize the way 1:N metadata of learning resources is retrieved

apply this pattern to: keywords, languages, learning resource types and media types
parent 385a1f8d
Branches
No related tags found
No related merge requests found
Pipeline #1545898 passed
Showing
with 132 additions and 250 deletions
from collections import defaultdict
from typing import Dict, List
from rdflib import RDF, URIRef, Variable
from rdflib import URIRef
from rdflib.term import Node
from project.dalia.query.utils import query_dalia_dataset
from project.dalia.query_builder.query_builder import QueryBuilder, VALUES
from project.dalia.rdf.namespace import SCHEMA, educor
from project.dalia.query.items.metadata.one_to_many_metadata import get_one_to_many_metadata_for_resources
from project.dalia.rdf.namespace import SCHEMA
_VARIABLES = {
"lr": Variable("lr"),
"keyword": Variable("keyword")
}
def prepare_query_for_keywords_metadata_for_resources(resource_uri_refs: List[URIRef]) -> str:
var_lr = _VARIABLES["lr"]
resource_uri_ref_blocks = [[uri_ref] for uri_ref in resource_uri_refs]
return QueryBuilder().SELECT(
*_VARIABLES.values()
).WHERE(
VALUES(
[var_lr],
resource_uri_ref_blocks
),
(var_lr, RDF.type, educor.EducationalResource),
(var_lr, SCHEMA.keywords, _VARIABLES["keyword"])
).build()
def keywords_from_results(results) -> Dict[URIRef, List[str]]:
lr_keywords = defaultdict(list)
for result in results:
lr_keywords[result.lr].append(str(result.keyword))
return lr_keywords
def _keyword_from_result_item(item: Node) -> str:
return str(item)
def get_keywords_metadata_for_resources(resource_uri_refs: List[URIRef]) -> Dict[URIRef, List[str]]:
......@@ -46,7 +18,8 @@ def get_keywords_metadata_for_resources(resource_uri_refs: List[URIRef]) -> Dict
:param resource_uri_refs: List of learning resource URIRefs
:return: Associations between the learning resource URIRefs and their respective list of keywords.
"""
query = prepare_query_for_keywords_metadata_for_resources(resource_uri_refs)
results = query_dalia_dataset(query)
return keywords_from_results(results)
return get_one_to_many_metadata_for_resources(
resource_uri_refs=resource_uri_refs,
relation=SCHEMA.keywords,
item_crosswalk_fn=_keyword_from_result_item
)
from collections import defaultdict
from typing import Dict, List
from rdflib import DCTERMS, Literal, RDF, URIRef, Variable
from rdflib import DCTERMS, URIRef
from rdflib.term import Node
from project.dalia.query.items.facets.facet_objects import LANGUAGE_FACET
from project.dalia.query.utils import query_dalia_dataset
from project.dalia.query_builder.query_builder import QueryBuilder, VALUES
from project.dalia.rdf.namespace import educor
_VARIABLES = {
"lr": Variable("lr"),
"language": Variable("language")
}
def prepare_query_for_languages_for_resources(resource_uri_refs: List[URIRef]) -> str:
var_lr = _VARIABLES["lr"]
resource_uri_ref_blocks = [[uri_ref] for uri_ref in resource_uri_refs]
return QueryBuilder().SELECT(
*_VARIABLES.values()
).WHERE(
VALUES(
[var_lr],
resource_uri_ref_blocks
),
(var_lr, RDF.type, educor.EducationalResource),
(var_lr, DCTERMS.language, _VARIABLES["language"])
).build()
from project.dalia.query.items.metadata.one_to_many_metadata import get_one_to_many_metadata_for_resources
LANGUAGE_LABEL_MAPPING = LANGUAGE_FACET.items
def languages_from_results(results) -> Dict[URIRef, List[str]]:
lr_languages = defaultdict(list)
for result in results:
lr_languages[result.lr].append(LANGUAGE_LABEL_MAPPING[result.language])
return lr_languages
def _language_from_result_item(item: Node) -> str:
return LANGUAGE_LABEL_MAPPING[item]
def get_languages_for_resources(resource_uri_refs: List[URIRef]) -> Dict[URIRef, List[str]]:
......@@ -50,7 +20,8 @@ def get_languages_for_resources(resource_uri_refs: List[URIRef]) -> Dict[URIRef,
:param resource_uri_refs: List of learning resource URIRefs
:return: Associations between the learning resource URIRefs and their respective list of languages.
"""
query = prepare_query_for_languages_for_resources(resource_uri_refs)
results = query_dalia_dataset(query)
return languages_from_results(results)
return get_one_to_many_metadata_for_resources(
resource_uri_refs=resource_uri_refs,
relation=DCTERMS.language,
item_crosswalk_fn=_language_from_result_item
)
from collections import defaultdict
from typing import Dict, List
from rdflib import RDF, URIRef, Variable
from rdflib import URIRef
from rdflib.term import Node
from project.dalia.api_models.api_models import LabelValueItem
from project.dalia.query.items.facets.facet_objects import LEARNING_RESOURCE_TYPE_FACET
from project.dalia.query.utils import query_dalia_dataset
from project.dalia.query_builder.query_builder import QueryBuilder, VALUES
from project.dalia.rdf.namespace import MoDalia, educor
_VARIABLES = {
"lr": Variable("lr"),
"learning_resource_type": Variable("learning_resource_type")
}
def prepare_query_for_learning_resource_types_for_resources(resource_uri_refs: List[URIRef]) -> str:
var_lr = _VARIABLES["lr"]
resource_uri_ref_blocks = [[uri_ref] for uri_ref in resource_uri_refs]
return QueryBuilder().SELECT(
*_VARIABLES.values()
).WHERE(
VALUES(
[var_lr],
resource_uri_ref_blocks
),
(var_lr, RDF.type, educor.EducationalResource),
(var_lr, MoDalia.hasLearningType, _VARIABLES["learning_resource_type"])
).build()
from project.dalia.query.items.metadata.one_to_many_metadata import get_one_to_many_metadata_for_resources
from project.dalia.rdf.namespace import MoDalia
LEARNING_RESOURCE_TYPES_LABEL_MAPPING = LEARNING_RESOURCE_TYPE_FACET.items
def learning_resource_types_from_results(results) -> Dict[URIRef, List[LabelValueItem]]:
lr_learning_resource_types = defaultdict(list)
for result in results:
lr_learning_resource_types[result.lr].append(
LabelValueItem(
label=LEARNING_RESOURCE_TYPES_LABEL_MAPPING[result.learning_resource_type],
value=str(result.learning_resource_type)
)
)
return lr_learning_resource_types
def _learning_resource_type_from_result_item(item: Node) -> LabelValueItem:
return LabelValueItem(
label=LEARNING_RESOURCE_TYPES_LABEL_MAPPING[item],
value=str(item)
)
def get_learning_resource_types_for_resources(resource_uri_refs: List[URIRef]) -> Dict[URIRef, List[LabelValueItem]]:
......@@ -56,7 +25,8 @@ def get_learning_resource_types_for_resources(resource_uri_refs: List[URIRef]) -
:param resource_uri_refs: List of learning resource URIRefs
:return: Associations between the learning resource URIRefs and their respective list of learning resource types.
"""
query = prepare_query_for_learning_resource_types_for_resources(resource_uri_refs)
results = query_dalia_dataset(query)
return learning_resource_types_from_results(results)
return get_one_to_many_metadata_for_resources(
resource_uri_refs=resource_uri_refs,
relation=MoDalia.hasLearningType,
item_crosswalk_fn=_learning_resource_type_from_result_item
)
from collections import defaultdict
from typing import Dict, List
from rdflib import RDF, URIRef, Variable
from rdflib import URIRef
from rdflib.term import Node
from project.dalia.api_models.api_models import LabelValueItem
from project.dalia.query.items.facets.facet_objects import MEDIA_TYPE_FACET
from project.dalia.query.utils import query_dalia_dataset
from project.dalia.query_builder.query_builder import QueryBuilder, VALUES
from project.dalia.rdf.namespace import MoDalia, educor
_VARIABLES = {
"lr": Variable("lr"),
"media_type": Variable("media_type")
}
def prepare_query_for_media_types_for_resources(resource_uri_refs: List[URIRef]) -> str:
var_lr = _VARIABLES["lr"]
resource_uri_ref_blocks = [[uri_ref] for uri_ref in resource_uri_refs]
return QueryBuilder().SELECT(
*_VARIABLES.values()
).WHERE(
VALUES(
[var_lr],
resource_uri_ref_blocks
),
(var_lr, RDF.type, educor.EducationalResource),
(var_lr, MoDalia.hasMediaType, _VARIABLES["media_type"])
).build()
from project.dalia.query.items.metadata.one_to_many_metadata import get_one_to_many_metadata_for_resources
from project.dalia.rdf.namespace import MoDalia
MEDIA_TYPES_LABEL_MAPPING = MEDIA_TYPE_FACET.items
def media_types_from_results(results) -> Dict[URIRef, List[LabelValueItem]]:
lr_media_types = defaultdict(list)
for result in results:
lr_media_types[result.lr].append(
LabelValueItem(
label=MEDIA_TYPES_LABEL_MAPPING[result.media_type],
value=str(result.media_type)
)
)
return lr_media_types
def _media_type_from_result_item(item: Node) -> LabelValueItem:
return LabelValueItem(
label=MEDIA_TYPES_LABEL_MAPPING[item],
value=str(item)
)
def get_media_types_for_resources(resource_uri_refs: List[URIRef]) -> Dict[URIRef, List[LabelValueItem]]:
......@@ -56,7 +25,8 @@ def get_media_types_for_resources(resource_uri_refs: List[URIRef]) -> Dict[URIRe
:param resource_uri_refs: List of learning resource URIRefs
:return: Associations between the learning resource URIRefs and their respective list of media types.
"""
query = prepare_query_for_media_types_for_resources(resource_uri_refs)
results = query_dalia_dataset(query)
return media_types_from_results(results)
return get_one_to_many_metadata_for_resources(
resource_uri_refs=resource_uri_refs,
relation=MoDalia.hasMediaType,
item_crosswalk_fn=_media_type_from_result_item
)
from collections import defaultdict
from typing import Any, Callable, Dict, List
from rdflib import RDF, URIRef, Variable
from rdflib.term import Node
from project.dalia.query.utils import query_dalia_dataset
from project.dalia.query_builder.query_builder import QueryBuilder, VALUES
from project.dalia.rdf.namespace import educor
_VARIABLES = {
"lr": Variable("lr"),
"item": Variable("item")
}
def prepare_query_for_one_to_many_metadata_for_resources(resource_uri_refs: List[URIRef], relation: URIRef) -> str:
var_lr = _VARIABLES["lr"]
resource_uri_ref_blocks = [[uri_ref] for uri_ref in resource_uri_refs]
return QueryBuilder().SELECT(
*_VARIABLES.values()
).WHERE(
VALUES(
[var_lr],
resource_uri_ref_blocks
),
(var_lr, RDF.type, educor.EducationalResource),
(var_lr, relation, _VARIABLES["item"])
).build()
def _metadata_from_results(results, item_crosswalk_fn: Callable[[Node], Any]) -> Dict[URIRef, List[str]]:
lr_metadata_items = defaultdict(list)
for result in results:
lr_metadata_items[result.lr].append(item_crosswalk_fn(result.item))
return lr_metadata_items
def get_one_to_many_metadata_for_resources(
resource_uri_refs: List[URIRef],
relation: URIRef,
item_crosswalk_fn: Callable[[Node], Any],
) -> Dict[URIRef, List[Any]]:
"""
Retrieve the one-to-many metadata for each of the given learning resource URIRefs.
:param resource_uri_refs: List of learning resource URIRefs
:param relation: relation between a learning resource and the metadata items (RDF property)
:param item_crosswalk_fn: function to generate a Python object from an item received by the SPARQL call
:return: Associations between the learning resource URIRefs and their respective list of metadata items.
"""
query = prepare_query_for_one_to_many_metadata_for_resources(resource_uri_refs, relation)
results = query_dalia_dataset(query)
return _metadata_from_results(results, item_crosswalk_fn)
from rdflib import URIRef
from project.dalia.query.items.metadata.keywords import (
get_keywords_metadata_for_resources,
prepare_query_for_keywords_metadata_for_resources,
)
from tests.project.dalia.utils import dedent_and_normalize, normalize
def test_prepare_query_for_keywords_metadata_for_resources():
query = prepare_query_for_keywords_metadata_for_resources([URIRef("abc"), URIRef("def")])
assert normalize(query) == dedent_and_normalize("""
SELECT ?lr ?keyword
WHERE {
VALUES ( ?lr ) {
( <abc> )
( <def> )
}
?lr <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <https://github.com/tibonto/educor#EducationalResource> .
?lr <https://schema.org/keywords> ?keyword .
}
""")
from project.dalia.query.items.metadata.keywords import get_keywords_metadata_for_resources
def test_get_keywords_metadata_for_resources(triplestore):
......@@ -33,10 +13,10 @@ def test_get_keywords_metadata_for_resources(triplestore):
assert lr_keywords == {
URIRef('https://id.dalia.education/learning-resource/20f0b586-8019-40d1-8f92-baa6ac168bca'): [
'Research Data Management',
'FAIR',
'I3D:bio',
'Microscopy',
'Research Data Management'
],
URIRef('https://id.dalia.education/learning-resource/1f794f99-c131-4fb6-b238-5750afa197c8'): [
'NFDI',
......
from rdflib import URIRef
from project.dalia.query.items.metadata.languages import (
get_languages_for_resources,
prepare_query_for_languages_for_resources,
)
from tests.project.dalia.utils import dedent_and_normalize, normalize
def test_prepare_query_for_languages_for_resources():
query = prepare_query_for_languages_for_resources([URIRef("abc"), URIRef("def")])
assert normalize(query) == dedent_and_normalize("""
SELECT ?lr ?language
WHERE {
VALUES ( ?lr ) {
( <abc> )
( <def> )
}
?lr <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <https://github.com/tibonto/educor#EducationalResource> .
?lr <http://purl.org/dc/terms/language> ?language .
}
""")
from project.dalia.query.items.metadata.languages import get_languages_for_resources
def test_get_languages_for_resources(triplestore):
......
from rdflib import URIRef
from project.dalia.api_models.api_models import LabelValueItem
from project.dalia.query.items.metadata.learning_resource_types import (
get_learning_resource_types_for_resources,
prepare_query_for_learning_resource_types_for_resources,
)
from project.dalia.query.items.metadata.learning_resource_types import get_learning_resource_types_for_resources
from project.dalia.rdf.namespace import MoDalia
from tests.project.dalia.utils import dedent_and_normalize, normalize
def test_prepare_query_for_learning_resource_types_for_resources():
query = prepare_query_for_learning_resource_types_for_resources([URIRef("abc"), URIRef("def")])
assert normalize(query) == dedent_and_normalize("""
SELECT ?lr ?learning_resource_type
WHERE {
VALUES ( ?lr ) {
( <abc> )
( <def> )
}
?lr <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <https://github.com/tibonto/educor#EducationalResource> .
?lr <https://purl.org/ontology/modalia#hasLearningType> ?learning_resource_type .
}
""")
def test_get_learning_resource_types_for_resources(triplestore):
......
from rdflib import URIRef
from project.dalia.api_models.api_models import LabelValueItem
from project.dalia.query.items.metadata.media_types import (
get_media_types_for_resources,
prepare_query_for_media_types_for_resources,
)
from project.dalia.query.items.metadata.media_types import get_media_types_for_resources
from project.dalia.rdf.namespace import SCHEMA
from tests.project.dalia.utils import dedent_and_normalize, normalize
def test_prepare_query_for_media_types_for_resources():
query = prepare_query_for_media_types_for_resources([URIRef("abc"), URIRef("def")])
assert normalize(query) == dedent_and_normalize("""
SELECT ?lr ?media_type
WHERE {
VALUES ( ?lr ) {
( <abc> )
( <def> )
}
?lr <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <https://github.com/tibonto/educor#EducationalResource> .
?lr <https://purl.org/ontology/modalia#hasMediaType> ?media_type .
}
""")
def test_get_media_types_for_resources(triplestore):
......
from rdflib import URIRef
from project.dalia.query.items.metadata.one_to_many_metadata import prepare_query_for_one_to_many_metadata_for_resources
from tests.project.dalia.utils import dedent_and_normalize, normalize
def test_prepare_query_for_one_to_many_metadata_for_resources():
query = prepare_query_for_one_to_many_metadata_for_resources([URIRef("abc"), URIRef("def")], URIRef("relation"))
assert normalize(query) == dedent_and_normalize("""
SELECT ?lr ?item
WHERE {
VALUES ( ?lr ) {
( <abc> )
( <def> )
}
?lr <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <https://github.com/tibonto/educor#EducationalResource> .
?lr <relation> ?item .
}
""")
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment