Generate a CSV file of dependencies from Cargo.toml and Cargo.lock

cargo_list.py

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())