Skip to content

symfonic.agent.fastapi.tenant_privacy_export

tenant_privacy_export

Export-side helpers for the tenant-privacy router.

Split from :mod:tenant_privacy_router to keep each file below the 300-LOC cap. Pure functions; never raise, never touch the audit log.

collect_edges_for_export async 

collect_edges_for_export(
    agent: SymfonicAgent, memory_scope: Any
) -> list[dict[str, Any]]

Return every edge owned by the tenant.

The GraphMemoryStore uniquely identifies edges (they live across all layers conceptually), so we hit the semantic layer's graph which is the canonical handle.

Source code in src/symfonic/agent/fastapi/tenant_privacy_export.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
async def collect_edges_for_export(
    agent: SymfonicAgent,
    memory_scope: Any,
) -> list[dict[str, Any]]:
    """Return every edge owned by the tenant.

    The GraphMemoryStore uniquely identifies edges (they live across
    all layers conceptually), so we hit the semantic layer's graph
    which is the canonical handle.
    """
    semantic = agent._orchestrator.get_layer(MemoryLayer.SEMANTIC)
    if semantic is None:
        return []
    graph = getattr(semantic, "_graph", None)
    if graph is None:
        return []
    try:
        edges_raw = await graph.list_edges(memory_scope, limit=100_000)
    except Exception:
        logger.debug("Export: list_edges failed", exc_info=True)
        return []
    return [
        {
            "id": str(e.id),
            "source": str(e.source),
            "target": str(e.target),
            "relationship": e.relationship,
            "weight": e.weight,
            "properties": getattr(e, "properties", {}),
        }
        for e in edges_raw
    ]

collect_nodes_for_export async 

collect_nodes_for_export(
    agent: SymfonicAgent, memory_scope: Any
) -> dict[str, list[dict[str, Any]]]

Return tenant nodes grouped by layer value.

Layers that are not registered are silently skipped (empty list entries stay out of the payload rather than implying data exists). Every layer-specific exception is logged but swallowed so a partial outage never short-circuits the rest of the export.

Source code in src/symfonic/agent/fastapi/tenant_privacy_export.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
async def collect_nodes_for_export(
    agent: SymfonicAgent,
    memory_scope: Any,
) -> dict[str, list[dict[str, Any]]]:
    """Return tenant nodes grouped by layer value.

    Layers that are not registered are silently skipped (empty list
    entries stay out of the payload rather than implying data exists).
    Every layer-specific exception is logged but swallowed so a partial
    outage never short-circuits the rest of the export.
    """
    grouped: dict[str, list[dict[str, Any]]] = {}
    for ml in MemoryLayer:
        store = agent._orchestrator.get_layer(ml)
        if store is None:
            continue
        graph = getattr(store, "_graph", None)
        if graph is None:
            # Working layer stores the graph handle under ``_graph_store``.
            graph = getattr(store, "_graph_store", None)
            if graph is None:
                continue
        try:
            # include_retracted=True: GDPR Art.20 portability must return ALL
            # of the user's data, including soft-retracted nodes not yet pruned
            # (they still contain the user's data and are hidden from the
            # default read path).
            nodes = await graph.query_nodes(
                memory_scope, layer=ml, include_retracted=True,
            )
        except Exception:
            logger.debug(
                "Export: failed to query layer %s", ml, exc_info=True,
            )
            continue
        grouped[ml.value] = [
            {
                "id": str(n.id),
                "label": str(n.label),
                "layer": n.layer.value,
                "properties": n.properties,
                "importance": n.importance,
                "access_count": n.access_count,
                "created_at": (
                    str(n.created_at) if n.created_at else None
                ),
                "updated_at": (
                    str(n.updated_at) if n.updated_at else None
                ),
            }
            for n in nodes
        ]
    return grouped