Skip to content

Commit

Permalink
Init
Browse files Browse the repository at this point in the history
  • Loading branch information
seriaati committed Apr 28, 2024
1 parent efe2022 commit f8f2d33
Show file tree
Hide file tree
Showing 28 changed files with 2,000 additions and 0 deletions.
53 changes: 53 additions & 0 deletions .github/workflows/main.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
name: Seria's CI

on:
push:
branches:
- main
schedule:
- cron: '0 0 * * 1' # every Monday at 00:00
workflow_dispatch:

jobs:
create-release:
runs-on: ubuntu-latest
if: github.event_name == 'push' || github.event_name == 'workflow_dispatch'
steps:
- name: Checkout code
uses: actions/checkout@v4
with:
fetch-depth: 0

- name: Create release
uses: seriaati/create-release@main

update-deps:
runs-on: ubuntu-latest
if: github.event_name == 'schedule' || github.event_name == 'workflow_dispatch'
steps:
- name: Checkout code
uses: actions/checkout@v4

- name: Update dependencies & pre-commits
uses: seriaati/update-deps@main

pytest:
runs-on: ubuntu-latest
if: github.event_name == 'push' || github.event_name == 'workflow_dispatch'
steps:
- name: Checkout code
uses: actions/checkout@v4

- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: 3.11

- name: Install and configure Poetry
uses: snok/install-poetry@v1

- name: Install dependencies
run: poetry install --with dev

- name: Run pytest
run: poetry run pytest
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -150,3 +150,4 @@ cython_debug/
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
test.py
4 changes: 4 additions & 0 deletions .vscode/settings.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
{
"python.analysis.typeCheckingMode": "standard",
"python.analysis.importFormat": "relative"
}
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
@@ -1,2 +1,3 @@
# hakushin-py

Async API wrapper for hakush.in written in Python.
5 changes: 5 additions & 0 deletions hakushin/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
from . import models, utils
from .client import *
from .constants import *
from .enums import *
from .errors import *
219 changes: 219 additions & 0 deletions hakushin/client.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,219 @@
import logging
from typing import Any, Final, Literal, Self, overload

from aiohttp_client_cache.backends.sqlite import SQLiteBackend
from aiohttp_client_cache.session import CachedSession

from hakushin.enums import Game, Language
from hakushin.errors import HakushinError, NotFoundError

from .constants import HSR_LANG_MAP
from .models import gi, hsr

__all__ = ("HakushinAPI",)

LOGGER_ = logging.getLogger(__name__)


class HakushinAPI:
"""Client to interact with the Hakushin API."""

BASE_URL: Final[str] = "https://api.hakush.in"

def __init__(
self,
lang: Language = Language.EN,
*,
cache_path: str = "./.cache/hakushin/aiohttp-cache.db",
cache_ttl: int = 3600,
headers: dict[str, Any] | None = None,
debug: bool = False,
) -> None:
"""Initializes the Hakushin API client.
Args:
lang (Language): The language to fetch data in.
cache_path (str): The path to the cache database.
cache_ttl (int): The time-to-live for cache entries.
headers (dict): The headers to pass with the request.
debug (bool): Whether to enable debug logging.
"""
self.lang = lang
self.cache_ttl = cache_ttl

self._session: CachedSession | None = None
self._cache = SQLiteBackend(cache_path, expire_after=cache_ttl)
self._headers = headers or {"User-Agent": "hakuashin-py"}
self._debug = debug
if self._debug:
logging.basicConfig(level=logging.DEBUG)

async def __aenter__(self) -> Self:
await self.start()
return self

async def __aexit__(self, exc_type, exc, tb) -> None: # noqa: ANN001
await self.close()

async def _request(
self, endpoint: str, game: Game, use_cache: bool, *, static: bool = False
) -> dict[str, Any]:
"""A helper function to make requests to the API.
Args:
endpoint (str): The endpoint to request.
game (Game): The game to fetch data for.
use_cache (bool): Whether to use the cache.
static (bool): Whether the endpoint is static data (not language specific), defaults to False.
Returns:
dict: The response data.
"""
if self._session is None:
msg = "Call `start` before making requests."
raise RuntimeError(msg)

lang = HSR_LANG_MAP[self.lang] if game is Game.HSR else self.lang.value

url = (
f"{self.BASE_URL}/{game.value}/{endpoint}.json"
if static
else f"{self.BASE_URL}/{game.value}/data/{lang}/{endpoint}.json"
)
LOGGER_.debug("Requesting %s...", url)

if not use_cache:
async with self._session.disabled(), self._session.get(url) as resp:
if resp.status != 200:
self._handle_error(resp.status, url)
data = await resp.json()
else:
async with self._session.get(url) as resp:
if resp.status != 200:
self._handle_error(resp.status, url)
data = await resp.json()

return data

def _handle_error(self, code: int, url: str) -> None:
"""Handles API errors based on the status code.
Args:
code (int): The status code.
url (str): The URL that caused the error.
"""
match code:
case 404:
raise NotFoundError(url)
case _:
raise HakushinError(code, "An error occurred while fetching data.", url)

async def start(self) -> None:
"""Starts the client session."""
self._session = CachedSession(headers=self._headers, cache=self._cache)

async def close(self) -> None:
"""Closes the client session."""
if self._session is not None:
await self._session.close()

