Skip to content

Commit

Permalink
Merge pull request #71 from LEv145:dev
Browse files Browse the repository at this point in the history 
Bug fix
  • Loading branch information
@LEv145
LEv145 authored Dec 18, 2021
2 parents 3796808 + bd46d00 commit 1d265e4
Show file tree
Hide file tree
Showing 3 changed files with 152 additions and 109 deletions.
22 changes: 11 additions & 11 deletions spore_api/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
172 changes: 85 additions & 87 deletions spore_api/client.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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
Expand All @@ -65,154 +56,161 @@ 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(
self,
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"<status>(\d+)</status>", text)
if api_status_parse is not None:
api_status = int(api_status_parse.group(1))

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")

Expand Down
Loading

0 comments on commit 1d265e4

Please sign in to comment.