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
108
109
110
|
#!/usr/bin/env python3
# This script is used to evaluate Nixpkgs and store the relationships between
# derivations and their input derivations in a directed acyclic graph (DAG).
# The DAG is represented using an adjacency list and saved in a SQLite db
import re
import sys
import subprocess
import json
import sqlite3
from typing import Set
class drv:
pname: str
input_drvs: Set[str] = set()
cursor: sqlite3.Cursor
def __init__(self, drv_string: str, cursor: sqlite3.Cursor) -> None:
self.cursor = cursor
j = json.loads(drv_string)
if 'error' in j:
raise TypeError(f'{j['attrPath']}: Failed to evaluate')
elif 'drvPath' not in j:
raise TypeError(f'{j['attrPath']}: Failed to read drvPath')
pname = self.pname_from_drv_path(j['drvPath'])
print(pname)
if pname is None:
raise TypeError(f'{j['attrPath']}: Failed to read pname')
self.pname = pname
for input_drv in j['inputDrvs']:
pname = self.pname_from_drv_path(input_drv)
if pname is not None:
self.input_drvs.add(pname)
def db_push(self):
parrent_id = self.rowid_from_pname(self.cursor, self.pname)
for input_drv in self.input_drvs:
child_id = self.rowid_from_pname(self.cursor, input_drv)
self.cursor.execute("""
INSERT INTO edges (start, end)
VALUES (?, ?)
""", (parrent_id, child_id))
@staticmethod
def rowid_from_pname(cursor: sqlite3.Cursor, pname: str) -> int:
s = cursor.execute("""
SELECT pnames.ROWID FROM pnames
WHERE pnames.pname = ?
""", (pname,))
id = s.fetchone()
if id:
return id[0]
s = cursor.execute("""
INSERT INTO pnames (pname)
VALUES (?)
""", (pname,))
if s.lastrowid is None:
raise TypeError('Failed to get lastrowid')
return s.lastrowid
@staticmethod
def pname_from_drv_path(drv_path: str) -> str | None:
f = open(drv_path, 'r')
drv_string = f.readline()
match = re.search('"pname","([^"]+)', drv_string)
if match is not None:
return match.group(1)
if __name__ == '__main__':
cmd = [
'nix-eval-jobs',
'--flake',
'github:nixos/nixpkgs#legacyPackages.x86_64-linux'
]
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL)
if proc.stdout is None:
raise EOFError('Failed to evaluate nixpkgs')
con = sqlite3.connect('nixpkgs_dag.db')
cur = con.cursor()
cur.execute("""
CREATE TABLE IF NOT EXISTS pnames (
pname TEXT NOT NULL UNIQUE
)
""")
cur.execute("""
CREATE TABLE IF NOT EXISTS edges (
start INTEGER NOT NULL,
end NOT NULL,
UNIQUE(start, end) ON CONFLICT REPLACE
)
""")
for line in proc.stdout:
try:
d = drv(line.decode('utf-8'), cur)
d.db_push()
except Exception as e:
print(f'>>> {e}', file=sys.stderr)
con.commit()
con.close()
|
