hpcman.software.db¶

class Software(SQLModel, table=True):  # type: ignore[call-arg]
    __table_args__ = (UniqueConstraint("name", "arch", name="unique_software"),)
    model_config = ConfigDict(arbitrary_types_allowed=True, validate_assignment=True)  # type: ignore
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    arch: CPUArch = Field(sa_type=ArchSQLStr)
    versions: list["InstalledVersion"] = Relationship(back_populates="software")
    updated: datetime
    latest: VerStr = Field(sa_type=VerSQLStr)
    provides: list[str] | None = Field(sa_type=StrListSQLStr, nullable=True)
    libs: list[str] | None = Field(sa_type=StrListSQLStr, nullable=True)
    tags: list[str] | None = Field(sa_type=StrListSQLStr, nullable=True)
    category: Category = Field(sa_type=CategorySQLStr)
    docs: list[AnyUrl] | None = Field(sa_type=URLListSQLStr, nullable=True)
    refs: list[AnyUrl] | None = Field(sa_type=URLListSQLStr, nullable=True)
    version_in_path: VerStr | None = Field(sa_type=VerSQLStr, nullable=True)

class InstalledVersion(SQLModel, table=True):  # type: ignore[call-arg]
    model_config = ConfigDict(arbitrary_types_allowed=True, validate_assignment=True)  # type: ignore
    id: int | None = Field(default=None, primary_key=True)
    version: VerStr = Field(sa_type=VerSQLStr)
    software_id: int | None = Field(default=None, foreign_key="software.id")
    software: Software = Relationship(back_populates="versions")
    installed: datetime
    method: InstallMethod = Field(sa_type=MethodSQLStr)
    prefix: Path = Field(sa_type=PathSQLStr)
    exes: list[Path] = Field(sa_type=PathListSQLStr)
    pkgs: list[str] | None = Field(sa_type=StrListSQLStr)
    version_message: str | None
    help_message: str | None
    linked_dir: bool
    linked_exes: bool
    linked_to: Path | None = Field(sa_type=PathSQLStr, nullable=True)

def start_engine(db_path: Path | None, db_type: Literal["sqlite"] = "sqlite", debug: bool = False) -> Engine:
    """Creates engine with sqlmodel and returns it"""
    if db_type == "sqlite":
        if db_path is None:
            file_url = "sqlite://"
        else:
            file_url = f"sqlite:///{db_path.resolve()}"
    else:
        rprint(f"Given db_type '{db_type}' is not supported. Only sqlite is allowed.")
        exit(1)

    engine = create_engine(file_url, echo=debug)
    create_db_and_tables(engine)

    return engine

def create_db_and_tables(engine: Engine) -> None:
    SQLModel.metadata.create_all(engine)

def add_software_row(
    session: Session,
    name: str,
    updated: datetime,
    version: Version,
    exe_names: list[str],
    tags: list[str],
    category: Category,
    docs: list[AnyUrl] | None,
    refs: list[AnyUrl] | None,
    linked_exes: bool,
    libs: list[str] | None,
    arch: CPUArch,
) -> Software:
    software = Software(
        name=name,
        updated=updated,
        latest=version,
        arch=arch,
        provides=exe_names,
        libs=libs,
        tags=tags,
        category=category,
        docs=docs,
        refs=refs,
        version_in_path=version if linked_exes else None,
    )

    session.add(software)
    session.commit()
    return software

def update_software_row(
    software: Software,
    version: Version,
    exe_names: list[str],
    libs: list[str] | None,
    tags: list[str] | None,
    category: Category,
    docs: list[AnyUrl] | None,
    refs: list[AnyUrl] | None,
    linkexe: bool,
    arch: CPUArch,
) -> Software:
    if software.arch != arch:
        raise ValueError(f"Architecture mismatch in DB: {software.arch} != {arch} for {software.name}")
    software.latest = version
    software.provides = exe_names
    software.libs = libs
    software.tags = tags
    software.category = category
    software.docs = docs
    software.refs = refs
    if linkexe:
        software.version_in_path = version
    return software

