ReVanced/revanced-api
1
0
Fork 0
mirror of https://github.com/ReVanced/revanced-api.git synced 2026-01-18 08:53:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: API rewrite (#2)

Browse Source 
* feat: sanic framework settings

* feat: initial implementation

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* refactor: backend changes

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix: docstrings out of place

* feat: more gh endpoints

* ci: fix pre-commit issues

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* feat: app info

* ci: merge CI and fix triggers

* chore: bump deps

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix: typing issues

* chore: deps

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* refactor: clean up returns

* ci: spread jobs correctly

* ci: move to quodana

* ci: fix issues with python modules

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* chore: pycharm config

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* refactor: improve code quality

* feat: better README

* ci: add quodana baseline config

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* ci: fix quodana config

* ci: more qodana stuff

* ci: revert qodana changes

* ci: python interpreter detection is broken

* feat: tests

* ci: testing

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* ci: fix workflow names

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* chore: add deps

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* test: more tests

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* feat: /tools compat

* feat: donations endpoint

* feat: teams endpoint

* fix: lock pydantic version

* chore: deps

* ci: docker builds

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* ci: remove coverage action and others

* ci: pre-commit fixes

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
Alexandre Teles (afterSt0rm)
2023-07-12 04:48:36 -03:00
committed by GitHub
parent cb52684edb
commit 45ef33741c
45 changed files with 16359 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
api/backends/__init__.py Normal file
View File

64
api/backends/apkdl.py Normal file
View File

@@ -0,0 +1,64 @@
from base64 import b64encode
from aiohttp import ClientResponse
from bs4 import BeautifulSoup
from sanic import SanicException
from toolz.functoolz import compose
from api.backends.backend import AppInfoProvider
from api.backends.entities import AppInfo
from api.utils.http_utils import http_get
name: str = "apkdl"
base_url: str = "https://apk-dl.com"
class ApkDl(AppInfoProvider):
def __init__(self):
super().__init__(name, base_url)
async def get_app_info(self, package_name: str) -> AppInfo:
"""Fetches information about an Android app from the ApkDl website.
Args:
package_name (str): The package name of the app to fetch.
Returns:
AppInfo: An AppInfo object containing the name, category, and logo of the app.
Raises:
SanicException: If the HTTP request fails or the app data is incomplete or not found.
"""
app_url: str = f"{base_url}/{package_name}"
response: ClientResponse = await http_get(headers={}, url=app_url)
if response.status != 200:
raise SanicException(
f"ApkDl: {response.status}", status_code=response.status
)
page = BeautifulSoup(await response.read(), "lxml")
find_div_text = compose(
lambda d: d.find_next_sibling("div"),
lambda d: page.find("div", text=d),
)
fetch_logo_url = compose(
lambda div: div.img["src"],
lambda _: page.find("div", {"class": "logo"}),
)
logo_response: ClientResponse = await http_get(
headers={}, url=fetch_logo_url(None)
)
logo: str = (
f"data:image/png;base64,{b64encode(await logo_response.content.read()).decode('utf-8')}"
if logo_response.status == 200
else ""
)
app_data = dict(
name=find_div_text("App Name").text,
category=find_div_text("Category").text,
logo=logo,
)
if not all(app_data.values()):
raise SanicException(
"ApkDl: App data incomplete or not found", status_code=500
)
return AppInfo(**app_data)

91
api/backends/backend.py Normal file
View File

