Executes a read-only Cypher query.
(query: str, context: Optional[str] = None)
| 531 | |
| 532 | |
| 533 | def cypher_helper(query: str, context: Optional[str] = None): |
| 534 | """Executes a read-only Cypher query.""" |
| 535 | services = _initialize_services(context) |
| 536 | if not all(services[:3]): |
| 537 | _fail_services_init() |
| 538 | |
| 539 | db_manager, _, _, ctx = services |
| 540 | |
| 541 | from ..utils.cypher_readonly import is_read_only_cypher, read_only_rejection_message |
| 542 | |
| 543 | if not is_read_only_cypher(query): |
| 544 | console.print(f"[bold red]Error:[/bold red] {read_only_rejection_message()}") |
| 545 | db_manager.close_driver() |
| 546 | raise typer.Exit(code=1) |
| 547 | |
| 548 | backend = getattr(db_manager, "get_backend_type", lambda: "neo4j")() |
| 549 | # Neo4j honours default_access_mode natively; FalkorDB routes READ sessions |
| 550 | # through GRAPH.RO_QUERY for server-enforced read-only execution. |
| 551 | session_kwargs = ( |
| 552 | {"default_access_mode": "READ"} |
| 553 | if backend in ("neo4j", "falkordb", "falkordb-remote") |
| 554 | else {} |
| 555 | ) |
| 556 | |
| 557 | try: |
| 558 | with db_manager.get_driver().session(**session_kwargs) as session: |
| 559 | result = session.run(query) |
| 560 | records = [record.data() for record in result] |
| 561 | console.print(json.dumps(records, indent=2)) |
| 562 | except Exception as e: |
| 563 | _print_query_exception(e, query) |
| 564 | db_manager.close_driver() |
| 565 | raise typer.Exit(code=1) |
| 566 | finally: |
| 567 | db_manager.close_driver() |
| 568 | |
| 569 | |
| 570 | def cypher_helper_visual(query: str, context: Optional[str] = None): |
no test coverage detected