PaiGramTeam/EnkaNetwork.py
3
0
Fork 0
mirror of https://github.com/PaiGramTeam/EnkaNetwork.py.git synced 2024-11-16 03:45:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1 from mrwan200/master

Browse Source 
test
This commit is contained in:
STACIA 2022-08-13 13:20:52 +07:00 committed by GitHub
commit f10e8b5432
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 11 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
enkanetwork/client.py
View File

@ -3,6 +3,7 @@ from __future__ import annotations
import os
import json
import logging
from time import time
from .http import HTTPClient
from .model import EnkaNetworkResponse
@ -22,7 +23,7 @@ class EnkaNetworkAPI:
LOGGER = logging.getLogger(__name__)
def __init__(self, lang: str = "en", *, debug: bool = False, key: str = "", cache: bool = True, agent: str = "") -> None: # noqa: E501
def __init__(self, lang: str = "en", *, debug: bool = False, key: str = "", cache: bool = True, agent: str = "", timeout: int = 10) -> None: # noqa: E501
# Logging
logging.basicConfig()
logging.getLogger("enkanetwork").setLevel(logging.DEBUG if debug else logging.ERROR) # noqa: E501
@ -35,7 +36,7 @@ class EnkaNetworkAPI:
self.cache = Cache(1024, 60 * 3)
# http client
self.__http = HTTPClient(key=key, agent=agent)
self.__http = HTTPClient(key=key, agent=agent, timeout=timeout)
self._closed = False
async def __aenter__(self) -> Self:

14
enkanetwork/http.py
View File

@ -48,6 +48,7 @@ class Route:
self.method = method
self.uid = uid
self.url = ''
if endpoint == 'enka':
self.url: str = self.BASE_URL.format(PATH=path)
else:
@ -57,11 +58,12 @@ class HTTPClient:
LOGGER = logging.getLogger(__name__)
def __init__(self, *, key: str = '', agent: str = '') -> None:
def __init__(self, *, key: str = '', agent: str = '', timeout: int = 5) -> None:
self.__session: aiohttp.ClientSession = MISSING
self.__headers: Dict = {}
self.__agent: str = agent
self.__key: str = key
self.__timeout = timeout or 10
async def close(self) -> None:
if self.__session is not MISSING:
@ -86,7 +88,7 @@ class HTTPClient:
data: Optional[Union[Dict[str, Any]]] = None
if self.__session is MISSING:
self.__session = aiohttp.ClientSession()
self.__session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=self.__timeout))
for tries in range(RETRY_MAX):
try:
@ -140,16 +142,16 @@ class HTTPClient:
r = Route(
'GET',
f'/u/{uid}/__data.json' + (f"?key={self.__key}" if self.__key else ""),
'enka',
uid
endpoint='enka',
uid=uid
)
return self.request(r)
def fetch_asset(self, folder: str, filename: str) -> Response[DefaultPayload]:
r = Route(
'GET',
f'master/exports/{folder}/{filename}'
'assets'
f'/master/exports/{folder}/{filename}',
endpoint='assets'
)
return self.request(r)