@@ -0,0 +1,91 @@
from abc import abstractmethod
from typing import Any, Protocol
from api.backends.entities import *
class Backend(Protocol):
"""Interface for a generic backend.
Attributes:
name (str): Name of the backend.
base_url (str): Base URL of the backend.
Methods:
list_releases: Retrieve a list of releases.
get_release_by_tag_name: Retrieve a release by its tag name.
get_latest_release: Retrieve the latest release.
get_latest_pre_release: Retrieve the latest pre-release.
get_release_notes: Retrieve the release notes of a specific release.
get_contributors: Retrieve the list of contributors.
get_patches: Retrieve the patches of a specific release.
"""
name: str
base_url: str
def __init__(self, name: str, base_url: str):
self.name = name
self.base_url = base_url
@abstractmethod
async def list_releases(self, *args: Any, **kwargs: Any) -> list[Release]:
raise NotImplementedError
@abstractmethod
async def get_release_by_tag_name(self, *args: Any, **kwargs: Any) -> Release:
raise NotImplementedError
@abstractmethod
async def get_latest_release(self, *args: Any, **kwargs: Any) -> Release:
raise NotImplementedError
@abstractmethod
async def get_latest_pre_release(self, *args: Any, **kwargs: Any) -> Release:
raise NotImplementedError
@abstractmethod
async def get_contributors(self, *args: Any, **kwargs: Any) -> list[Contributor]:
raise NotImplementedError
@abstractmethod
async def get_patches(self, *args: Any, **kwargs: Any) -> list[dict]:
raise NotImplementedError
@abstractmethod
async def get_team_members(self, *args: Any, **kwargs: Any) -> list[Contributor]:
raise NotImplementedError
class Repository:
"""A repository that communicates with a specific backend.
Attributes:
backend (Backend): The backend instance used to communicate with the repository.
"""
def __init__(self, backend: Backend):
self.backend = backend
class AppInfoProvider(Protocol):
"""Interface for a generic app info provider.
Attributes:
name (str): Name of the app info provider.
base_url (str): Base URL of the app info provider.
Methods:
get_app_info: Retrieve information about an app.
"""
name: str
base_url: str
def __init__(self, name: str, base_url: str):
self.name = name
self.base_url = base_url
@abstractmethod
async def get_app_info(self, *args: Any, **kwargs: Any) -> AppInfo:
raise NotImplementedError

126
api/backends/entities.py Normal file
View File

@@ -0,0 +1,126 @@
from typing import Optional
from dataclasses import dataclass
@dataclass
class Metadata(dict):
"""
Represents the metadata of a release.
Attributes:
- tag_name (str): The name of the release tag.
- name (str): The name of the release.
- body (str): The body of the release.
- draft (bool): Whether the release is a draft.
- prerelease (bool): Whether the release is a prerelease.
- created_at (str): The creation date of the release.
- published_at (str): The publication date of the release.
"""
def __init__(
self,
tag_name: str,
name: str,
draft: bool,
prerelease: bool,
created_at: str,
published_at: str,
body: str,
repository: Optional[str] = None,
):
dict.__init__(
self,
tag_name=tag_name,
name=name,
draft=draft,
prerelease=prerelease,
created_at=created_at,
published_at=published_at,
body=body,
repository=repository,
)
@dataclass
class Asset(dict):
"""
Represents an asset in a release.
Attributes:
- name (str): The name of the asset.
- content_type (str): The MIME type of the asset content.
- download_url (str): The URL to download the asset.
"""
def __init__(self, name: str, content_type: str, browser_download_url: str):
dict.__init__(
self,
name=name,
content_type=content_type,
browser_download_url=browser_download_url,
)
@dataclass
class Release(dict):
"""
Represents a release.
Attributes:
- metadata (Metadata): The metadata of the release.
- assets (list[Asset]): The assets of the release.
"""
def __init__(self, metadata: Metadata, assets: list[Asset]):
dict.__init__(self, metadata=metadata, assets=assets)
@dataclass
class Contributor(dict):
"""
Represents a contributor to a repository.
Attributes:
- login (str): The GitHub username of the contributor.
- avatar_url (str): The URL to the contributor's avatar image.
- html_url (str): The URL to the contributor's GitHub profile.
- contributions (Optional[int]): The number of contributions the contributor has made to the repository.
"""
def __init__(
self,
login: str,
avatar_url: str,
html_url: str,
contributions: Optional[int] = None,
):
if contributions:
dict.__init__(
self,
login=login,
avatar_url=avatar_url,
html_url=html_url,
contributions=contributions,
)
else:
dict.__init__(self, login=login, avatar_url=avatar_url, html_url=html_url)
@dataclass
class AppInfo(dict):
"""
Represents the information of an app.
Attributes:
- name (str): The name of the app.
- category (str): The app category.
- logo (str): The base64 enconded app logo.
"""
def __init__(self, name: str, category: str, logo: str):
dict.__init__(
self,
name=name,
category=category,
logo=logo,
)

