(scan: dict, cursor)
| 129 | |
| 130 | |
| 131 | def add_scan(scan: dict, cursor): |
| 132 | try: |
| 133 | pkg_name = scan["metadata"]["uri_input"]["package"] |
| 134 | print(f"Processing package: `{pkg_name}`") |
| 135 | |
| 136 | cursor.execute(""" |
| 137 | INSERT INTO scans (package_name, metadata) VALUES (?, ?) |
| 138 | """, (pkg_name, rapidjson.dumps(scan["metadata"]))) |
| 139 | except sqlite3.IntegrityError: |
| 140 | # Data for this package already exists: |
| 141 | return |
| 142 | |
| 143 | scan_id = cursor.lastrowid |
| 144 | for d in scan.get("detections", []): |
| 145 | add_detection(scan_id, d, pkg_name, cursor) |
| 146 | |
| 147 | |
| 148 | def process_scans(dataset_path, cursor): |
no test coverage detected