diff --git a/spore_api/__init__.py b/spore_api/__init__.py index 22714f6..ab83966 100644 --- a/spore_api/__init__.py +++ b/spore_api/__init__.py @@ -12,17 +12,17 @@ from spore_api.errors import ( SporeApiStatusError, ) -from spore_api.builders import ( - build_achievements, - build_asset_comments, - build_assets, - build_buddies, - build_creature, - build_full_asset, - build_sporecast_assets, - build_sporecasts, - build_stats, - build_user, +from .parsers import ( + parse_stats, + parse_creature, + parse_user, + parse_assets, + parse_sporecasts, + parse_sporecast_assets, + parse_achievements, + parse_full_asset, + parse_asset_comments, + parse_buddies, ) from spore_api.models import ( Achievement, diff --git a/spore_api/client.py b/spore_api/client.py index 187c813..7af71eb 100644 --- a/spore_api/client.py +++ b/spore_api/client.py @@ -2,31 +2,27 @@ from types import TracebackType from typing import ( TYPE_CHECKING, - Any, - Callable, Optional, Type, Union, - Dict ) import aiohttp -import xmltodict from .errors import SporeApiStatusError from .constants import BASE_URL from .enums import AssetType, ViewType -from .builders import ( - build_stats, - build_creature, - build_user, - build_assets, - build_sporecasts, - build_sporecast_assets, - build_achievements, - build_full_asset, - build_asset_comments, - build_buddies +from .parsers import ( + parse_stats, + parse_creature, + parse_user, + parse_assets, + parse_sporecasts, + parse_sporecast_assets, + parse_achievements, + parse_full_asset, + parse_asset_comments, + parse_buddies, ) if TYPE_CHECKING: @@ -40,23 +36,18 @@ Buddies, SporecastAssets, FullAsset, - Sporecasts + Sporecasts, ) class SporeClient(): - _decoder: Callable[[str], Dict[str, Any]] - _base_url: str - def __init__(self) -> None: self._session = None - self._init() - - def _init(self): - self._base_url = BASE_URL - self._decoder = xmltodict.parse - async def create(self, session: Optional[aiohttp.ClientSession] = None) -> None: + async def create( + self, + session: Optional[aiohttp.ClientSession] = None, + ) -> None: self._session = ( aiohttp.ClientSession() if session is None else @@ -65,120 +56,128 @@ async def create(self, session: Optional[aiohttp.ClientSession] = None) -> None: async def __aenter__( self, - session: Optional[aiohttp.ClientSession] = None + session: Optional[aiohttp.ClientSession] = None, ) -> "SporeClient": await self.create(session) return self async def get_stats(self) -> "Stats": - return build_stats( - await self.get_data_from_url(f"{self._base_url}/rest/stats") + url = f"{BASE_URL}/rest/stats" + + return parse_stats( + await self.get_response_text(url) ) async def get_creature( self, - asset_id: Union[int, str] + asset_id: Union[int, str], ) -> "Creature": - return build_creature( - await self.get_data_from_url(f"{self._base_url}/rest/creature/{asset_id}") + url = f"{BASE_URL}/rest/creature/{asset_id}" + + return parse_creature( + await self.get_response_text(url) ) async def get_user_info( self, - username: str + username: str, ) -> "User": - return build_user( - await self.get_data_from_url(f"{self._base_url}/rest/user/{username}") + url = f"{BASE_URL}/rest/user/{username}" + + return parse_user( + await self.get_response_text(url) ) async def get_user_assets( self, username: str, start_index: Union[int, str], - length: Union[int, str] + length: Union[int, str], ) -> "Assets": - return build_assets( - await self.get_data_from_url( - f"{self._base_url}/rest/assets/user/{username}/{start_index}/{length}" - ) + url = f"{BASE_URL}/rest/assets/user/{username}/{start_index}/{length}" + + return parse_assets( + await self.get_response_text(url) ) async def get_user_sporecasts( self, - username: str + username: str, ) -> "Sporecasts": - return build_sporecasts( - await self.get_data_from_url(f"{self._base_url}/rest/sporecasts/{username}") + url = f"{BASE_URL}/rest/sporecasts/{username}" + + return parse_sporecasts( + await self.get_response_text(url) ) async def get_sporecast_assets( self, sporecast_id: Union[int, str], start_index: Union[int, str], - length: Union[int, str] + length: Union[int, str], ) -> "SporecastAssets": - return build_sporecast_assets( - await self.get_data_from_url( - f"{self._base_url}/rest/assets/sporecast/{sporecast_id}/{start_index}/{length}" - ) + url = f"{BASE_URL}/rest/assets/sporecast/{sporecast_id}/{start_index}/{length}" + + return parse_sporecast_assets( + await self.get_response_text(url) ) async def get_user_achievements( self, username: str, start_index: Union[int, str], - length: Union[int, str] + length: Union[int, str], ) -> "Achievements": - return build_achievements( - await self.get_data_from_url( - f"{self._base_url}/rest/achievements/{username}/{start_index}/{length}" - ) + url = f"{BASE_URL}/rest/achievements/{username}/{start_index}/{length}" + + return parse_achievements( + await self.get_response_text(url) ) async def get_asset_info( self, - asset_id: Union[int, str] + asset_id: Union[int, str], ) -> "FullAsset": - return build_full_asset( - await self.get_data_from_url( - f"{self._base_url}/rest/asset/{asset_id}" - ) + url = f"{BASE_URL}/rest/asset/{asset_id}" + + return parse_full_asset( + await self.get_response_text(url) ) async def get_asset_comments( self, asset_id: Union[int, str], start_index: Union[int, str], - length: Union[int, str] + length: Union[int, str], ) -> "AssetComments": - return build_asset_comments( - await self.get_data_from_url( - f"{self._base_url}/rest/comments/{asset_id}/{start_index}/{length}" - ) + url = f"{BASE_URL}/rest/comments/{asset_id}/{start_index}/{length}" + + return parse_asset_comments( + await self.get_response_text(url) ) async def get_user_buddies( self, username: str, start_index: Union[int, str], - length: Union[int, str] + length: Union[int, str], ) -> "Buddies": - return build_buddies( - await self.get_data_from_url( - f"{self._base_url}/rest/users/buddies/{username}/{start_index}/{length}" - ) + url = f"{BASE_URL}/rest/users/buddies/{username}/{start_index}/{length}" + + return parse_buddies( + await self.get_response_text(url) ) async def get_user_subscribers( self, username: str, start_index: Union[int, str], - length: Union[int, str] + length: Union[int, str], ) -> "Buddies": - return build_buddies( - await self.get_data_from_url( - f"{self._base_url}/rest/users/subscribers/{username}/{start_index}/{length}" - ) + url = f"{BASE_URL}/rest/users/subscribers/{username}/{start_index}/{length}" + + return parse_buddies( + await self.get_response_text(url) ) async def search_assets( @@ -186,22 +185,24 @@ async def search_assets( view_type: ViewType, start_index: Union[int, str], length: Union[int, str], - asset_type: Optional[AssetType] = None + asset_type: Optional[AssetType] = None, ) -> "Assets": - if asset_type is None: - url = f"{self._base_url}/rest/assets/search/{view_type}/{start_index}/{length}" - else: - url = f"{self._base_url}/rest/assets/search/{view_type}/{start_index}/{length}/{asset_type}" + url = ( + f"{BASE_URL}/rest/assets/search/{view_type}/{start_index}/{length}" + if asset_type is None else + f"{BASE_URL}/rest/assets/search/{view_type}/{start_index}/{length}/{asset_type}" + ) - return build_assets( - await self.get_data_from_url(url) + return parse_assets( + await self.get_response_text(url) ) - async def get_data_from_url(self, url: str): - text = await self.get_response_text(url) - return await self.parse_response_text(text) + async def get_response_text(self, url: str) -> str: + text = await self._get_response_text(url) + self.check_status_spore_api(text) + return text - async def parse_response_text(self, text: str): + def check_status_spore_api(self, text: str) -> None: api_status_parse = re.search(r"(\d+)", text) if api_status_parse is not None: api_status = int(api_status_parse.group(1)) @@ -209,10 +210,7 @@ async def parse_response_text(self, text: str): if api_status != 1: raise SporeApiStatusError(api_status) - raw_data = self._decoder(text) - return raw_data - - async def get_response_text(self, url: str) -> str: + async def _get_response_text(self, url: str) -> str: if self._session is None: raise ValueError("The session does not exist") diff --git a/spore_api/builders.py b/spore_api/parsers.py similarity index 85% rename from spore_api/builders.py rename to spore_api/parsers.py index 7545932..1c1f748 100644 --- a/spore_api/builders.py +++ b/spore_api/parsers.py @@ -3,6 +3,8 @@ from typing import Any, Dict, List from pathlib import Path +import xmltodict + from spore_api.constants import BASE_URL from .utils import datatime_from_string, find_dict_by_value @@ -27,11 +29,13 @@ ) -def build_stats(raw_data: Dict[str, Any]) -> Stats: +def parse_stats(text: str) -> Stats: """ Build stats http://www.spore.com/rest/stats """ + raw_data: Dict[str, Any] = xmltodict.parse(text) # type: ignore + data: Dict[str, str] = raw_data["stats"] return Stats( @@ -42,13 +46,15 @@ def build_stats(raw_data: Dict[str, Any]) -> Stats: ) -def build_creature(raw_data: Dict[str, Any]) -> Creature: +def parse_creature(text: str) -> Creature: """ [Pages] Creature stats: http://www.spore.com/rest/creature/ """ + raw_data: Dict[str, Any] = xmltodict.parse(text) # type: ignore + data: Dict[str, str] = raw_data["creature"] return Creature( @@ -79,13 +85,15 @@ def build_creature(raw_data: Dict[str, Any]) -> Creature: ) -def build_user(raw_data: Dict[str, Any]) -> User: +def parse_user(text: str) -> User: """ [Pages] Profile Info: http://www.spore.com/rest/user/ """ + raw_data: Dict[str, Any] = xmltodict.parse(text) # type: ignore + data: Dict[str, str] = raw_data["user"] return User( @@ -97,7 +105,7 @@ def build_user(raw_data: Dict[str, Any]) -> User: ) -def build_assets(raw_data: Dict[str, Any]) -> Assets: +def parse_assets(text: str) -> Assets: """ [Pages] @@ -107,6 +115,11 @@ def build_assets(raw_data: Dict[str, Any]) -> Assets: Special Searches: http://www.spore.com/rest/assets/search/// """ + raw_data: Dict[str, Any] = xmltodict.parse( + text, + force_list=("asset",), + ) # type: ignore + data: Dict[str, Any] = raw_data["assets"] raw_assets: List[Dict[str, str]] = data["asset"] @@ -143,13 +156,18 @@ def build_assets(raw_data: Dict[str, Any]) -> Assets: ) -def build_sporecasts(raw_data: Dict[str, Any]) -> Sporecasts: +def parse_sporecasts(text: str) -> Sporecasts: """ [Pages] Sporecasts for user: http://www.spore.com/rest/sporecasts/ """ + raw_data: Dict[str, Any] = xmltodict.parse( + text, + force_list=("sporecast",) + ) # type: ignore + data: Dict[str, Any] = raw_data["sporecasts"] raw_sporecast: List[Dict[str, str]] = data["sporecast"] @@ -172,13 +190,18 @@ def build_sporecasts(raw_data: Dict[str, Any]) -> Sporecasts: ) -def build_sporecast_assets(raw_data: Dict[str, Any]) -> SporecastAssets: +def parse_sporecast_assets(text: str) -> SporecastAssets: """ [Pages] Assets for sporecast: http://www.spore.com/rest/assets/sporecast/// """ + raw_data: Dict[str, Any] = xmltodict.parse( + text, + force_list=("asset",), + ) # type: ignore + data: Dict[str, Any] = raw_data["assets"] raw_assets: List[Dict[str, str]] = data["asset"] @@ -217,13 +240,18 @@ def build_sporecast_assets(raw_data: Dict[str, Any]) -> SporecastAssets: ) -def build_achievements(raw_data: Dict[str, Any]) -> Achievements: +def parse_achievements(text: str) -> Achievements: """ [Pages] Achievements for user: http://www.spore.com/rest/assets/achievements/// """ + raw_data: Dict[str, Any] = xmltodict.parse( + text, + force_list=("achievement",), + ) # type: ignore + data: Dict[str, Any] = raw_data["achievements"] raw_achievements: List[Dict[str, str]] = data["achievement"] @@ -254,7 +282,9 @@ def build_achievements(raw_data: Dict[str, Any]) -> Achievements: name=name, description=description, guid=raw_achievement["guid"], - image_url=f"{BASE_URL}/static/war/images/achievements/{raw_achievement['guid']}.png", + image_url=( + f"{BASE_URL}/static/war/images/achievements/{raw_achievement['guid']}.png" + ), date=datatime_from_string(raw_achievement["date"]) ) ) @@ -266,13 +296,18 @@ def build_achievements(raw_data: Dict[str, Any]) -> Achievements: @staticmethod -def build_full_asset(raw_data: Dict[str, Any]) -> FullAsset: +def parse_full_asset(text: str) -> FullAsset: """ [Pages] Info about an asset: http://www.spore.com/rest/asset/ """ + raw_data: Dict[str, Any] = xmltodict.parse( + text, + force_list=("comment",), + ) # type: ignore + data: Dict[str, Any] = raw_data["asset"] raw_comments: List[Dict[str, str]] = data["comments"]["comment"] @@ -312,13 +347,18 @@ def build_full_asset(raw_data: Dict[str, Any]) -> FullAsset: ) -def build_asset_comments(raw_data: Dict[str, Any]) -> AssetComments: +def parse_asset_comments(text: str) -> AssetComments: """ [Pages] Comments for an asset: http://www.spore.com/rest/comments/// """ + raw_data: Dict[str, Any] = xmltodict.parse( + text, + force_list=("comment",), + ) # type: ignore + data: Dict[str, Any] = raw_data["comments"] raw_comments: List[Dict[str, str]] = data["comment"] @@ -335,7 +375,7 @@ def build_asset_comments(raw_data: Dict[str, Any]) -> AssetComments: ) -def build_buddies(raw_data: Dict[str, Any]) -> Buddies: +def parse_buddies(text: str) -> Buddies: """ [Pages] @@ -345,6 +385,11 @@ def build_buddies(raw_data: Dict[str, Any]) -> Buddies: Who has added a user as a buddy: http://www.spore.com/rest/users/subscribers/// """ + raw_data: Dict[str, Any] = xmltodict.parse( + text, + force_list=("buddy",), + ) # type: ignore + data: Dict[str, Any] = raw_data["users"] raw_buddy: List[Dict[str, str]] = data["buddy"]