Create FullWarApi.py (#7)

Create FullWarApi.py

Author: MagicTheDev

coc/abc.py

@@ -150,8 +150,8 @@ def share_link(self) -> str:
 class DataContainerMetaClass(type):
     def __repr__(cls):
         attrs = [
-            ("name", cls.name),
-            ("id", cls.id),
+            ("name", cls.name if 'name' in cls.__dict__ else "not initialized"),
+            ("id", cls.id if 'id' in cls.__dict__ else "not initialized"),
         ]
         return "<%s %s>" % (cls.__name__, " ".join("%s=%r" % t for t in attrs),)
 

coc/client.py

@@ -270,7 +270,7 @@ async def login(self, email: str, password: str) -> None:
         self._create_holders()
         LOG.debug("HTTP connection created. Client is ready for use.")
 
-    def login_with_keys(self, *keys: str) -> None:
+    async def login_with_keys(self, *keys: str) -> None:
         """Retrieves all keys and creates an HTTP connection ready for use.
 
         Parameters
@@ -281,7 +281,7 @@ def login_with_keys(self, *keys: str) -> None:
         http._keys = keys
         http.keys = cycle(http._keys)
         http.key_count = len(keys)
-        self.loop.run_until_complete(http.create_session(self.connector, self.timeout))
+        await http.create_session(self.connector, self.timeout)
         self._create_holders()
 
         LOG.debug("HTTP connection created. Client is ready for use.")

coc/ext/FullWarApi.py

@@ -0,0 +1,216 @@
+"""An extension that helps interact with the FullWar API."""
+
+import asyncio
+import base64
+import logging
+import json
+
+from collections import namedtuple
+from datetime import datetime
+from typing import List
+
+import aiohttp
+
+from ..http import json_or_text
+from ..utils import correct_tag
+from ..wars import ClanWar
+
+LOG = logging.getLogger(__name__)
+
+AccessToken = namedtuple("AccessToken", ["token", "expires_at"])
+
+
+def extract_expiry_from_jwt_token(token):
+    if isinstance(token, str):
+        token = token.encode("utf-8")
+    elif not isinstance(token, bytes):
+        # token was wrong somehow
+        return None
+
+    try:
+        signing, _ = token.rsplit(b".", 1)
+        _, payload = signing.split(b".", 1)
+    except ValueError:
+        return None  # not enough segments
+
+    if len(payload) % 4 > 0:
+        payload += b"=" * (4 - len(payload) % 4)
+
+    bytes_payload = base64.urlsafe_b64decode(payload)
+    dict_payload = json.loads(bytes_payload)
+    try:
+        expiry = dict_payload["exp"]
+        return datetime.fromtimestamp(expiry)
+    except KeyError:
+        return None
+
+
+async def login(username: str, password: str, clash_client) -> "FullWarClient":
+    """Eases logging into the API client.
+
+    For more information on this project, please join the discord server - <discord.gg/Eaja7gJ>
+
+    You must have your username and password as given on the server.
+    If unsure as to what this means, please reach out to an admin.
+
+    Parameters
+    -----------
+    username : str
+        Your username as given on the discord server.
+    password : str
+        Your password as given on the discord server
+    loop : Optional[:class:`asyncio.AbstractEventLoop`]
+        The :class:`asyncio.AbstractEventLoop` to use for HTTP requests.
+        An :func:`asyncio.get_event_loop()` will be used if ``None`` is passed
+    """
+    if not isinstance(username, str) or not isinstance(password, str):
+        raise TypeError("username and password must both be a string")
+    if not username or not password:
+        raise ValueError("username or password must not be an empty string.")
+
+    loop = asyncio.get_running_loop()
+    return FullWarClient(username, password, clash_client, loop)
+
+
+class FullWarClient:
+    """An extension that helps interact with the Full War API.
+
+    For more information on this project, please join the discord server - <discord.gg/Eaja7gJ>
+
+    You must have your username and password as given on the server.
+    If unsure as to what this means, please reach out to an admin.
+
+    Parameters
+    -----------
+    username : str
+        Your username as given on the discord server.
+    password : str
+        Your password as given on the discord server
+    clash_client: coc.Client
+        Client to use
+
+    loop : Optional[:class:`asyncio.AbstractEventLoop`]
+        The :class:`asyncio.AbstractEventLoop` to use for HTTP requests.
+        An :func:`asyncio.get_event_loop()` will be used if ``None`` is passed
+
+    """
+
+    BASE_URL = "http://teamutils.com:8081"
+
+    __slots__ = ("username", "password", "clash_client", "loop", "key", "http_session")
+
+    def __init__(self, username: str, password: str, clash_client, loop: asyncio.AbstractEventLoop = None):
+        self.username = username
+        self.password = password
+        self.clash_client = clash_client
+
+        self.loop = loop or asyncio.get_event_loop()
+        self.key = None  # set in get_key()
+
+        self.http_session = aiohttp.ClientSession(loop=self.loop)
+
+    async def close(self):
+        """Close the client session established"""
+        await self.http_session.close()
+
+    async def _request(self, method, url, *, token_request: bool = False, **kwargs):
+        url = self.BASE_URL + url
+
+        if not token_request:
+            key = await self._get_key()
+
+            headers = {"authorization": "Bearer {}".format(key)}
+            kwargs["headers"] = headers
+
+        async with self.http_session.request(method, url, **kwargs) as response:
+            LOG.debug("%s (%s) has returned %s", url, method, response.status)
+            data = await json_or_text(response)
+            LOG.debug(data)
+
+            if 200 <= response.status < 300:
+                LOG.debug("%s has received %s", url, data)
+                return data
+
+            if response.status == 401:
+                await self._refresh_key()
+                return await self._request(method, url, **kwargs)
+
+    async def _get_key(self):
+        if not self.key or self.key.expires_at < datetime.utcnow():
+            await self._refresh_key()
+
+        return self.key.token
+
+    async def _refresh_key(self):
+        #REMOVE THIS EMAIL WHEN POSSIBLE
+        data = {
+            "email" : "[email protected]",
+            "username": self.username,
+            "password": self.password,
+        }
+
+        payload = await self._request("POST", "/login", token_request=True, json=data)
+        self.key = AccessToken(payload["access_token"], extract_expiry_from_jwt_token(payload["access_token"]))
+
+    async def war_result(self, clan_tag: str, preparation_start: int = 0) -> ClanWar:
+        """Get a stored war result.
+
+        Parameters
+        ----------
+        client: coc.Client
+            instance of the clash client
+        clan_tag: str
+            The clan tag to find war result for.
+        preparation_start: int
+            Preparation start of a specific war result to find.
+
+        Returns
+        --------
+        #NOOOOO idea if this is correct xD
+        Optional[:class:`ClanWar`]
+            War result, or ``None`` if no war found.
+        """
+        data = await self._request("GET", f"/war_result?clan_tag={correct_tag(clan_tag, '%23')}&prep_start={str(preparation_start)}")
+        try:
+            return ClanWar(data=data["response"], client=self.clash_client)
+        except (IndexError, KeyError, TypeError, ValueError):
+            return None
+
+    async def war_result_log(self, clan_tag: str, preparation_start: int = 0) -> List[ClanWar]:
+        """Get all stored war results for a clan.
+
+        Parameters
+        ----------
+        client: coc.Client
+            instance of the clash client
+        clan_tag: str
+            The clan tag to find war result for.
+        preparation_start: int
+            Preparation start of a specific war result to find.
+
+        Returns
+        --------
+        #NOOOOO idea if this is correct xD
+        Optional[:class:`ClanWar`]
+            List of war results, or ``None`` if no wars found.
+        """
+        data = await self._request("GET", f"/war_result_log?clan_tag={correct_tag(clan_tag, '%23')}")
+        try:
+            responses = data["log"]
+            return [ClanWar(data=response["response"], client=self.clash_client) for response in responses]
+        except (IndexError, KeyError, TypeError, ValueError):
+            return None
+
+
+    async def register_war(self, clan_tag: str, preparation_start: int = 0):
+        """Registers a war.
+
+        Parameters
+        ----------
+        clan_tag : str
+            The clan to register a war for
+        preparation_start: int
+            Preparation time of the war
+        """
+        return await self._request("POST", f"/war_result?clan_tag={correct_tag(clan_tag, '%23')}&prep_start={str(preparation_start)}")
+

coc/hero.py

@@ -1,6 +1,8 @@
 from typing import TYPE_CHECKING, Dict, List, Type
 from pathlib import Path
 
+from attr import dataclass
+
 from .abc import DataContainer, DataContainerHolder
 
 if TYPE_CHECKING:
@@ -106,6 +108,7 @@ class HeroHolder(DataContainerHolder):
     data_object = Hero
 
 
+@dataclass
 class Pet(DataContainer):
     """Represents a Pet object as returned by the API, optionally filled with game data.
 

coc/login.py

@@ -78,5 +78,5 @@ def login_with_keys(*keys: str, client: Type[Union[Client, EventsClient]] = Clie
         Any kwargs you wish to pass into the Client object.
     """
     instance = client(**kwargs)
-    instance.login_with_keys(*keys)
+    instance.loop.run_until_complete(instance.login_with_keys(*keys))
     return instance

docs/code_overview/client.rst

@@ -16,9 +16,16 @@ Example
 ~~~~~~~
 .. code-block:: python3
 
+    import asyncio
     import coc
 
-    client = coc.login("email", "password", key_names="keys for my windows pc", key_count=5)
+    async def main():
+        async with coc.Client() as coc_client:
+            await coc_client.login("email", "password")
+
+        # do stuff
+
+    asyncio.run(main())
 
 With the returned instance, you can complete any of the operations detailed below.