Fetch pynwb dependies for hdmf and h5py

fetch_pynwb_requirements.py

"""
Fetches hdmf and h5py versions for each pynwb release.
Stores the versions in a DataJoint table for later use.

This was developed based on the misunderstanding that pynwb's requirements.txt
reflects the actual versions used in each release. Instead, this information
is stored in the pyproject.toml file, without the hard version pins.

Author: Chris Broz
Date: 2025-01-25
"""

import warnings

import datajoint as dj
import h5py
import hdmf
import requests

from spyglass.utils import logger

warnings.filterwarnings("ignore", module="hdmf")
warnings.filterwarnings("ignore", module="pynwb")

schema = dj.schema("cbroz_temp")


@schema
class PyNWBDeps(dj.Lookup):
    definition = """
    version: varchar(32)
    ---
    hdmf: varchar(32)
    h5py: varchar(32)
    """

    OWNER = "NeurodataWithoutBorders"
    REPO = "pynwb"
    RELEASES_API_URL = f"https://api.github.com/repos/{OWNER}/{REPO}/releases"
    RAW_REQUIREMENTS_URL = (
        f"https://raw.githubusercontent.com/{OWNER}/{REPO}/"
        + "{{version}}/requirements.txt"
    )

    def fetch_dep(self, version, attempt=1):
        """Fetch requirements.txt for a given version."""
        if tbl := self & {"version": version}:
            return tbl.fetch1("hdmf", "h5py")
        if attempt > 2:
            raise RuntimeError(
                f"Failed to fetch requirements for version: {version}"
            )
        _ = self.get_contents()
        return self.fetch_dep(version, attempt + 1)

    def get_releases(self):
        """Fetch releases from GitHub API."""
        resp = requests.get(self.RELEASES_API_URL)

        if resp.status_code != 200:
            raise RuntimeError(
                f"Failed to fetch releases. Status code: {resp.status_code}"
            )
            return []

        ret = []
        existing_releases = self.fetch("version")
        for info in resp.json():
            ver = info["tag_name"]
            if ver not in existing_releases and ver != "latest":
                ret.append(info["tag_name"])
        return ret

    def fetch_requirements(self, version):
        """Fetch requirements.txt for a given version."""
        url = self.RAW_REQUIREMENTS_URL.format(version=version)
        response = requests.get(url)
        if response.status_code != 200:
            raise RuntimeError(
                f"Requirements.txt not found for version: {version}"
            )
        return response.text

    def parse_requirements(self, version, requirements_text):
        """Parse requirements.txt for hdmf and h5py pins."""

        def parse_line(line):
            line = line.strip()
            if "==" in line:
                dep, version = line.split("==")
            else:
                dep, version = line[:4], line[4:]
            return {dep: version}

        deps = {"version": version, "hdmf": None, "h5py": None}
        for line in requirements_text.splitlines():
            if line.startswith("hdmf") or line.startswith("h5py"):
                deps.update(parse_line(line))
        return deps

    def get_contents(self):
        dependencies = []
        for version in self.get_releases():
            logger.info(f"Fetching pynwb dependencies for version: {version}")
            requirements_text = self.fetch_requirements(version)
            dependencies.append(
                self.parse_requirements(version, requirements_text)
            )
        self.insert(dependencies, skip_duplicates=True)
        return dependencies

    @property
    def valid_ver_restr(self) -> str:
        query = self & {"hdmf": hdmf.__version__, "h5py": h5py.__version__}
        vers = query.fetch("version")
        if len(vers) == 1:
            return f"version='{vers[0]}'"
        multi = '"' + '", "'.join(vers) + '"'
        return f"version IN ({multi})"


if __name__ == "__main__":
    PyNWBDeps().get_contents()
    print(PyNWBDeps().valid_ver_restr)
    print(PyNWBDeps().fetch_dep("0.11.0"))