52 lines
1.9 KiB
Python
52 lines
1.9 KiB
Python
from collections.abc import Sequence
|
|
from typing import Any, Protocol
|
|
|
|
|
|
class GraphDriver(Protocol):
|
|
async def execute_query(self, query: str, **parameters: Any) -> Sequence[Any]: ...
|
|
|
|
|
|
class GraphDegradedResult:
|
|
def __init__(self, reason: str) -> None:
|
|
self.degraded = True
|
|
self.reason = reason
|
|
|
|
|
|
class RelationshipService:
|
|
"""Controlled Neo4j boundary; callers cannot submit arbitrary labels or Cypher."""
|
|
|
|
ALLOWED_RELATIONSHIPS = frozenset({
|
|
"PREFERS", "HAS_GOAL", "INTERESTED_IN", "TRADED", "HOLDS",
|
|
"TRIGGERED_RISK", "BELONGS_TO_CATEGORY", "EXPOSED_TO_INDUSTRY",
|
|
})
|
|
|
|
def __init__(self, driver: GraphDriver) -> None:
|
|
self.driver = driver
|
|
|
|
async def neighbors(
|
|
self, customer_id: int, relationship: str, *, limit: int = 50
|
|
) -> Sequence[Any]:
|
|
if relationship not in self.ALLOWED_RELATIONSHIPS:
|
|
raise ValueError("relationship is not allowed")
|
|
safe_limit = max(1, min(limit, 100))
|
|
query = (
|
|
"MATCH (c:Customer {customer_id: $customer_id})"
|
|
f"-[r:{relationship}]->(n) RETURN n, r LIMIT $limit"
|
|
)
|
|
try:
|
|
return await self.driver.execute_query(query, customer_id=customer_id, limit=safe_limit)
|
|
except Exception as exc:
|
|
return [GraphDegradedResult(f"neo4j_unavailable:{type(exc).__name__}")]
|
|
|
|
async def paths(self, customer_id: int, *, max_hops: int = 2, limit: int = 50) -> Sequence[Any]:
|
|
hops = max(1, min(max_hops, 2))
|
|
safe_limit = max(1, min(limit, 100))
|
|
query = (
|
|
"MATCH p=(c:Customer {customer_id: $customer_id})-[*1.."
|
|
f"{hops}]->(n) RETURN p LIMIT $limit"
|
|
)
|
|
try:
|
|
return await self.driver.execute_query(query, customer_id=customer_id, limit=safe_limit)
|
|
except Exception as exc:
|
|
return [GraphDegradedResult(f"neo4j_unavailable:{type(exc).__name__}")]
|