-
Notifications
You must be signed in to change notification settings - Fork 4
/
Copy paths4a_pathway_subpathways.py
executable file
·51 lines (39 loc) · 2.14 KB
/
s4a_pathway_subpathways.py
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
#!/usr/bin/env python
"""Retrieving pathways, subpathways, and superpathways."""
# https://reactome.org/dev/graph-database/extract-participating-molecules#retrieving-pathways
from __future__ import print_function
__copyright__ = "Copyright (C) 2018-present, DV Klopfenstein. All rights reserved."
__author__ = "DV Klopfenstein"
import sys
from neo4j import GraphDatabase
def main(password, prt=sys.stdout):
"""Retrieving the reactions for a given pathway."""
# To find out its subpathways, the slot to query is "hasEvent":
# Direct subpathways for the pathway with stable identifier R-HSA-983169
qry = ('MATCH (p:Pathway{stId:"R-HSA-983169"})-[:hasEvent]->(sp:Pathway)'
'RETURN p.stId AS Pathway, sp.stId AS SubPathway, sp.displayName as DisplayName')
data = _get_data(qry, password)
prt.write('\n{N} subpathways directly in Pathway(R-HSA-983169)'.format(N=len(data)))
_prt_data(data, prt)
# It is important to note that subpathways might contain other subpathways,
# so to get ALL the subpathways of R-HSA-198933, the query is as follows:
qry = ('MATCH (p:Pathway{stId:"R-HSA-983169"})-[:hasEvent*]->(sp:Pathway)'
'RETURN p.stId AS Pathway, sp.stId AS SubPathway, sp.displayName as DisplayName')
prt.write('\nALL {N} subpathways in Pathway(R-HSA-983169)'.format(N=len(data)))
data = _get_data(qry, password)
_prt_data(data, prt)
def _prt_data(data, prt):
"""Print the reactions for a given pathway."""
prt.write('Pathway SubPathway DisplayName\n')
prt.write('------------ -------------- ------------\n')
for dct in sorted(data, key=lambda d: [d['Pathway'], d['SubPathway']]):
prt.write('{Pathway:13} {SubPathway:13} {DisplayName}\n'.format(**dct))
def _get_data(qry, password):
"""Run query. Collect data."""
gdbdr = GraphDatabase.driver('bolt://localhost:7687', auth=('neo4j', password))
with gdbdr.session() as session:
return session.run(qry).data()
if __name__ == '__main__':
assert len(sys.argv) != 1, 'First arg must be your Neo4j database password'
main(sys.argv[1])
# Copyright (C) 2018-present, DV Klopfenstein. All rights reserved.