在rdflib中创建和使用自定义SPARQL函数指南

from rdflib import Graph, URIRef, Literal
from rdflib.plugins.sparql.operators import custom_function

# 定义一个URI来标识你的自定义函数
MY_CUSTOM_FUNCTION_URI = URIRef("http://example.org/myCustomFunction")

@custom_function(MY_CUSTOM_FUNCTION_URI)
def myCustomFunction(*args):
    # 函数逻辑
    return Literal("some_result")

from rdflib import Graph, URIRef, Literal
from rdflib.plugins.sparql.operators import custom_function

g = Graph()

@custom_function(URIRef("http://example.org/myCustomFunction"))
def myCustomFunction(args): # 定义时期望一个参数 'args'
    return Literal("test")

query = """
SELECT ?result WHERE {
    BIND(<http://example.org/myCustomFunction>() AS ?result) # 调用时未提供任何参数
}
"""

print("--- 错误示例结果 ---")
for row in g.query(query):
    print(f"Result: {row.result}") # 将不会打印任何内容

from rdflib import Graph, URIRef, Literal
from rdflib.plugins.sparql.operators import custom_function

# 初始化一个图（即使为空，也可以用于测试自定义函数）
g = Graph()

# 定义自定义函数的URI
MY_ADD_FUNCTION_URI = URIRef("http://example.org/myAddFunction")

@custom_function(MY_ADD_FUNCTION_URI)
def myAddFunction(a, b): # 定义时期望两个参数 a 和 b
    # SPARQL中的字面量会作为rdflib.Literal对象传入
    # 需要将其转换为Python原生类型进行计算
    try:
        val_a = int(a.toPython())
        val_b = int(b.toPython())
        return Literal(val_a + val_b)
    except (ValueError, AttributeError):
        # 处理非数字输入的情况，返回错误或默认值
        return Literal("Error: Invalid input")

# 构建SPARQL查询，调用时提供两个参数
query = f"""
SELECT ?result WHERE {{
    BIND({MY_ADD_FUNCTION_URI}(5, 6) AS ?result1)
    BIND({MY_ADD_FUNCTION_URI}(?x, 10) AS ?result2) # 示例如何与图中的数据结合
}}
"""

# 为了演示 ?x 的情况，向图中添加一些数据
g.add((URIRef("http://example.com/item1"), URIRef("http://example.com/value"), Literal(20)))
g.add((URIRef("http://example.com/item2"), URIRef("http://example.com/value"), Literal("abc"))) # 故意添加一个非数字值

query_with_data = f"""
SELECT ?s ?val ?calculatedResult WHERE {{
    ?s <http://example.com/value> ?val .
    BIND({MY_ADD_FUNCTION_URI}(?val, 10) AS ?calculatedResult)
}}
"""

print("\n--- 正确示例结果 (固定参数) ---")
for row in g.query(query):
    print(f"Result1: {row.result1}") # 预期输出 11

print("\n--- 正确示例结果 (与图中数据结合) ---")
for row in g.query(query_with_data):
    print(f"Subject: {row.s}, Value: {row.val}, Calculated: {row.calculatedResult}")