求助：Redis Graph无法通过Python为节点/边设置自定义TTL

由于Redis Graph本身不支持为单个节点原生设置TTL，结合你的Django+django-redis==5.2.0场景，以下两种方案可以实现自定义节点TTL需求：

方案一：节点内置过期时间字段+定期清理 #

通过在节点中存储过期时间戳，配合定时任务删除过期节点，适合大多数场景。

1. 创建带TTL的节点 #

在CREATE查询中加入expire_at字段，存储该节点的过期Unix时间戳（当前时间+自定义TTL秒数）：

from django_redis import get_redis_connection
import time

redis_conn = get_redis_connection("default")

# 自定义TTL（示例：1小时过期）
custom_ttl = 3600
expire_at = int(time.time()) + custom_ttl

# 执行Graph查询，传入过期时间参数
query_result = redis_conn.execute_command(
    "GRAPH.QUERY", "你的图名称",
    "CREATE (n:label {a:1, expire_at: $expire_at}) RETURN n",
    "--params", f'{{"expire_at": {expire_at}}}'
)

2. 定时清理过期节点 #

使用Django的Celery Beat或系统Cron，定期执行清理查询，删除expire_at小于当前时间的节点：

from celery import shared_task
from django_redis import get_redis_connection
import time

@shared_task
def cleanup_expired_graph_nodes():
    redis_conn = get_redis_connection("default")
    current_time = int(time.time())
    # 按标签批量清理，每次处理1000个避免阻塞
    labels = ["label", "another_label"]  # 替换为你的节点标签
    
    for label in labels:
        while True:
            clean_result = redis_conn.execute_command(
                "GRAPH.QUERY", "你的图名称",
                f"MATCH (n:{label}) WHERE n.expire_at < $current_time WITH n LIMIT 1000 DELETE n",
                "--params", f'{{"current_time": {current_time}}}'
            )
            # 当没有节点可删除时退出循环
            if "Nodes deleted: 0" in str(clean_result):
                break

方案二：Redis有序集合（ZSET）追踪节点过期时间 #

通过ZSET存储节点ID与过期时间的映射，更精准地管理节点生命周期，适合对过期时间要求严格的场景。

1. 创建节点并记录过期信息 #

先创建节点获取ID，再将ID存入ZSET（分数为过期时间戳）：

from django_redis import get_redis_connection
import time

redis_conn = get_redis_connection("default")
custom_ttl = 3600
expire_at = int(time.time()) + custom_ttl

# 创建节点并返回ID
create_result = redis_conn.execute_command(
    "GRAPH.QUERY", "你的图名称",
    "CREATE (n:label {a:1}) RETURN ID(n)"
)
# 解析查询结果获取节点ID（Redis Graph结果格式为嵌套列表，需根据实际结构调整）
node_id = create_result[1][0][0]

# 将节点ID存入ZSET，键名按标签区分
redis_conn.zadd(f"graph:expire:label", {node_id: expire_at})

2. 清理过期节点 #

从ZSET中取出过期的节点ID，批量删除后清理ZSET中的记录：

from celery import shared_task
from django_redis import get_redis_connection
import time

@shared_task
def cleanup_expired_graph_nodes():
    redis_conn = get_redis_connection("default")
    current_time = int(time.time())
    zset_keys = ["graph:expire:label", "graph:expire:another_label"]  # 对应你的ZSET键名
    
    for zset_key in zset_keys:
        # 获取所有过期的节点ID
        expired_ids = redis_conn.zrangebyscore(zset_key, 0, current_time)
        if not expired_ids:
            continue
        
        # 转换为整数列表用于Graph查询
        expired_id_list = [int(node_id) for node_id in expired_ids]
        # 批量删除过期节点
        redis_conn.execute_command(
            "GRAPH.QUERY", "你的图名称",
            "MATCH (n) WHERE ID(n) IN $expired_ids DELETE n",
            "--params", f'{{"expired_ids": {expired_id_list}}}'
        )
        # 清理ZSET中的过期记录
        redis_conn.zremrangebyscore(zset_key, 0, current_time)

注意事项 #

• 方案一实现简单，无需额外存储，但过期节点不会立即删除，需依赖清理任务频率；
• 方案二更精准，但需要维护ZSET与Graph节点的一致性，避免节点已删除但ZSET记录未清理的情况；
• 清理任务的执行频率需根据业务中最短TTL设置，比如最短TTL为5分钟，则每5分钟执行一次清理。

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