使用plpython3u在PostgreSQL中访问AGE边属性的问题排查

我写的Python查询函数单独运行能得到正确结果，但通过PostgreSQL的plpython3u扩展创建函数后，无报错却无法返回结果。移除以下航班时间连续性判断代码段后，函数可正常返回所有路径，因此怀疑是边属性访问方式有误：

for i in range(len(flight) - 1):
     arrival_time = datetime.strptime(flight[i]["arrival_time"], '%Y-%m-%d %H:%M')
     next_departure_time=datetime.strptime(flight[i+1]["departure_time"], '%Y-%m-%d %H:%M')
     if arrival_time < next_departure_time:
            is_continuous = False
            break

附上完整函数代码：

CREATE OR REPLACE FUNCTION graph.paircpath(
    origin text,
    edge text,
    destination text)
    RETURNS TABLE(city1 ag_catalog.agtype, airport1 ag_catalog.agtype, flight ag_catalog.agtype, airport2 ag_catalog.agtype, city2 ag_catalog.agtype) 
    LANGUAGE 'plpython3u'
    COST 100
    VOLATILE PARALLEL UNSAFE
    ROWS 1000

AS $BODY$
    from datetime import datetime
    import age
    import psycopg2

    def paircpath(origin, edge, destination):
        conn = psycopg2.connect(host="localhost", port="5432", dbname="postgres", user="postgres", password="13711992")
        with conn.cursor() as cursor:
            try:
                cursor.execute("SET search_path = ag_catalog, public, graph;")
                cursor.execute("LOAD 'age';")
                cursor.execute("GRANT USAGE ON SCHEMA ag_catalog TO postgres;")

                query = f"""SELECT * FROM cypher('graph', $$ 
                    MATCH (a)-[:LocatedAt*]->(c:City {{name: '{origin}'}})
                    MATCH (a:Airport)-[e:{edge}]->(b:Airport) 
                    MATCH (b)-[:LocatedAt*]->(c1:City {{name: '{destination}'}}) 
                    RETURN c, a, e, b, c1 
                $$) AS (city1 agtype, airport1 agtype, flight agtype, airport2 agtype, city2 agtype); """

                cursor.execute(query)
                paths = cursor.fetchall()

                for row in paths:
                    city1 = row[0]
                    airport1 = row[1]
                    flight = row[2]
                    airport2 = row[3]
                    city2 = row[4]
                    is_continuous = True
                    for i in range(len(flight) - 1):
                        arrival_time = datetime.strptime(flight[i]["arrival_time"], '%Y-%m-%d %H:%M')
                        next_departure_time = datetime.strptime(flight[i+1]["departure_time"], '%Y-%m-%d %H:%M')
                        if arrival_time < next_departure_time:
                            is_continuous = False
                            break
                    if is_continuous:
                        yield (city1, airport1, flight, airport2, city2)

            except Exception as ex:
                print(type(ex), ex)

    for result in paircpath(origin, edge, destination):
        yield result

$BODY$;

ALTER FUNCTION graph.paircpath(text, text, text)
    OWNER TO postgres;

问题原因 #

从Cypher查询返回的flight是ag_catalog.agtype类型，在PL/Python环境中它不是原生的Python列表或字典，直接按索引flight[i]访问会触发隐性错误（无报错但逻辑不执行），最终没有符合条件的结果返回。

解决方法 #

需要先用age.loads()方法将agtype对象转换为Python可操作的原生数据结构，再进行属性访问。修改后的完整代码如下：

CREATE OR REPLACE FUNCTION graph.paircpath(
    origin text,
    edge text,
    destination text)
    RETURNS TABLE(city1 ag_catalog.agtype, airport1 ag_catalog.agtype, flight ag_catalog.agtype, airport2 ag_catalog.agtype, city2 ag_catalog.agtype) 
    LANGUAGE 'plpython3u'
    COST 100
    VOLATILE PARALLEL UNSAFE
    ROWS 1000

AS $BODY$
    from datetime import datetime
    import age
    import psycopg2

    def paircpath(origin, edge, destination):
        conn = psycopg2.connect(host="localhost", port="5432", dbname="postgres", user="postgres", password="13711992")
        with conn.cursor() as cursor:
            try:
                cursor.execute("SET search_path = ag_catalog, public, graph;")
                cursor.execute("LOAD 'age';")
                cursor.execute("GRANT USAGE ON SCHEMA ag_catalog TO postgres;")

                query = f"""SELECT * FROM cypher('graph', $$ 
                    MATCH (a)-[:LocatedAt*]->(c:City {{name: '{origin}'}})
                    MATCH path=(a:Airport)-[e:{edge}*]->(b:Airport) 
                    MATCH (b)-[:LocatedAt*]->(c1:City {{name: '{destination}'}}) 
                    RETURN c, a, relationships(path), b, c1 
                $$) AS (city1 agtype, airport1 agtype, flight agtype, airport2 agtype, city2 agtype); """

                cursor.execute(query)
                paths = cursor.fetchall()

                for row in paths:
                    city1 = row[0]
                    airport1 = row[1]
                    flight_agtype = row[2]
                    airport2 = row[3]
                    city2 = row[4]
                    is_continuous = True
                    # 将agtype转换为Python列表
                    flight_list = age.loads(flight_agtype)
                    # 单段航班无需判断连续性
                    if len(flight_list) > 1:
                        for i in range(len(flight_list) - 1):
                            # 正确访问边的属性
                            arrival_time = datetime.strptime(flight_list[i]["properties"]["arrival_time"], '%Y-%m-%d %H:%M')
                            next_departure_time = datetime.strptime(flight_list[i+1]["properties"]["departure_time"], '%Y-%m-%d %H:%M')
                            # 修正逻辑：前序航班到达时间晚于后续出发时间才不符合连续性
                            if arrival_time > next_departure_time:
                                is_continuous = False
                                break
                    if is_continuous:
                        yield (city1, airport1, flight_agtype, airport2, city2)

            except Exception as ex:
                # 抛出异常便于调试
                raise Exception(f"Error: {type(ex).__name__}, {str(ex)}")

    for result in paircpath(origin, edge, destination):
        yield result

$BODY$;

ALTER FUNCTION graph.paircpath(text, text, text)
    OWNER TO postgres;

关键修改点： #

1. Cypher查询调整：将原查询中的[e:{edge}]改为[e:{edge}*]并使用relationships(path)获取路径中的所有边，确保能拿到多段航班的边列表。
2. agtype转换：用age.loads(flight_agtype)将AG边对象转换为Python列表，每个边元素是包含properties键的字典。
3. 属性访问方式：通过flight_list[i]["properties"]["arrival_time"]访问边的属性，替代原错误的直接索引访问。
4. 逻辑修正：原代码中判断arrival_time < next_departure_time就标记为不连续，逻辑完全相反，已修正为前序航班到达时间晚于后续出发时间才判定为不连续。
5. 异常处理优化：将print改为raise，方便在PostgreSQL中查看具体错误信息，快速定位问题。

内容的提问来源于stack exchange，提问作者Jaber