def update_software_db(
    db_path: Path,
    name: str,
    version: Version,
    installed: datetime,
    method: InstallMethod,
    target: Path,
    exe_paths: list[Path],
    libs: list[str] | None,
    ver_msg: str | None,
    help_msg: str | None,
    linked_dir: bool,
    linked_exes: bool,
    bindir: Path,
    exe_names: list[str],
    tags: list[str],
    category: Category,
    docs: list[AnyUrl] | None,
    refs: list[AnyUrl] | None,
    pkgs: list[str] | None,
    arch: CPUArch,
    debug: bool,
) -> None:
    engine = start_engine(db_path=db_path, debug=debug)
    with Session(engine) as session:
        software: Software | None = session.exec(
            select(Software).where(Software.name == name).where(Software.arch == arch)
        ).one_or_none()
        if software is None:
            software = add_software_row(
                session=session,
                name=name,
                updated=installed,
                version=version,
                arch=arch,
                exe_names=exe_names,
                tags=tags,
                category=category,
                docs=docs,
                refs=refs,
                linked_exes=linked_exes,
                libs=libs,
            )
        current_version = InstalledVersion(
            version=version,
            installed=installed,
            software_id=software.id,
            method=method,
            prefix=target,
            exes=exe_paths,  # type: ignore
            pkgs=pkgs,
            version_message=ver_msg,
            help_message=help_msg,
            linked_dir=linked_dir,
            linked_exes=linked_exes,
            linked_to=bindir if linked_exes else None,
        )
        for old_version in software.versions:
            if current_version.prefix == old_version.prefix:
                session.delete(old_version)
        software.versions.append(current_version)
        if current_version.version == max([x.version for x in software.versions]):
            software = update_software_row(
                software=software,
                version=version,
                exe_names=exe_names,
                tags=tags,
                category=category,
                docs=docs,
                refs=refs,
                linkexe=linked_exes,
                libs=libs,
                arch=arch,
            )
            for previous_version in software.versions:
                if previous_version.id == current_version.id:
                    continue
                else:
                    if current_version.linked_dir:
                        previous_version.linked_dir = False
                    if current_version.linked_exes:
                        previous_version.linked_exes = False
        session.commit()

def list_software(db_path: Path, markdown: bool = False) -> None:
    engine = start_engine(db_path=db_path)

    headerkeys = [
        "name",
        "arch",
        "latest",
        "updated",
        "versions",
        "provides",
        "tags",
        "category",
    ]
    if markdown:
        headerkeys.extend(["docs", "refs"])

    table = get_table(headerkeys=headerkeys, title="Software List", md=markdown)

    with Session(engine) as session:
        software_list = session.exec(select(Software)).all()
        for software in software_list:
            data = software.model_dump(mode="json")
            versions = []
            for version in software.versions:
                if (cur_ver := version.version) == software.version_in_path:
                    if markdown:
                        versions.append(f"**{cur_ver}\\***")
                    else:
                        versions.append(f"[bold]{cur_ver}*[/bold]")
                else:
                    versions.append(version.version)
            data["versions"] = list_to_csv(versions)
            if len(data["provides"]) > 1:
                data["provides"] = [data["provides"][0], "..."]

            for key in ("provides", "tags"):
                data[key] = list_to_csv(data[key])

            for key in ("docs", "refs"):
                data[key] = list_to_csv(data[key], links=True)

            if markdown:
                data["updated"] = datetime.fromisoformat(data["updated"]).strftime("%Y-%m-%d")
            else:
                data["updated"] = datetime.fromisoformat(data["updated"]).strftime("%a %d %b %Y")
            try:
                table.add_row(*[data.get(x, "") for x in headerkeys])
            except AttributeError:
                table.value_matrix.append([data.get(x, "") for x in headerkeys])

        print_rich_table(table)

def list_to_csv(data: Iterable | None, links: bool = False) -> str | None:
    if data is None:
        return None
    if links:
        return ", ".join(f"[link]({x})" for x in data)
    else:
        return ", ".join(str(x) for x in data)

def get_table(headerkeys: list[str], title: str, md: bool = False) -> Union[Table, TableWriter]:
    if md:
        return get_table_writer(headerkeys, title)
    else:
        return get_rich_table(headerkeys, title)

def get_rich_table(headerkeys: list[str], title: str) -> Table:
    borders = box.HEAVY_HEAD
    if not sys.stdout.isatty():
        borders = None
    header: list[Column] = [Column(x, overflow="fold") for x in headerkeys]

    table = Table(*header, title=title, box=borders, expand=True)
    return table

def get_table_writer(headerkeys: list[str], title: str) -> TableWriter:
    table = TableWriter(table_name=title, headers=headerkeys, value_matrix=[])
    return table

def print_rich_table(table: Union[Table, TableWriter]) -> None:
    """
    Prints a rich.Table view.
    """
    console = Console()
    try:
        print(table.dumps())
    except AttributeError:
        console.print(table)

def show_info(name: str, db_path: Path) -> None:
    engine = start_engine(db_path=db_path)

    with Session(engine) as session:
        software = session.exec(select(Software).where(Software.name == name)).one_or_none()
        if software is None:
            rprint(f"There is no software named '{name}'.")
            software_names = session.exec(select(Software.name)).all()
            rprint(f"Choose one from this list: {software_names}")
            exit(1)
        pprint(software.model_dump(mode="json"))
        for installed_version in software.versions:
            pprint(installed_version.model_dump(mode="json"))