-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #11 from letsbuilda/gh10
Retrieve package metadata from the JSON API
- Loading branch information
Showing
6 changed files
with
215 additions
and
60 deletions.
There are no files selected for viewing
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
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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 = "[email protected]" }, | ||
|
@@ -11,6 +11,7 @@ requires-python = ">=3.10" | |
dependencies = [ | ||
"aiohttp", | ||
"xmltodict", | ||
"pendulum", | ||
] | ||
|
||
[project.urls] | ||
|
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
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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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()) |
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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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, | ||
) |
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
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -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": "[email protected]", | ||
"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": "[email protected]", | ||
"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 |