355
api/backends/github.py Normal file
View File

@@ -0,0 +1,355 @@
import asyncio
import os
from operator import eq
from typing import Any, Optional
import ujson
from aiohttp import ClientResponse
from sanic import SanicException
from toolz import filter, map
from toolz.dicttoolz import get_in, keyfilter
from toolz.itertoolz import mapcat
from api.backends.backend import Backend, Repository
from api.backends.entities import *
from api.backends.entities import Contributor
from api.utils.http_utils import http_get
repo_name: str = "github"
base_url: str = "https://api.github.com"
class GithubRepository(Repository):
"""
A repository class that represents a Github repository.
Args:
owner (str): The username of the owner of the Github repository.
name (str): The name of the Github repository.
"""
def __init__(self, owner: str, name: str):
"""
Initializes a new instance of the GithubRepository class.
Args:
owner (str): The username of the owner of the Github repository.
name (str): The name of the Github repository.
"""
super().__init__(Github())
self.owner = owner
self.name = name
class Github(Backend):
"""
A backend class that interacts with the Github API.
Attributes:
name (str): The name of the Github backend.
base_url (str): The base URL of the Github API.
token (str): The Github access token used for authentication.
headers (dict[str, str]): The HTTP headers to be sent with each request to the Github API.
"""
def __init__(self):
"""
Initializes a new instance of the Github class.
"""
super().__init__(repo_name, base_url)
self.token: Optional[str] = os.getenv("GITHUB_TOKEN")
self.headers: dict[str, str] = {
"Authorization": f"Bearer {self.token}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
}
self.repositories_rest_endpoint: str = f"{base_url}/repos"
@staticmethod
async def __assemble_release(release: dict) -> Release:
async def __assemble_asset(asset: dict) -> Asset:
asset_data: dict = keyfilter(
lambda key: key in {"name", "content_type", "browser_download_url"},
asset,
)
return Asset(**asset_data)
filter_metadata = keyfilter(
lambda key: key
in {
"tag_name",
"name",
"draft",
"prerelease",
"created_at",
"published_at",
"body",
},
release,
)
metadata = Metadata(**filter_metadata)
assets = await asyncio.gather(*map(__assemble_asset, release["assets"]))
return Release(metadata=metadata, assets=assets)
@staticmethod
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)
filter_contributor = keyfilter(
lambda key: key in {"login", "avatar_url", "html_url", "contributions"},
contributor,
)
return Contributor(**filter_contributor)
async def list_releases(
self, repository: GithubRepository, per_page: int = 30, page: int = 1
) -> list[Release]:
"""
Returns a list of Release objects for a given Github repository.
Args:
repository (GithubRepository): The Github repository for which to retrieve the releases.
per_page (int): The number of releases to return per page.
page (int): The page number of the releases to return.
Returns:
list[Release]: A list of Release objects.
"""
list_releases_endpoint: str = f"{self.repositories_rest_endpoint}/{repository.owner}/{repository.name}/releases?per_page={per_page}&page={page}"
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,
)
releases: list[Release] = await asyncio.gather(
*map(
lambda release: self.__assemble_release(release),
await response.json(loads=ujson.loads),
)
)
return releases
async def get_release_by_tag_name(
self, repository: GithubRepository, tag_name: str
) -> Release:
"""
Retrieves a specific release for a given Github repository by its tag name.
Args:
repository (GithubRepository): The Github repository for which to retrieve the release.
tag_name (str): The tag name of the release to retrieve.
Returns:
Release: The Release object representing the retrieved release.
"""
release_by_tag_endpoint: str = f"{self.repositories_rest_endpoint}/{repository.owner}/{repository.name}/releases/tags/{tag_name}"
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,
)
return await self.__assemble_release(await response.json(loads=ujson.loads))
async def get_latest_release(
self,
repository: GithubRepository,
) -> Release:
"""Get the latest release for a given repository.
Args:
repository (GithubRepository): The Github repository for which to retrieve the release.
Returns:
Release: The latest release for the given repository.
"""
latest_release_endpoint: str = f"{self.repositories_rest_endpoint}/{repository.owner}/{repository.name}/releases/latest"
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,
)
return await self.__assemble_release(await response.json(loads=ujson.loads))
async def get_latest_pre_release(
self,
repository: GithubRepository,
) -> Release:
"""Get the latest pre-release for a given repository.
Args:
repository (GithubRepository): The Github repository for which to retrieve the release.
Returns:
Release: The latest pre-release for the given repository.
"""
list_releases_endpoint: str = f"{self.repositories_rest_endpoint}/{repository.owner}/{repository.name}/releases?per_page=10&page=1"
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,
)
latest_pre_release = next(
filter(
lambda release: release["prerelease"],
await response.json(loads=ujson.loads),
)
)
return await self.__assemble_release(latest_pre_release)
async def get_contributors(self, repository: GithubRepository) -> list[Contributor]:
"""Get a list of contributors for a given repository.
Args:
repository (GithubRepository): The repository for which to retrieve contributors.
Returns:
list[Contributor]: A list of contributors for the given repository.
"""
contributors_endpoint: str = f"{self.repositories_rest_endpoint}/{repository.owner}/{repository.name}/contributors"
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,
)
contributors: list[Contributor] = await asyncio.gather(
*map(self.__assemble_contributor, await response.json(loads=ujson.loads))
)
return contributors
async def get_patches(
self, repository: GithubRepository, tag_name: str
) -> 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.
Returns:
list[dict]: A JSON object containing the patches.
"""
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(
repository=repository, tag_name=tag_name
)
),
)
if response.status != 200:
raise SanicException(
context=await response.json(loads=ujson.loads),
status_code=response.status,
)
return ujson.loads(await response.read())
async def get_team_members(self, repository: GithubRepository) -> list[Contributor]:
"""Get the list of team members from the owner organization of a given repository.
Args:
repository (GithubRepository): The repository for which to retrieve team members in the owner organization.
Returns:
list[Contributor]: A list of members in the owner organization.
"""
team_members_endpoint: str = f"{self.base_url}/orgs/{repository.owner}/members"
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,
)
team_members: list[Contributor] = await asyncio.gather(
*map(
lambda member: self.__assemble_contributor(member, team_view=True),
await response.json(loads=ujson.loads),
)
)
return team_members
async def compat_get_tools(
self, repositories: list[GithubRepository], dev: bool
) -> list:
"""Get the latest releases for a set of repositories (v1 compat).
Args:
repositories (set[GithubRepository]): The repositories for which to retrieve releases.
dev: If we should get the latest pre-release instead.
Returns:
list[dict[str, str]]: A JSON object containing the releases.
"""
def transform(data, repository):
"""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.
Returns:
_[list]: A list of dictionaries with the desired structure.
"""
def process_asset(asset):
"""Transforms an asset dictionary into a new dictionary with the desired structure.
Args:
asset(dict): An asset dictionary.
Returns:
_[dict]: A new dictionary with the desired structure.
"""
return {
"repository": f"{repository.owner}/{repository.name}",
"version": data["metadata"]["tag_name"],
"timestamp": data["metadata"]["published_at"],
"name": asset["name"],
"browser_download_url": asset["browser_download_url"],
"content_type": asset["content_type"],
}
return map(process_asset, data["assets"])
results = await asyncio.gather(
*map(
lambda release: self.get_latest_release(release),
repositories,
)
)
return list(mapcat(lambda pair: transform(*pair), zip(results, repositories)))