diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst index 7639710..96959a0 100644 --- a/docs/source/changelog.rst +++ b/docs/source/changelog.rst @@ -4,6 +4,9 @@ Changelog ========= +- :release:`3.0.0 <21th April 2023>` +- :feature:`10` Retrieve package metadata from the JSON API + - :release:`2.0.0 <14th April 2023>` - :feature:`4` Use single method for all RSS feeds diff --git a/pyproject.toml b/pyproject.toml index 487b65c..3047fef 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "letsbuilda-pypi" -version = "2.0.0" +version = "3.0.0" description = "A wrapper for PyPI's API and RSS feed" authors = [ { name = "Bradley Reynolds", email = "bradley.reynolds@darbia.dev" }, @@ -11,6 +11,7 @@ requires-python = ">=3.10" dependencies = [ "aiohttp", "xmltodict", + "pendulum", ] [project.urls] diff --git a/src/letsbuilda/pypi/__init__.py b/src/letsbuilda/pypi/__init__.py index a8072f2..3110c77 100644 --- a/src/letsbuilda/pypi/__init__.py +++ b/src/letsbuilda/pypi/__init__.py @@ -2,63 +2,11 @@ A wrapper for PyPI's API and RSS feed """ -from dataclasses import dataclass -from datetime import datetime -from typing import Final - -import xmltodict -from aiohttp import ClientSession +from .client import PyPIServices +from .models import PackageMetadata, RSSPackageMetadata __all__: list[str] = [ "PyPIServices", "PackageMetadata", + "RSSPackageMetadata", ] - - -def _parse_publication_date(publication_date: str) -> datetime: - return datetime.strptime(publication_date, "%a, %d %b %Y %H:%M:%S %Z") - - -@dataclass(frozen=True, slots=True) -class PackageMetadata: - """Package metadata""" - - title: str - package_link: str - guid: str - description: str | None - author: str | None - publication_date: datetime - - @classmethod - def build_from(cls, data: dict[str, str]) -> "PackageMetadata": - """Build an instance from raw data""" - publication_date: str | None = data.get("pubDate") - if publication_date is not None: - publication_date: datetime = _parse_publication_date(publication_date) - - return cls( - title=data.get("title").split()[0], - package_link=data.get("link"), - guid=data.get("guid"), - description=data.get("description"), - author=data.get("author"), - publication_date=publication_date, - ) - - -class PyPIServices: - """A class for interacting with PyPI""" - - NEWEST_PACKAGES_FEED_URL: Final[str] = "https://pypi.org/rss/packages.xml" - PACKAGE_UPDATES_FEED_URL: Final[str] = "https://pypi.org/rss/updates.xml" - - def __init__(self, http_session: ClientSession) -> None: - self.http_session = http_session - - async def get_rss_feed(self, feed_url: str) -> list[PackageMetadata]: - """Get the new packages RSS feed""" - async with self.http_session.get(feed_url) as response: - response_text = await response.text() - rss_data = xmltodict.parse(response_text)["rss"]["channel"]["item"] - return [PackageMetadata.build_from(package_data) for package_data in rss_data] diff --git a/src/letsbuilda/pypi/client.py b/src/letsbuilda/pypi/client.py new file mode 100644 index 0000000..e8821a2 --- /dev/null +++ b/src/letsbuilda/pypi/client.py @@ -0,0 +1,35 @@ +"""Service wrapper""" + +from typing import Final + +import xmltodict +from aiohttp import ClientSession + +from .models import PackageMetadata, RSSPackageMetadata + + +class PyPIServices: + """A class for interacting with PyPI""" + + NEWEST_PACKAGES_FEED_URL: Final[str] = "https://pypi.org/rss/packages.xml" + PACKAGE_UPDATES_FEED_URL: Final[str] = "https://pypi.org/rss/updates.xml" + + def __init__(self, http_session: ClientSession) -> None: + self.http_session = http_session + + async def get_rss_feed(self, feed_url: str) -> list[RSSPackageMetadata]: + """Get the new packages RSS feed""" + async with self.http_session.get(feed_url) as response: + response_text = await response.text() + rss_data = xmltodict.parse(response_text)["rss"]["channel"]["item"] + return [RSSPackageMetadata.build_from(package_data) for package_data in rss_data] + + async def get_package_metadata(self, package_name: str) -> PackageMetadata: + """Get the new packages RSS feed""" + async with self.http_session.get(f"https://pypi.org/pypi/{package_name}/json") as response: + return PackageMetadata.from_dict(await response.json()) + + async def get_package_metadata_for_release(self, package_name: str, package_version: str) -> PackageMetadata: + """Get the new packages RSS feed""" + async with self.http_session.get(f"https://pypi.org/pypi/{package_name}/{package_version}/json") as response: + return PackageMetadata.from_dict(await response.json()) diff --git a/src/letsbuilda/pypi/models.py b/src/letsbuilda/pypi/models.py new file mode 100644 index 0000000..8af92bb --- /dev/null +++ b/src/letsbuilda/pypi/models.py @@ -0,0 +1,168 @@ +"""Models""" + +from dataclasses import dataclass +from datetime import datetime + +import pendulum +from pendulum import DateTime + + +def _parse_publication_date(publication_date: str) -> datetime: + return datetime.strptime(publication_date, "%a, %d %b %Y %H:%M:%S %Z") + + +@dataclass(frozen=True, slots=True) +class Vulnerability: + """Security vulnerability""" + + id: str + aliases: list[str] + link: str + source: str + withdrawn: DateTime | None + summary: str + details: str + fixed_in: list[str] + + @classmethod + def from_dict(cls, data: dict) -> "Vulnerability": + if data["withdrawn"] is not None: + data["withdrawn"]: DateTime = pendulum.parse(data["withdrawn"]) + + return cls(**data) + + +@dataclass(frozen=True, slots=True) +class Downloads: + """Release download counts""" + + last_day: int + last_month: int + last_week: int + + @classmethod + def from_dict(cls, data: dict) -> "Downloads": + return cls(**data) + + +@dataclass(frozen=True, slots=True) +class Digests: + """URL file digests""" + + blake2_b_256: str + md5: str + sha256: str + + @classmethod + def from_dict(cls, data: dict) -> "Digests": + return cls(**data) + + +@dataclass(frozen=True, slots=True) +class URL: + """Package release URL""" + + comment_text: str + digests: Digests + downloads: int + filename: str + has_sig: bool + md5_digest: str + packagetype: str + python_version: str + requires_python: str | None + size: int + upload_time: DateTime + upload_time_iso_8601: DateTime + url: str + yanked: bool + yanked_reason: None + + @classmethod + def from_dict(cls, data: dict) -> "URL": + data["upload_time"]: DateTime = pendulum.parse(data["upload_time"]) + data["upload_time_iso_8601"]: DateTime = pendulum.parse(data["upload_time_iso_8601"]) + return cls(**data) + + +@dataclass(frozen=True, slots=True) +class Info: + """Package metadata internal info block""" + + author: str + author_email: str + bugtrack_url: None + classifiers: list[str] + description: str + description_content_type: str + docs_url: None + download_url: str + downloads: Downloads + home_page: str + keywords: str + license: str + maintainer: str + maintainer_email: str + name: str + package_url: str + platform: str | None + project_url: str + project_urls: dict[str, str] + release_url: str + requires_dist: list[str] + requires_python: str + summary: str + version: str + yanked: bool + yanked_reason: str | None + + @classmethod + def from_dict(cls, data: dict) -> "Info": + return cls(**data) + + +@dataclass(frozen=True, slots=True) +class PackageMetadata: + """Package metadata""" + + info: Info + last_serial: int + urls: list[URL] + vulnerabilities: list["Vulnerability"] + + @classmethod + def from_dict(cls, data: dict) -> "PackageMetadata": + return cls( + info=Info.from_dict(data["info"]), + last_serial=data["last_serial"], + urls=[URL.from_dict(url_data) for url_data in data["urls"]], + vulnerabilities=[Vulnerability.from_dict(vuln_data) for vuln_data in data["vulnerabilities"]], + ) + + +@dataclass(frozen=True, slots=True) +class RSSPackageMetadata: + """RSS Package metadata""" + + title: str + package_link: str + guid: str + description: str | None + author: str | None + publication_date: datetime | None + + @classmethod + def build_from(cls, data: dict[str, str]) -> "RSSPackageMetadata": + """Build an instance from raw data""" + publication_date: str | None = data.get("pubDate") + if publication_date is not None: + publication_date: datetime = _parse_publication_date(publication_date) + + return cls( + title=data.get("title").split()[0], + package_link=data.get("link"), + guid=data.get("guid"), + description=data.get("description"), + author=data.get("author"), + publication_date=publication_date, + ) diff --git a/tests/test_new_packages_feed.py b/tests/test_new_packages_feed.py index 0512896..181a059 100644 --- a/tests/test_new_packages_feed.py +++ b/tests/test_new_packages_feed.py @@ -2,7 +2,7 @@ from datetime import datetime -from letsbuilda.pypi import PackageMetadata +from letsbuilda.pypi import RSSPackageMetadata def test_parse_publication_date() -> None: @@ -15,7 +15,7 @@ def test_parse_publication_date() -> None: "author": "test-author@example.com", "pubDate": "Wed, 29 Mar 2023 21:30:05 GMT", } - parsed_data = PackageMetadata.build_from(data) + parsed_data = RSSPackageMetadata.build_from(data) assert parsed_data.publication_date == datetime(2023, 3, 29, 21, 30, 5) @@ -28,7 +28,7 @@ def test_author_missing() -> None: "description": "a test package", "pubDate": "Wed, 29 Mar 2023 21:30:05 GMT", } - parsed_data = PackageMetadata.build_from(data) + parsed_data = RSSPackageMetadata.build_from(data) assert parsed_data.author is None @@ -41,5 +41,5 @@ def test_description_missing() -> None: "author": "test-author@example.com", "pubDate": "Wed, 29 Mar 2023 21:30:05 GMT", } - parsed_data = PackageMetadata.build_from(data) + parsed_data = RSSPackageMetadata.build_from(data) assert parsed_data.description is None