Spaces:
Sleeping
Sleeping
File size: 4,620 Bytes
df3294b 9d88d95 df3294b 9d88d95 df3294b 9d88d95 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 |
# genesis/graphdb.py
import os
from neo4j import GraphDatabase
from typing import List, Dict, Optional
NEO4J_URI = os.getenv("NEO4J_URI")
NEO4J_USER = os.getenv("NEO4J_USER")
NEO4J_PASSWORD = os.getenv("NEO4J_PASSWORD")
class GraphDB:
def __init__(self):
if not all([NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD]):
raise ValueError("Neo4j credentials are missing from environment variables.")
self.driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
def close(self):
self.driver.close()
# ---------- Basic Helpers ----------
def run_query(self, query: str, params: Optional[Dict] = None) -> List[Dict]:
with self.driver.session() as session:
result = session.run(query, params or {})
return [record.data() for record in result]
def create_node(self, label: str, properties: Dict):
props = {k: v for k, v in properties.items() if v is not None}
query = f"MERGE (n:{label} {{name: $name}}) SET n += $props RETURN n"
return self.run_query(query, {"name": props.get("name"), "props": props})
def create_relationship(self, start_label: str, start_name: str,
rel_type: str, end_label: str, end_name: str):
query = f"""
MATCH (a:{start_label} {{name: $start_name}})
MATCH (b:{end_label} {{name: $end_name}})
MERGE (a)-[r:{rel_type}]->(b)
RETURN r
"""
return self.run_query(query, {
"start_name": start_name,
"end_name": end_name
})
# ---------- Domain-Specific Functions ----------
def add_molecule(self, name: str, description: str = None):
return self.create_node("Molecule", {"name": name, "description": description})
def add_pathway(self, name: str, description: str = None):
return self.create_node("Pathway", {"name": name, "description": description})
def add_company(self, name: str, country: str = None, funding: float = None):
return self.create_node("Company", {"name": name, "country": country, "funding": funding})
def add_trial(self, trial_id: str, title: str = None, status: str = None):
return self.create_node("Trial", {"name": trial_id, "title": title, "status": status})
def add_regulation(self, region: str, title: str, summary: str = None):
return self.create_node("Regulation", {"name": title, "region": region, "summary": summary})
def add_biosecurity_alert(self, title: str, severity: str, details: str = None):
return self.create_node("BiosecurityAlert", {"name": title, "severity": severity, "details": details})
# ---------- Relationships ----------
def link_molecule_pathway(self, molecule: str, pathway: str):
return self.create_relationship("Molecule", molecule, "INVOLVED_IN", "Pathway", pathway)
def link_company_molecule(self, company: str, molecule: str):
return self.create_relationship("Company", company, "DEVELOPS", "Molecule", molecule)
def link_trial_molecule(self, trial_id: str, molecule: str):
return self.create_relationship("Trial", trial_id, "TESTS", "Molecule", molecule)
def link_company_funder(self, company: str, funder: str):
return self.create_relationship("Company", company, "FUNDED_BY", "Funder", funder)
def link_regulation_pathway(self, regulation: str, pathway: str):
return self.create_relationship("Regulation", regulation, "REGULATES", "Pathway", pathway)
def link_biosecurity_molecule(self, alert: str, molecule: str):
return self.create_relationship("BiosecurityAlert", alert, "ASSOCIATED_WITH", "Molecule", molecule)
# ---------- Queries ----------
def get_global_funding_network(self) -> List[Dict]:
query = """
MATCH (c:Company)-[r:FUNDED_BY]->(f:Funder)
RETURN c.name AS company, f.name AS funder, c.country AS country, c.funding AS amount
ORDER BY amount DESC
"""
return self.run_query(query)
def get_pathway_network(self, pathway_name: str) -> List[Dict]:
query = """
MATCH (m:Molecule)-[:INVOLVED_IN]->(p:Pathway {name: $pathway})
OPTIONAL MATCH (m)<-[:DEVELOPS]-(c:Company)
RETURN m.name AS molecule, c.name AS company
"""
return self.run_query(query, {"pathway": pathway_name})
def get_all_entities(self) -> List[Dict]:
query = """
MATCH (n) RETURN DISTINCT labels(n) AS labels, n.name AS name
"""
return self.run_query(query)
# Singleton instance for app-wide use
graphdb = GraphDB()
|