Last active
July 6, 2024 08:48
-
-
Save samuelcolvin/da7d043df79a9ad212d3488073e169f8 to your computer and use it in GitHub Desktop.
Generate a CSV file of dependencies from Cargo.toml and Cargo.lock
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from __future__ import annotations | |
| import asyncio | |
| import csv | |
| import sys | |
| from dataclasses import dataclass, asdict, fields | |
| from pathlib import Path | |
| from typing import Literal | |
| import toml | |
| from httpx import AsyncClient | |
| from rich.console import Console | |
| from rich.progress import Progress | |
| type Designation = Literal['server - rust', 'development - rust', 'open source - rust'] | |
| # PROD_DESIGNATION: Designation = 'server - rust' | |
| PROD_DESIGNATION: Designation = 'open source - rust' | |
| DEV_DESIGNATION: Designation = 'development - rust' | |
| @dataclass | |
| class Package: | |
| name: str | |
| version: str | |
| license: str | |
| link: str | |
| designation: Designation | |
| modified: Literal['Yes', 'Non'] = 'No' | |
| async def get_package( | |
| console: Console, client: AsyncClient, name: str, version: str, prod: bool = False | |
| ) -> Package | None: | |
| response = await client.get(f'https://crates.io/api/v1/crates/{name}') | |
| if response.status_code == 404: | |
| console.print(f'[red]Package {name} not found on crates.io[/red]') | |
| return None | |
| response.raise_for_status() | |
| data = response.json() | |
| version_info = next((v for v in data['versions'] if v['num'] == version), None) | |
| if not version_info: | |
| console.print(f'[red]Version {version} not found for package {name}[/red]') | |
| else: | |
| return Package( | |
| name=name, | |
| version=version, | |
| license=version_info['license'], | |
| link=data['crate'].get('homepage') or f'https://crates.io/crates/{name}', | |
| designation=PROD_DESIGNATION if prod else DEV_DESIGNATION, | |
| ) | |
| async def main(): | |
| assert len(sys.argv) == 4, 'Usage: cargo_list.py Cargo.toml Cargo.lick <output.csv>' | |
| cargo_toml_path = Path(sys.argv[1]) | |
| with cargo_toml_path.open() as f: | |
| cargo_toml = toml.load(f) | |
| cargo_lock_path = Path(sys.argv[2]) | |
| with cargo_lock_path.open() as f: | |
| cargo_lock = toml.load(f) | |
| lock_packages = cargo_lock['package'] | |
| prod_packages: set[str] = set() | |
| visited: set[str] = set() | |
| lock_tree: dict[str, list[str]] = {} | |
| for p in lock_packages: | |
| name = p['name'] | |
| deps = p.get('dependencies', []) | |
| if existing := lock_tree.get(name): | |
| existing.extend(deps) | |
| else: | |
| lock_tree[name] = deps | |
| def add_to_prod(p: str): | |
| if p in visited: | |
| return | |
| visited.add(p) | |
| prod_packages.add(p) | |
| for dep in lock_tree.get(p, []): | |
| add_to_prod(dep.split(' ')[0]) | |
| for package in cargo_toml.get('dependencies', {}).keys(): | |
| add_to_prod(package) | |
| csv_file = Path(sys.argv[3]) | |
| known_names: set[str] = set() | |
| csv_existed = csv_file.exists() | |
| if csv_existed: | |
| # Read the existing CSV file to avoid duplicate entries | |
| with csv_file.open() as file: | |
| reader = csv.DictReader(file) | |
| for row in reader: | |
| known_names.add(row['name']) | |
| async with AsyncClient() as client: | |
| with csv_file.open(mode='a', newline='') as file: | |
| writer = csv.DictWriter(file, fieldnames=[f.name for f in fields(Package)]) | |
| if not csv_existed: | |
| writer.writeheader() | |
| with Progress() as progress: | |
| for package in progress.track(lock_packages, description='Fetching package info...'): | |
| name = package['name'] | |
| if name in known_names: | |
| continue | |
| progress.console.print(name) | |
| package = await get_package( | |
| progress.console, client, name, package['version'], name in prod_packages | |
| ) | |
| if package: | |
| writer.writerow(asdict(package)) | |
| if __name__ == '__main__': | |
| asyncio.run(main()) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment