mirror of
https://github.com/ReVanced/revanced-api.git
synced 2026-01-17 16:33:57 +00:00
feat: API Fixes and Adjustments (#23)
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
Alexandre Teles (afterSt0rm)
committed by
GitHub
GitHub
parent
1273f9224b
commit
b18097e030
@@ -1,14 +1,15 @@
|
||||
import asyncio
|
||||
from json import loads
|
||||
import os
|
||||
from operator import eq
|
||||
from typing import Any, Optional
|
||||
from typing import Optional
|
||||
|
||||
import ujson
|
||||
from aiohttp import ClientResponse
|
||||
from sanic import SanicException
|
||||
from toolz import filter, map
|
||||
from toolz import filter, map, partial
|
||||
from toolz.dicttoolz import get_in, keyfilter
|
||||
from toolz.itertoolz import mapcat
|
||||
from toolz.itertoolz import mapcat, pluck
|
||||
|
||||
from api.backends.backend import Backend, Repository
|
||||
from api.backends.entities import *
|
||||
@@ -95,19 +96,27 @@ class Github(Backend):
|
||||
async def __assemble_contributor(
|
||||
contributor: dict, team_view: bool = False
|
||||
) -> Contributor:
|
||||
if team_view:
|
||||
filter_contributor = keyfilter(
|
||||
lambda key: key in {"login", "avatar_url", "html_url"},
|
||||
contributor,
|
||||
)
|
||||
return Contributor(**filter_contributor)
|
||||
match team_view:
|
||||
case True:
|
||||
keys = {"login", "avatar_url", "html_url", "bio"}
|
||||
case _:
|
||||
keys = {"login", "avatar_url", "html_url", "contributions"}
|
||||
|
||||
filter_contributor = keyfilter(
|
||||
lambda key: key in {"login", "avatar_url", "html_url", "contributions"},
|
||||
lambda key: key in keys,
|
||||
contributor,
|
||||
)
|
||||
|
||||
return Contributor(**filter_contributor)
|
||||
|
||||
@staticmethod
|
||||
async def __validate_request(_response: ClientResponse) -> None:
|
||||
if _response.status != 200:
|
||||
raise SanicException(
|
||||
context=await _response.json(loads=ujson.loads),
|
||||
status_code=_response.status,
|
||||
)
|
||||
|
||||
async def list_releases(
|
||||
self, repository: GithubRepository, per_page: int = 30, page: int = 1
|
||||
) -> list[Release]:
|
||||
@@ -126,11 +135,7 @@ class Github(Backend):
|
||||
response: ClientResponse = await http_get(
|
||||
headers=self.headers, url=list_releases_endpoint
|
||||
)
|
||||
if response.status != 200:
|
||||
raise SanicException(
|
||||
context=await response.json(loads=ujson.loads),
|
||||
status_code=response.status,
|
||||
)
|
||||
await self.__validate_request(response)
|
||||
releases: list[Release] = await asyncio.gather(
|
||||
*map(
|
||||
lambda release: self.__assemble_release(release),
|
||||
@@ -156,11 +161,7 @@ class Github(Backend):
|
||||
response: ClientResponse = await http_get(
|
||||
headers=self.headers, url=release_by_tag_endpoint
|
||||
)
|
||||
if response.status != 200:
|
||||
raise SanicException(
|
||||
context=await response.json(loads=ujson.loads),
|
||||
status_code=response.status,
|
||||
)
|
||||
await self.__validate_request(response)
|
||||
return await self.__assemble_release(await response.json(loads=ujson.loads))
|
||||
|
||||
async def get_latest_release(
|
||||
@@ -179,11 +180,7 @@ class Github(Backend):
|
||||
response: ClientResponse = await http_get(
|
||||
headers=self.headers, url=latest_release_endpoint
|
||||
)
|
||||
if response.status != 200:
|
||||
raise SanicException(
|
||||
context=await response.json(loads=ujson.loads),
|
||||
status_code=response.status,
|
||||
)
|
||||
await self.__validate_request(response)
|
||||
return await self.__assemble_release(await response.json(loads=ujson.loads))
|
||||
|
||||
async def get_latest_pre_release(
|
||||
@@ -202,11 +199,7 @@ class Github(Backend):
|
||||
response: ClientResponse = await http_get(
|
||||
headers=self.headers, url=list_releases_endpoint
|
||||
)
|
||||
if response.status != 200:
|
||||
raise SanicException(
|
||||
context=await response.json(loads=ujson.loads),
|
||||
status_code=response.status,
|
||||
)
|
||||
await self.__validate_request(response)
|
||||
latest_pre_release = next(
|
||||
filter(
|
||||
lambda release: release["prerelease"],
|
||||
@@ -229,11 +222,7 @@ class Github(Backend):
|
||||
response: ClientResponse = await http_get(
|
||||
headers=self.headers, url=contributors_endpoint
|
||||
)
|
||||
if response.status != 200:
|
||||
raise SanicException(
|
||||
context=await response.json(loads=ujson.loads),
|
||||
status_code=response.status,
|
||||
)
|
||||
await self.__validate_request(response)
|
||||
contributors: list[Contributor] = await asyncio.gather(
|
||||
*map(self.__assemble_contributor, await response.json(loads=ujson.loads))
|
||||
)
|
||||
@@ -241,38 +230,43 @@ class Github(Backend):
|
||||
return contributors
|
||||
|
||||
async def get_patches(
|
||||
self, repository: GithubRepository, tag_name: str
|
||||
self, repository: GithubRepository, tag_name: str = "latest", dev: bool = False
|
||||
) -> list[dict]:
|
||||
"""Get a dictionary of patch URLs for a given repository.
|
||||
|
||||
Args:
|
||||
repository (GithubRepository): The repository for which to retrieve patches.
|
||||
tag_name: The name of the release tag.
|
||||
tag_name (str): The name of the release tag.
|
||||
dev (bool): If we should get the latest pre-release instead.
|
||||
|
||||
Returns:
|
||||
list[dict]: A JSON object containing the patches.
|
||||
"""
|
||||
|
||||
async def __fetch_download_url(release: Release) -> str:
|
||||
asset = get_in(["assets"], release)
|
||||
async def __fetch_download_url(_release: Release) -> str:
|
||||
asset = get_in(["assets"], _release)
|
||||
patch_asset = next(
|
||||
filter(lambda x: eq(get_in(["name"], x), "patches.json"), asset), None
|
||||
)
|
||||
return get_in(["browser_download_url"], patch_asset)
|
||||
|
||||
response: ClientResponse = await http_get(
|
||||
headers=self.headers,
|
||||
url=await __fetch_download_url(
|
||||
await self.get_release_by_tag_name(
|
||||
match tag_name:
|
||||
case "latest":
|
||||
match dev:
|
||||
case True:
|
||||
release = await self.get_latest_pre_release(repository)
|
||||
case _:
|
||||
release = await self.get_latest_release(repository)
|
||||
case _:
|
||||
release = await self.get_release_by_tag_name(
|
||||
repository=repository, tag_name=tag_name
|
||||
)
|
||||
),
|
||||
|
||||
response: ClientResponse = await http_get(
|
||||
headers=self.headers,
|
||||
url=await __fetch_download_url(_release=release),
|
||||
)
|
||||
if response.status != 200:
|
||||
raise SanicException(
|
||||
context=await response.json(loads=ujson.loads),
|
||||
status_code=response.status,
|
||||
)
|
||||
await self.__validate_request(response)
|
||||
return ujson.loads(await response.read())
|
||||
|
||||
async def get_team_members(self, repository: GithubRepository) -> list[Contributor]:
|
||||
@@ -285,18 +279,30 @@ class Github(Backend):
|
||||
list[Contributor]: A list of members in the owner organization.
|
||||
"""
|
||||
team_members_endpoint: str = f"{self.base_url}/orgs/{repository.owner}/members"
|
||||
user_info_endpoint: str = f"{self.base_url}/users/"
|
||||
response: ClientResponse = await http_get(
|
||||
headers=self.headers, url=team_members_endpoint
|
||||
)
|
||||
if response.status != 200:
|
||||
raise SanicException(
|
||||
context=await response.json(loads=ujson.loads),
|
||||
status_code=response.status,
|
||||
await self.__validate_request(response)
|
||||
logins: list[str] = list(pluck("login", await response.json()))
|
||||
_http_get = partial(http_get, headers=self.headers)
|
||||
user_data_response: list[dict] = await asyncio.gather(
|
||||
*map(
|
||||
lambda login: _http_get(url=f"{user_info_endpoint}{login}"),
|
||||
logins,
|
||||
)
|
||||
)
|
||||
user_data = await asyncio.gather(
|
||||
*map(
|
||||
lambda _response: _response.json(loads=ujson.loads),
|
||||
user_data_response,
|
||||
)
|
||||
)
|
||||
print(await response.json(loads=ujson.loads))
|
||||
team_members: list[Contributor] = await asyncio.gather(
|
||||
*map(
|
||||
lambda member: self.__assemble_contributor(member, team_view=True),
|
||||
await response.json(loads=ujson.loads),
|
||||
list(user_data),
|
||||
)
|
||||
)
|
||||
|
||||
@@ -315,17 +321,18 @@ class Github(Backend):
|
||||
list[dict[str, str]]: A JSON object containing the releases.
|
||||
"""
|
||||
|
||||
def transform(data, repository):
|
||||
def transform(data: dict, repository: GithubRepository):
|
||||
"""Transforms a dictionary from the input list into a list of dictionaries with the desired structure.
|
||||
|
||||
Args:
|
||||
data(dict): A dictionary from the input list.
|
||||
repository(GithubRepository): The repository for which to retrieve releases.
|
||||
|
||||
Returns:
|
||||
_[list]: A list of dictionaries with the desired structure.
|
||||
"""
|
||||
|
||||
def process_asset(asset):
|
||||
def process_asset(asset: dict) -> dict:
|
||||
"""Transforms an asset dictionary into a new dictionary with the desired structure.
|
||||
|
||||
Args:
|
||||
@@ -353,3 +360,39 @@ class Github(Backend):
|
||||
)
|
||||
|
||||
return list(mapcat(lambda pair: transform(*pair), zip(results, repositories)))
|
||||
|
||||
async def compat_get_contributors(
|
||||
self, repositories: list[GithubRepository]
|
||||
) -> list:
|
||||
"""Get the contributors for a set of repositories (v1 compat).
|
||||
|
||||
Args:
|
||||
repositories (set[GithubRepository]): The repositories for which to retrieve contributors.
|
||||
|
||||
Returns:
|
||||
list[dict[str, str]]: A JSON object containing the contributors.
|
||||
"""
|
||||
|
||||
def transform(data: dict, repository: GithubRepository) -> list:
|
||||
"""Transforms a dictionary from the input list into a list of dictionaries with the desired structure.
|
||||
|
||||
Args:
|
||||
data(dict): A dictionary from the input list.
|
||||
repository(GithubRepository): The repository for which to retrieve contributors.
|
||||
|
||||
Returns:
|
||||
_[list]: A list of dictionaries with the desired structure.
|
||||
"""
|
||||
return {
|
||||
"name": f"{repository.owner}/{repository.name}",
|
||||
"contributors": data,
|
||||
}
|
||||
|
||||
results = await asyncio.gather(
|
||||
*map(
|
||||
lambda repository: self.get_contributors(repository),
|
||||
repositories,
|
||||
)
|
||||
)
|
||||
|
||||
return list(map(lambda pair: transform(*pair), zip(results, repositories)))
|
||||
|
||||
Reference in New Issue
Block a user