ReVanced/revanced-releases-api
1
0
Fork 0
mirror of https://github.com/revanced/revanced-releases-api.git synced 2025-04-29 22:14:28 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #39 from mirgb/dev

Browse Source 
refactor: __get_release() and get_latest_releases() functions
This commit is contained in:
Alexandre Teles 2022-11-26 22:08:37 -03:00 committed by GitHub
commit 88b767f9ae
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 90 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

173
app/controllers/Releases.py
View File

@ -7,49 +7,53 @@ from app.utils.HTTPXClient import HTTPXClient
class Releases: class Releases:
"""Implements the methods required to get the latest releases and patches from revanced repositories.""" """Implements the methods required to get the latest releases and patches from revanced repositories."""
uvloop.install() uvloop.install()
httpx_client = HTTPXClient.create() httpx_client = HTTPXClient.create()
async def __get_release(self, repository: str) -> list: async def __get_release(self, repository: str) -> list:
# Get assets from latest release in a given repository. """Get assets from latest release in a given repository.
#
# Args: Args:
# repository (str): Github's standard username/repository notation repository (str): Github's standard username/repository notation
#
# Returns: Returns:
# dict: dictionary of filename and download url dict: dictionary of filename and download url
"""
assets: list = [] assets: list = []
response = await self.httpx_client.get(f"https://api.github.com/repos/{repository}/releases/latest") response = await self.httpx_client.get(f"https://api.github.com/repos/{repository}/releases/latest")
if response.status_code == 200: if response.status_code == 200:
release_assets: dict = response.json()['assets'] release_assets: dict = response.json()['assets']
release_version: str = response.json()['tag_name'] release_version: str = response.json()['tag_name']
release_tarball: str = response.json()['tarball_url'] release_tarball: str = response.json()['tarball_url']
release_timestamp: str = response.json()['published_at'] release_timestamp: str = response.json()['published_at']
if release_assets: async def get_asset_data(asset: dict) -> dict:
for asset in release_assets: return {'repository': repository,
assets.append({ 'repository': repository, 'version': release_version,
'version': release_version, 'timestamp': asset['updated_at'],
'timestamp': asset['updated_at'], 'name': asset['name'],
'name': asset['name'], 'size': asset['size'],
'size': asset['size'], 'browser_download_url': asset['browser_download_url'],
'browser_download_url': asset['browser_download_url'], 'content_type': asset['content_type']
'content_type': asset['content_type'] }
})
else: no_release_assets_data: dict = {'repository': repository,
assets.append({ 'repository': repository, 'version': release_version,
'version': release_version, 'timestamp': release_timestamp,
'timestamp': release_timestamp, 'name': f"{repository.split('/')[1]}-{release_version}.tar.gz",
'name': f"{repository.split('/')[1]}-{release_version}.tar.gz", 'browser_download_url': release_tarball,
'browser_download_url': release_tarball, 'content_type': 'application/gzip'
'content_type': 'application/gzip' }
})
assets = [map(get_asset_data, release_assets)
if release_assets else no_release_assets_data]
return assets return assets
async def get_latest_releases(self, repositories: list) -> dict: async def get_latest_releases(self, repositories: list) -> dict:
@ -61,58 +65,60 @@ class Releases:
Returns: Returns:
dict: A dictionary containing assets from each repository dict: A dictionary containing assets from each repository
""" """
releases: dict[str, list] = {} releases: dict[str, list] = {}
releases['tools'] = [] releases['tools'] = []
results: list = await asyncio.gather(*[self.__get_release(repository) for repository in repositories]) results: list = await asyncio.gather(*[self.__get_release(repository) for repository in repositories])
for result in results: releases['tools'] = [asset for result in results for asset in result]
for asset in result:
releases['tools'].append(asset)
return releases return releases
async def __get_patches_json(self) -> dict: async def __get_patches_json(self) -> dict:
# Get revanced-patches repository's README.md. """Get revanced-patches repository's README.md.
#
# Returns: Returns:
# dict: JSON content dict: JSON content
# """
response = await self.httpx_client.get(f"https://api.github.com/repos/revanced/revanced-patches/contents/patches.json") response = await self.httpx_client.get(f"https://api.github.com/repos/revanced/revanced-patches/contents/patches.json")
content = orjson.loads(b64decode(response.json()['content']).decode('utf-8')) content = orjson.loads(
b64decode(response.json()['content']).decode('utf-8'))
return content return content
async def get_patches_json(self) -> dict: async def get_patches_json(self) -> dict:
"""Get patches.json from revanced-patches repository. """Get patches.json from revanced-patches repository.
Returns: Returns:
dict: Patches available for a given app dict: Patches available for a given app
""" """
patches: dict = await self.__get_patches_json() patches: dict = await self.__get_patches_json()
return patches return patches
async def __get_contributors(self, repository: str) -> list: async def __get_contributors(self, repository: str) -> list:
# Get contributors from a given repository. """Get contributors from a given repository.
#
# Args: Args:
# repository (str): Github's standard username/repository notation repository (str): Github's standard username/repository notation
#
# Returns: Returns:
# list: a list of dictionaries containing the repository's contributors list: a list of dictionaries containing the repository's contributors
"""
keep: set = {'login', 'avatar_url', 'html_url'} keep: set = {'login', 'avatar_url', 'html_url'}
response = await self.httpx_client.get(f"https://api.github.com/repos/{repository}/contributors") response = await self.httpx_client.get(f"https://api.github.com/repos/{repository}/contributors")
# Looping over each contributor, filtering each contributor so that
# keyfilter() returns a dictionary with only the key-value pairs that are in the "keep" set.
contributors: list = [keyfilter(lambda k: k in keep, contributor) for contributor in response.json()] contributors: list = [keyfilter(lambda k: k in keep, contributor) for contributor in response.json()]
return contributors return contributors
async def get_contributors(self, repositories: list) -> dict: async def get_contributors(self, repositories: list) -> dict:
"""Runs get_contributors() asynchronously for each repository. """Runs get_contributors() asynchronously for each repository.
@ -122,22 +128,23 @@ class Releases:
Returns: Returns:
dict: A dictionary containing the contributors from each repository dict: A dictionary containing the contributors from each repository
""" """
contributors: dict[str, list] contributors: dict[str, list]
contributors = {} contributors = {}
contributors['repositories'] = [] contributors['repositories'] = []
revanced_repositories = [repository for repository in repositories if 'revanced' in repository] revanced_repositories = [
repository for repository in repositories if 'revanced' in repository]
results: list[dict] = await asyncio.gather(*[self.__get_contributors(repository) for repository in revanced_repositories]) results: list[dict] = await asyncio.gather(*[self.__get_contributors(repository) for repository in revanced_repositories])
for key, value in zip(revanced_repositories, results): for key, value in zip(revanced_repositories, results):
data = { 'name': key, 'contributors': value } data = {'name': key, 'contributors': value}
contributors['repositories'].append(data) contributors['repositories'].append(data)
return contributors return contributors
async def get_commits(self, org: str, repository: str, path: str) -> dict: async def get_commits(self, org: str, repository: str, path: str) -> dict:
"""Get commit history from a given repository. """Get commit history from a given repository.
@ -154,28 +161,28 @@ class Releases:
Returns: Returns:
dict: a dictionary containing the repository's latest commits dict: a dictionary containing the repository's latest commits
""" """
payload: dict = {} payload: dict = {}
payload["repository"] = f"{org}/{repository}" payload["repository"] = f"{org}/{repository}"
payload["path"] = path payload["path"] = path
payload["commits"] = [] payload["commits"] = []
if org == 'revanced' or org == 'vancedapp': if org == 'revanced' or org == 'vancedapp':
_releases = await self.httpx_client.get( _releases = await self.httpx_client.get(
f"https://api.github.com/repos/{org}/{repository}/releases?per_page=2" f"https://api.github.com/repos/{org}/{repository}/releases?per_page=2"
) )
releases = _releases.json() releases = _releases.json()
since = releases[1]['created_at'] since = releases[1]['created_at']
until = releases[0]['created_at'] until = releases[0]['created_at']
_response = await self.httpx_client.get( _response = await self.httpx_client.get(
f"https://api.github.com/repos/{org}/{repository}/commits?path={path}&since={since}&until={until}" f"https://api.github.com/repos/{org}/{repository}/commits?path={path}&since={since}&until={until}"
) )
response = _response.json() response = _response.json()
for commit in response: for commit in response:
data: dict[str, str] = {} data: dict[str, str] = {}
data["sha"] = commit["sha"] data["sha"] = commit["sha"]
@ -183,7 +190,7 @@ class Releases:
data["message"] = commit["commit"]["message"] data["message"] = commit["commit"]["message"]
data["html_url"] = commit["html_url"] data["html_url"] = commit["html_url"]
payload['commits'].append(data) payload['commits'].append(data)
return payload return payload
else: else:
raise Exception("Invalid organization.") raise Exception("Invalid organization.")