63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
"""
|
|
FILE: app/core/graph/graph_db_adapter.py
|
|
DESCRIPTION: Datenbeschaffung aus Qdrant für den Graphen.
|
|
AUDIT v1.1.0: Nutzt nun die zentrale database-Infrastruktur für Namen.
|
|
"""
|
|
from typing import List, Dict, Optional
|
|
from qdrant_client import QdrantClient
|
|
from qdrant_client.http import models as rest
|
|
|
|
# ENTSCHEIDENDER FIX: Nutzt die neue Infrastruktur für konsistente Collection-Namen
|
|
from app.core.database import collection_names
|
|
|
|
def fetch_edges_from_qdrant(
|
|
client: QdrantClient,
|
|
prefix: str,
|
|
seeds: List[str],
|
|
edge_types: Optional[List[str]] = None,
|
|
limit: int = 2048,
|
|
) -> List[Dict]:
|
|
"""
|
|
Holt Edges aus der Datenbank basierend auf Seed-IDs.
|
|
Filtert auf source_id, target_id oder note_id.
|
|
"""
|
|
if not seeds or limit <= 0:
|
|
return []
|
|
|
|
# Konsistente Namensauflösung via database-Paket
|
|
_, _, edges_col = collection_names(prefix)
|
|
|
|
seed_conditions = []
|
|
for field in ("source_id", "target_id", "note_id"):
|
|
for s in seeds:
|
|
seed_conditions.append(
|
|
rest.FieldCondition(key=field, match=rest.MatchValue(value=str(s)))
|
|
)
|
|
seeds_filter = rest.Filter(should=seed_conditions) if seed_conditions else None
|
|
|
|
type_filter = None
|
|
if edge_types:
|
|
type_conds = [
|
|
rest.FieldCondition(key="kind", match=rest.MatchValue(value=str(k)))
|
|
for k in edge_types
|
|
]
|
|
type_filter = rest.Filter(should=type_conds)
|
|
|
|
must = []
|
|
if seeds_filter:
|
|
must.append(seeds_filter)
|
|
if type_filter:
|
|
must.append(type_filter)
|
|
|
|
flt = rest.Filter(must=must) if must else None
|
|
|
|
# Abfrage via Qdrant Scroll API
|
|
pts, _ = client.scroll(
|
|
collection_name=edges_col,
|
|
scroll_filter=flt,
|
|
limit=limit,
|
|
with_payload=True,
|
|
with_vectors=False,
|
|
)
|
|
|
|
return [dict(p.payload) for p in pts if p.payload] |