76 lines
1.5 KiB
Python
76 lines
1.5 KiB
Python
import os
|
|
from dataclasses import dataclass
|
|
|
|
from neo4j import GraphDatabase
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Neo4jSettings:
|
|
uri: str
|
|
user: str
|
|
password: str
|
|
database: str
|
|
|
|
|
|
def get_neo4j_settings() -> Neo4jSettings:
|
|
password = os.getenv("NEO4J_PASSWORD")
|
|
|
|
if not password:
|
|
raise RuntimeError(
|
|
"NEO4J_PASSWORD is not configured."
|
|
)
|
|
|
|
return Neo4jSettings(
|
|
uri=os.getenv(
|
|
"NEO4J_URI",
|
|
"bolt://localhost:7687",
|
|
),
|
|
user=os.getenv(
|
|
"NEO4J_USER",
|
|
"neo4j",
|
|
),
|
|
password=password,
|
|
database=os.getenv(
|
|
"NEO4J_DATABASE",
|
|
"neo4j",
|
|
),
|
|
)
|
|
|
|
|
|
def create_neo4j_driver(
|
|
settings: Neo4jSettings | None = None,
|
|
):
|
|
resolved = settings or get_neo4j_settings()
|
|
|
|
return GraphDatabase.driver(
|
|
resolved.uri,
|
|
auth=(
|
|
resolved.user,
|
|
resolved.password,
|
|
),
|
|
)
|
|
|
|
|
|
def verify_neo4j_connection() -> Neo4jSettings:
|
|
settings = get_neo4j_settings()
|
|
|
|
with create_neo4j_driver(settings) as driver:
|
|
driver.verify_connectivity()
|
|
|
|
with driver.session(
|
|
database=settings.database
|
|
) as session:
|
|
record = session.run(
|
|
"RETURN 1 AS value"
|
|
).single()
|
|
|
|
if (
|
|
record is None
|
|
or record["value"] != 1
|
|
):
|
|
raise RuntimeError(
|
|
"Neo4j connectivity check failed."
|
|
)
|
|
|
|
return settings
|