mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-17 17:21:02 +02:00
feat: preserve RDF language tags and datatypes through ingestion and query (#1047)
Language tags and XSD datatypes on RDF literals were being silently dropped at multiple points in the pipeline, making SPARQL FILTER(LANG()) queries and multilingual datasets non-functional. Ingestion fixes: - load_knowledge.py and load_turtle.py now extract .language and .datatype from rdflib Literal objects instead of discarding them - Triple dataclass gains optional o_datatype and o_language fields - _string_to_term() emits "dt" and "ln" in the wire format - AsyncBulkClient uses the same term serialization as BulkClient Query fix: - TriplesClient.query_gen() preserves Term objects directly instead of converting through Uri/Literal (which are str subclasses with no room for language metadata), enabling SPARQL LANG() filters to work
This commit is contained in:
parent
f9f5a2318f
commit
0c36dda443
6 changed files with 53 additions and 20 deletions
|
|
@ -4,6 +4,7 @@ import websockets
|
|||
from typing import Optional, AsyncIterator, Dict, Any, Iterator
|
||||
|
||||
from . types import Triple
|
||||
from . bulk_client import _string_to_term
|
||||
|
||||
|
||||
class AsyncBulkClient:
|
||||
|
|
@ -45,9 +46,13 @@ class AsyncBulkClient:
|
|||
async with websockets.connect(ws_url, ping_interval=20, ping_timeout=self.timeout) as websocket:
|
||||
async for triple in triples:
|
||||
message = {
|
||||
"s": triple.s,
|
||||
"p": triple.p,
|
||||
"o": triple.o
|
||||
"s": _string_to_term(triple.s),
|
||||
"p": _string_to_term(triple.p),
|
||||
"o": _string_to_term(
|
||||
triple.o,
|
||||
datatype=triple.o_datatype,
|
||||
language=triple.o_language,
|
||||
),
|
||||
}
|
||||
await websocket.send(json.dumps(message))
|
||||
|
||||
|
|
|
|||
|
|
@ -15,13 +15,19 @@ from . types import Triple
|
|||
from . exceptions import ProtocolException
|
||||
|
||||
|
||||
def _string_to_term(value: str) -> Dict[str, Any]:
|
||||
def _string_to_term(
|
||||
value: str, datatype: str = "", language: str = ""
|
||||
) -> Dict[str, Any]:
|
||||
"""Convert a string value to Term format for the gateway."""
|
||||
# Treat URIs as IRI type, otherwise as literal
|
||||
if value.startswith("http://") or value.startswith("https://") or "://" in value:
|
||||
return {"t": "i", "i": value}
|
||||
else:
|
||||
return {"t": "l", "v": value}
|
||||
result: Dict[str, Any] = {"t": "l", "v": value}
|
||||
if datatype:
|
||||
result["dt"] = datatype
|
||||
if language:
|
||||
result["ln"] = language
|
||||
return result
|
||||
|
||||
|
||||
class BulkClient:
|
||||
|
|
@ -141,7 +147,11 @@ class BulkClient:
|
|||
batch.append({
|
||||
"s": _string_to_term(triple.s),
|
||||
"p": _string_to_term(triple.p),
|
||||
"o": _string_to_term(triple.o)
|
||||
"o": _string_to_term(
|
||||
triple.o,
|
||||
datatype=triple.o_datatype,
|
||||
language=triple.o_language,
|
||||
),
|
||||
})
|
||||
if len(batch) >= batch_size:
|
||||
message = {
|
||||
|
|
|
|||
|
|
@ -19,10 +19,14 @@ class Triple:
|
|||
s: Subject (entity URI or value)
|
||||
p: Predicate (relationship URI)
|
||||
o: Object (entity URI, literal value, or typed value)
|
||||
o_datatype: XSD datatype URI for literal objects (e.g. "xsd:string")
|
||||
o_language: Language tag for literal objects (e.g. "en", "lt")
|
||||
"""
|
||||
s : str
|
||||
p : str
|
||||
o : str
|
||||
o_datatype : str = ""
|
||||
o_language : str = ""
|
||||
|
||||
@dataclasses.dataclass
|
||||
class ConfigKey:
|
||||
|
|
|
|||
|
|
@ -58,7 +58,7 @@ class TriplesClient(RequestResponse):
|
|||
raise RuntimeError(resp.error.message)
|
||||
|
||||
batch = [
|
||||
Triple(to_value(v.s), to_value(v.p), to_value(v.o))
|
||||
Triple(v.s, v.p, v.o)
|
||||
for v in resp.triples
|
||||
]
|
||||
await queue.put(batch)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue