Skip to content

Commit

Permalink
Merge pull request #20 from TabulateJarl8/improved_typing
Browse files Browse the repository at this point in the history
Finish fully typing
  • Loading branch information
TabulateJarl8 authored Nov 15, 2024
2 parents e344dd5 + fa2e98b commit 5ef0ff9
Show file tree
Hide file tree
Showing 8 changed files with 703 additions and 358 deletions.
743 changes: 436 additions & 307 deletions poetry.lock

Large diffs are not rendered by default.

5 changes: 4 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[tool.poetry]
name = "vapor-steam"
version = "1.5.7"
version = "1.5.8"
description = "TUI program to check the ProtonDB compatibility of all the games of a Steam user."
authors = ["TabulateJarl8 <[email protected]>"]
license = "GPLv3"
Expand Down Expand Up @@ -46,6 +46,9 @@ pytest-asyncio = "^0.24.0"
[tool.pytest.ini_options]
asyncio_default_fixture_loop_scope = "function"

[tool.pyright]
reportUnusedCallResult = false

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
Expand Down
42 changes: 25 additions & 17 deletions vapor/api_interface.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,16 +13,21 @@
HTTP_UNAUTHORIZED,
RATING_DICT,
STEAM_USER_ID_LENGTH,
AntiCheatAPIResponse,
AntiCheatData,
AntiCheatStatus,
Game,
ProtonDBAPIResponse,
Response,
SteamAPINameResolutionResponse,
SteamAPIPlatformsResponse,
SteamAPIUserDataResponse,
SteamUserData,
)
from vapor.exceptions import InvalidIDError, PrivateAccountError, UnauthorizedError


