"""ArangoDB client singleton.""" import logging from arango import ArangoClient from django.conf import settings logger = logging.getLogger(__name__) _db = None def get_db(): """Get ArangoDB database connection (singleton).""" global _db if _db is None: hosts = settings.ARANGODB_INTERNAL_URL if not hosts.startswith("http"): hosts = f"http://{hosts}" client = ArangoClient(hosts=hosts) _db = client.db( settings.ARANGODB_DATABASE, username='root', password=settings.ARANGODB_PASSWORD, ) logger.info( "Connected to ArangoDB: %s/%s", hosts, settings.ARANGODB_DATABASE, ) return _db def ensure_graph(): """Ensure named graph exists for K_SHORTEST_PATHS queries.""" db = get_db() graph_name = 'optovia_graph' if db.has_graph(graph_name): return db.graph(graph_name) logger.info("Creating graph: %s", graph_name) return db.create_graph( graph_name, edge_definitions=[{ 'edge_collection': 'edges', 'from_vertex_collections': ['nodes'], 'to_vertex_collections': ['nodes'], }], )