aboutsummaryrefslogtreecommitdiff
path: root/bin/nixpkgs_dag.py
diff options
context:
space:
mode:
Diffstat (limited to 'bin/nixpkgs_dag.py')
-rwxr-xr-xbin/nixpkgs_dag.py110
1 files changed, 0 insertions, 110 deletions
diff --git a/bin/nixpkgs_dag.py b/bin/nixpkgs_dag.py
deleted file mode 100755
index 4f574f0..0000000
--- a/bin/nixpkgs_dag.py
+++ /dev/null
@@ -1,110 +0,0 @@
-#!/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()