async def async_get(url: str, **session_kwargs: Any) -> Response:
async def async_get(url: str, **session_kwargs: Any) -> Response: # pyright: ignore[reportAny]
"""Async get request for fetching web content.
Args:
Expand All @@ -32,7 +37,7 @@ async def async_get(url: str, **session_kwargs: Any) -> Response:
Returns:
Response: A Response object containing the body and status code.
"""
async with aiohttp.ClientSession(**session_kwargs) as session, session.get(
async with aiohttp.ClientSession(**session_kwargs) as session, session.get( # pyright: ignore[reportAny]
url,
) as response:
return Response(data=await response.text(), status=response.status)
Expand All @@ -53,16 +58,19 @@ async def check_game_is_native(app_id: str) -> bool:
if data.status != HTTP_SUCCESS:
return False

json_data = json.loads(data.data)
json_data: Dict[str, SteamAPIPlatformsResponse] = json.loads(data.data)

return _extract_game_is_native(json_data, app_id)


def _extract_game_is_native(data: Dict, app_id: str) -> bool:
def _extract_game_is_native(
data: Dict[str, SteamAPIPlatformsResponse],
app_id: str,
) -> bool:
"""Extract whether or not a game is Linux native from API data.
Args:
data (Dict): the data from the Steam API.
data (Dict[str, SteamAPIPlatformsResponse]): the data from the Steam API.
app_id (str): The App ID of the game
Returns:
Expand Down Expand Up @@ -98,7 +106,7 @@ async def get_anti_cheat_data() -> Optional[Cache]:
return None

try:
anti_cheat_data = json.loads(data.data)
anti_cheat_data: List[AntiCheatAPIResponse] = json.loads(data.data)
except json.JSONDecodeError:
return None

Expand All @@ -109,11 +117,11 @@ async def get_anti_cheat_data() -> Optional[Cache]:
return cache


def parse_anti_cheat_data(data: List[Dict]) -> List[AntiCheatData]:
def parse_anti_cheat_data(data: List[AntiCheatAPIResponse]) -> List[AntiCheatData]:
"""Parse and return data from AreWeAntiCheatYet.
Args:
data (List[Dict]): The data from AreWeAntiCheatYet
data (List[AntiCheatAPIResponse]): The data from AreWeAntiCheatYet
Returns:
List[AntiCheatData]: the anticheat statuses of each game in the given data
Expand Down Expand Up @@ -152,7 +160,7 @@ async def get_game_average_rating(app_id: str, cache: Cache) -> str:
if data.status != HTTP_SUCCESS:
return 'pending'

json_data = json.loads(data.data)
json_data: ProtonDBAPIResponse = json.loads(data.data)

return json_data.get('tier', 'pending')

Expand All @@ -178,7 +186,7 @@ async def resolve_vanity_name(api_key: str, name: str) -> str:
if data.status == HTTP_FORBIDDEN:
raise UnauthorizedError

user_data = json.loads(data.data)
user_data: SteamAPINameResolutionResponse = json.loads(data.data)
if 'response' not in user_data or user_data['response']['success'] != 1:
raise InvalidIDError

Expand Down Expand Up @@ -218,19 +226,19 @@ async def get_steam_user_data(api_key: str, user_id: str) -> SteamUserData:
if data.status == HTTP_UNAUTHORIZED:
raise UnauthorizedError

data = json.loads(data.data)
user_data: SteamAPIUserDataResponse = json.loads(data.data)

return await parse_steam_user_games(data, cache)
return await parse_steam_user_games(user_data, cache)


async def parse_steam_user_games(
data: Dict,
data: SteamAPIUserDataResponse,
cache: Cache,
) -> SteamUserData:
"""Parse user data from the Steam API and return information on their games.
Args:
data (Dict): user data from the Steam API
data (SteamAPIUserDataResponse): user data from the Steam API
cache (Cache): the loaded Cache file
Returns:
Expand All @@ -240,12 +248,12 @@ async def parse_steam_user_games(
PrivateAccountError: if `games` is not present in `data['response']`
(the user's account was found but is private)
"""
data = data['response']
game_data = data['response']

if 'games' not in data:
if 'games' not in game_data:
raise PrivateAccountError

games = data['games']
games = game_data['games']
game_ratings = [
Game(
name=game['name'],
Expand Down
6 changes: 4 additions & 2 deletions vapor/argument_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,12 @@ def parse_args() -> None:
),
)
parser.add_argument(
'--clear-cache', action='store_true', help="Clear all of vapor's cache",
'--clear-cache',
action='store_true',
help="Clear all of vapor's cache",
)

args = parser.parse_args()

if args.clear_cache:
if args.clear_cache: # pyright: ignore[reportAny]
cache_handler.CACHE_PATH.unlink(missing_ok=True)
31 changes: 21 additions & 10 deletions vapor/cache_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,20 @@

import json
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Tuple

from typing_extensions import Self
from typing_extensions import Self, override

from vapor.data_structures import CONFIG_DIR, AntiCheatData, AntiCheatStatus, Game
from vapor.data_structures import (
CONFIG_DIR,
AntiCheatData,
AntiCheatStatus,
CacheFile,
Game,
SerializedAnticheatData,
SerializedGameData,
)

CACHE_PATH = CONFIG_DIR / 'cache.json'
"""The path to the cache file."""
Expand All @@ -26,20 +35,21 @@ class Cache:

def __init__(self) -> None:
"""Construct a new Cache object."""
self.cache_path = CACHE_PATH
self.cache_path: Path = CACHE_PATH
self._games_data: Dict[str, Tuple[Game, str]] = {}
self._anti_cheat_data: Dict[str, AntiCheatData] = {}
self._anti_cheat_timestamp: str = ''

@override
def __repr__(self) -> str:
"""Return the string representation of the Cache object."""
return f'Cache({self.__dict__!r})'

def _serialize_game_data(self) -> dict:
def _serialize_game_data(self) -> Dict[str, SerializedGameData]:
"""Serialize the game data into a valid JSON dict.
Returns:
dict: Valid JSON dict.
Dict[str, SerializedGameData]: Valid JSON dict.
"""
return {
app_id: {
Expand All @@ -50,11 +60,11 @@ def _serialize_game_data(self) -> dict:
for app_id, game in self._games_data.items()
}

def _serialize_anti_cheat_data(self) -> dict:
def _serialize_anti_cheat_data(self) -> SerializedAnticheatData:
"""Serialize the anticheat data into a valid JSON dict.
Returns:
dict: Valid JSON dict.
SerializedAnticheatData: Valid JSON dict.
"""
return {
'data': {
Expand Down Expand Up @@ -118,7 +128,7 @@ def load_cache(self, prune: Optional[bool] = True) -> Self:
self.prune_cache()

try:
data = json.loads(self.cache_path.read_text())
data: CacheFile = json.loads(self.cache_path.read_text())
except Exception:
return self

Expand Down Expand Up @@ -197,7 +207,7 @@ def prune_cache(self) -> Self:
Self: self.
"""
try:
data = json.loads(self.cache_path.read_text())
data: CacheFile = json.loads(self.cache_path.read_text())
except Exception:
return self

Expand All @@ -217,7 +227,8 @@ def prune_cache(self) -> Self:
if 'anticheat_cache' in data:
try:
parsed_date = datetime.strptime(
data['anticheat_cache']['timestamp'], TIMESTAMP_FORMAT,
data['anticheat_cache']['timestamp'],
TIMESTAMP_FORMAT,
)
if (datetime.now() - parsed_date).days > CACHE_INVALIDATION_DAYS:
# cache is too old, delete game
Expand Down
3 changes: 2 additions & 1 deletion vapor/config_handler.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Vapor configuration file handling."""

from configparser import ConfigParser
from pathlib import Path
from typing import Optional

from typing_extensions import Self
Expand All @@ -20,7 +21,7 @@ class Config:

def __init__(self) -> None:
"""Construct a new Config object."""
self._config_path = CONFIG_PATH
self._config_path: Path = CONFIG_PATH
self._config_data: Optional[ConfigParser] = None

def set_value(self, key: str, value: str) -> Self:
Expand Down
Loading

0 comments on commit 5ef0ff9

Please sign in to comment.