在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(arg1, arg2):
    """
    这是一个接收两个参数并返回它们之和的自定义函数。
    注意：arg1和arg2在Python函数中将是rdflib.Literal对象。
    """
    try:
        # 尝试将Literal值转换为Python原生类型进行计算
        val1 = int(arg1.value) if isinstance(arg1, Literal) else arg1
        val2 = int(arg2.value) if isinstance(arg2, Literal) else arg2
        return Literal(val1 + val2)
    except (ValueError, TypeError):
        # 处理非数字参数的情况，例如返回一个错误Literal或None
        return Literal("Error: Non-numeric arguments provided")

# 如果函数不需要任何参数，则定义时也不应有参数（或使用*args处理）
@custom_function(URIRef("http://example.org/noArgFunction"))
def noArgFunction():
    return Literal("This function takes no arguments.")

SELECT ?result WHERE {
    # 调用自定义函数，并将其结果绑定到?result变量
    BIND(<http://example.org/myCustomFunction>(5, 6) AS ?result)
}

# Python定义 (期望两个参数)
@custom_function(URIRef("http://example.org/myCustomFunction"))
def myCustomFunction(arg1, arg2):
    return Literal(arg1 + arg2)

# SPARQL调用 (不带参数)
query_error = """
SELECT ?result WHERE {
    BIND(<http://example.org/myCustomFunction>() AS ?result) # 错误：期望两个参数，但提供了零个
}
"""
# 执行此查询将不会返回任何结果，因为参数数量不匹配。

# Python定义 (期望两个参数)
@custom_function(URIRef("http://example.org/myCustomFunction"))
def myCustomFunction(arg1, arg2):
    # arg1和arg2将是rdflib.Literal对象，其值可以通过.value属性访问
    # 对于简单的数值运算，rdflib.Literal对象可以直接相加，但最好转换为Python原生类型
    try:
        val1 = int(arg1.value)
        val2 = int(arg2.value)
        return Literal(val1 + val2)
    except (ValueError, TypeError):
        return Literal("Error: Invalid arguments")

# SPARQL调用 (提供两个参数)
query_correct = """
SELECT ?result WHERE {
    BIND(<http://example.org/myCustomFunction>(5, 6) AS ?result) # 正确：提供了两个参数
}
"""
# 执行此查询将返回结果 `11`

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

# 1. 初始化RDF图
g = Graph()

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

# 3. 使用@custom_function装饰器定义Python函数
@custom_function(MY_ADD_FUNCTION_URI)
def myAddFunction(a, b):
    """
    一个简单的加法函数，接收两个rdflib.Literal参数，并返回它们的和。
    """
    try:
        # 将Literal值转换为Python原生类型进行计算
        # 注意：SPARQL中的数字字面量会被rdflib解析为带有XSD数据类型的Literal
        val_a = a.toPython() if isinstance(a, Literal) else a
        val_b = b.toPython() if isinstance(b, Literal) else b
        return Literal(val_a + val_b, datatype=XSD.integer)
    except Exception as e:
        print(f"Error in myAddFunction: {e}")
        return Literal(f"Calculation Error: {e}")

# 4. 构造SPARQL查询，调用自定义函数
# 注意：SPARQL中传递的字面量（如5, 6）会被自动转换为rdflib.Literal对象传递给Python函数
query = """
SELECT ?sumResult WHERE {
    BIND(<http://example.org/myAddFunction>(5, 6) AS ?sumResult)
}
"""

# 5. 执行查询并打印结果
print("Executing query with custom function:")
for row in g.query(query):
    print(f"Sum Result: {row.sumResult}")

print("\n--- Another example with different arguments ---")
query_sub = """
SELECT ?subResult WHERE {
    BIND(<http://example.org/myAddFunction>(10, -3) AS ?subResult)
}
"""
for row in g.query(query_sub):
    print(f"Another Sum Result: {row.subResult}")

# 演示一个无参数函数
NO_ARG_FUNCTION_URI = URIRef("http://example.org/getCurrentTime")

@custom_function(NO_ARG_FUNCTION_URI)
def getCurrentTime():
    import datetime
    return Literal(datetime.datetime.now().isoformat())

query_time = """
SELECT ?currentTime WHERE {
    BIND(<http://example.org/getCurrentTime>() AS ?currentTime)
}
"""
print("\n--- Querying current time ---")
for row in g.query(query_time):
    print(f"Current Time: {row.currentTime}")