@overload
async def fetch_new(self, game: Literal[Game.GI], *, use_cache: bool = True) -> gi.New: ...
@overload
async def fetch_new(self, game: Literal[Game.HSR], *, use_cache: bool = True) -> hsr.New: ...
async def fetch_new(self, game: Game, *, use_cache: bool = True) -> gi.New | hsr.New:
"""Fetches the IDs of new stuff in the game.
Args:
game (Game): The game to fetch data for.
use_cache (bool): Whether to use the cache.
Returns:
New: The new stuff object.
"""
endpoint = "new"
data = await self._request(endpoint, game, use_cache, static=True)
return gi.New(**data) if game is Game.GI else hsr.New(**data)

@overload
async def fetch_character(
self, character_id: int, game: Literal[Game.GI], *, use_cache: bool = True
) -> gi.GICharacter: ...
@overload
async def fetch_character(
self, character_id: int, game: Literal[Game.HSR], *, use_cache: bool = True
) -> hsr.HSRCharacter: ...
async def fetch_character(
self, character_id: int, game: Game, *, use_cache: bool = True
) -> gi.GICharacter | hsr.HSRCharacter:
"""Fetches a character.
Args:
character_id (int): The character ID.
game (Game): The game to fetch data for.
use_cache (bool): Whether to use the cache.
Returns:
Character: The character object.
"""
endpoint = f"character/{character_id}"
data = await self._request(endpoint, game, use_cache)
return gi.GICharacter(**data) if game is Game.GI else hsr.HSRCharacter(**data)

async def fetch_weapon(self, weapon_id: int, *, use_cache: bool = True) -> gi.Weapon:
"""Fetches a Genshin Impact weapon.
Args:
weapon_id (int): The weapon ID.
use_cache (bool): Whether to use the cache.
Returns:
Weapon: The weapon object.
"""
endpoint = f"weapon/{weapon_id}"
data = await self._request(endpoint, Game.GI, use_cache)
return gi.Weapon(**data)

async def fetch_light_cone(
self, light_cone_id: int, *, use_cache: bool = True
) -> hsr.LightCone:
"""Fetches a HSR light cone.
Args:
light_cone_id (int): The light cone ID.
use_cache (bool): Whether to use the cache.
Returns:
LightCone: The light cone object.
"""
endpoint = f"lightcone/{light_cone_id}"
data = await self._request(endpoint, Game.HSR, use_cache)
return hsr.LightCone(**data)

async def fetch_artifact_set(self, set_id: int, *, use_cache: bool = True) -> gi.ArtifactSet:
"""Fetches an artifact set.
Args:
set_id (int): The artifact set ID.
use_cache (bool): Whether to use the cache.
Returns:
ArtifactSet: The artifact set object.
"""
endpoint = f"artifact/{set_id}"
data = await self._request(endpoint, Game.GI, use_cache)
return gi.ArtifactSet(**data)

async def fetch_relic_set(self, set_id: int, *, use_cache: bool = True) -> hsr.RelicSet:
"""Fetches a relic set.
Args:
set_id (int): The relic set ID.
use_cache (bool): Whether to use the cache.
Returns:
RelicSet: The relic set object.
"""
endpoint = f"relicset/{set_id}"
data = await self._request(endpoint, Game.HSR, use_cache)
return hsr.RelicSet(**data)
28 changes: 28 additions & 0 deletions hakushin/constants.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
from typing import Final

from .enums import Language

__all__ = ("GI_CHARA_RARITY_MAP", "HSR_CHARA_RARITY_MAP", "HSR_LIGHT_CONE_RARITY_MAP")

GI_CHARA_RARITY_MAP: Final[dict[str, int]] = {
"QUALITY_PURPLE": 4,
"QUALITY_ORANGE": 5,
}

HSR_CHARA_RARITY_MAP: Final[dict[str, int]] = {
"CombatPowerAvatarRarityType4": 4,
"CombatPowerAvatarRarityType5": 5,
}

HSR_LIGHT_CONE_RARITY_MAP: Final[dict[str, int]] = {
"CombatPowerLightconeRarity4": 4,
"CombatPowerLightconeRarity5": 5,
}

HSR_LANG_MAP: Final[dict[Language, str]] = {
Language.EN: "en",
Language.JA: "jp",
Language.KO: "kr",
Language.ZH: "cn",
}
"""Map to convert language enum for GI to for HSR."""
19 changes: 19 additions & 0 deletions hakushin/enums.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
from enum import StrEnum

__all__ = ("Game", "Language")


class Game(StrEnum):
"""Games supported by the Hakushin API."""

GI = "gi" # Genshin Impact
HSR = "hsr" # Honkai Star Rail


class Language(StrEnum):
"""Lanauges supported by the Hakushin API."""

EN = "en" # English
ZH = "zh" # Simplified Chinese
KO = "ko" # Korean
JA = "ja" # Japanese
24 changes: 24 additions & 0 deletions hakushin/errors.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
__all__ = ("HakushinError", "NotFoundError")


class HakushinError(Exception):
"""Base class for exceptions in the Hakushin API."""

def __init__(self, status: int, message: str, url: str) -> None:
super().__init__(message)
self.status = status
self.message = message
self.url = url

def __str__(self) -> str:
return f"{self.status}: {self.message} ({self.url})"

def __repr__(self) -> str:
return f"<{self.__class__.__name__} status={self.status} message={self.message!r} url={self.url!r}>"


class NotFoundError(HakushinError):
"""Raised when the requested resource is not found."""

def __init__(self, url: str) -> None:
super().__init__(404, "The requested resource was not found.", url)
1 change: 1 addition & 0 deletions hakushin/models/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
from . import gi, hsr
Loading

0 comments on commit f8f2d33

Please sign in to comment.