feat(client): dynamically support single and multi models w/ typing (#11

)

TODO.md

@@ -1,7 +1,5 @@
 # TODO
 
-- Add a proper logger to debug or just spit out general info on what's happening
-
 - Add registration for listener classes (so you do not need to have access to the client object, similar to discord.py's Cogs)
 
 - Implement all models/endpoints
@@ -10,3 +8,7 @@ endpoints
 
   - Item class inherits from WorldstateObject and MultiQueryModel
   - Put all major infos in the Item class (split them later)
+
+## Version 2.0
+
+- Remove `WorldstateClient.query_list_of(type)` as `WorldstateClient.query(type)` now does the same while keeping the type.

examples/worldstate/custom_models.py

@@ -1,30 +1,37 @@
 import asyncio
-from datetime import datetime
-from typing import Literal, Optional
-
-from msgspec import field  # use this to rename response keys
+from typing import Literal
 
 from warframe.worldstate import WorldstateClient
-from warframe.worldstate.common.core import (
+from warframe.worldstate.common.core import (  # this import might change
     SingleQueryModel,
-)  # this import might change
+    TimedEvent,
+)
 
 
-class CustomCambionDrift(SingleQueryModel):
-    __endpoint__ = "/cambionCycle"  # specify endpoint here
+class CustomCambionDrift(SingleQueryModel, TimedEvent):
+    """
+    This "feature" is there to provide an option to implement something that is not yet implemented.
 
-    # required
+    Short explanation:
+    The Cambion Drift's endpoint responds in a single (json) object that's why we give it `SingleQueryModel`.
+    It is also a `TimedEvent` because it has an activation, expiry, etc. (basically a cycle).
 
-    # at the endpoint, it is called expiry, to rename it use this:
-    expiry_dt: datetime = field(name="expiry")
+    Now, to add the attributes to the class simply look at the API docs.
+    Note that existing names will be converted from `camelCase` to `snake_case`.
+    To rename a given key (from the API response) simply use `msgspec.field` and set the `name` to its original, snake_case'd name.
+    e.g.:
+    ```py
+        my_renamed_state: Literal["vome", "fass"] = msgspec.field(name="state")
+    ```
 
-    activation: datetime
-    state: Literal["vome", "fass"]
+    `https://api.warframestat.us/pc` will be concatenated with `__endpoint__`.
 
-    # optional
+    After that's all done, simply slap your own Model into `client.query` and let warframe.py do its magic :)
+    """
 
-    # to rename with default values, use this:
-    time_remaining: Optional[str] = field(name="timeLeft", default=None)
+    __endpoint__ = "/cambionCycle"  # specify endpoint here
+
+    state: Literal["vome", "fass"]
 
 
 async def main():

examples/worldstate/query.py

@@ -4,15 +4,18 @@
 import asyncio
 
 from warframe.worldstate import WorldstateClient
-from warframe.worldstate.models import Arbitration
+from warframe.worldstate.models import Arbitration, Fissure
 
 
 async def main():
     async with WorldstateClient() as client:
         # import from models and pass the type you want the object of
-        arbi = await client.query(Arbitration)
+        arbi = await client.query(Arbitration)  # of Type SingleQuery - a single  object  of type `Arbitration`
+        fissures = await client.query(Fissure)  # of Type MultiQuery  - a list of objects of type `Fissure`
 
         print(arbi)
+        for fissure in fissures:
+            print(fissure)
 
 
 if __name__ == "__main__":

warframe/worldstate/client.py

@@ -2,7 +2,17 @@
 import logging
 from datetime import datetime, timezone
 from functools import wraps
-from typing import Any, Callable, Coroutine, List, Optional, Type, TypeVar, Union
+from typing import (
+    Any,
+    Callable,
+    Coroutine,
+    List,
+    Optional,
+    Type,
+    TypeVar,
+    Union,
+    overload,
+)
 
 import aiohttp
 import msgspec
@@ -125,8 +135,11 @@ async def _request(
     # Queries
     #
 
+    @overload
     async def query(
-        self, cls: Type[SupportsSingleQuery], language: Optional[Language] = None
+        self,
+        cls: Type[SupportsSingleQuery],
+        language: Optional[Language] = None,
     ) -> SupportsSingleQuery:
         """
         Queries the model of type `SingleQueryModel` to return its corresponding object.
@@ -136,18 +149,44 @@ async def query(
         cls : Type[SupportsSingleQuery]
             The model to query.
         language : Optional[Language], optional
-            The language to return the object in, by default None.
-
-        Raises
-        ------
-        UnsupportedSingleQueryError
-            When the passed type `cls` is not a subclass of `SingleQueryModel`.
+            The language to return the queried model in, by default None
 
         Returns
         -------
         SupportsSingleQuery
             The queried model.
         """
+        ...
+
+    @overload
+    async def query(
+        self,
+        cls: Type[SupportsMultiQuery],
+        language: Optional[Language] = None,
+    ) -> List[SupportsMultiQuery]:
+        """
+        Queries the model of type `MultiQueryModel` to return a list of its corresponding object.
+
+        Parameters
+        ----------
+        cls : Type[SupportsMultiQuery]
+            The model to query.
+        language : Optional[Language], optional
+            The language to return the queried model in, by default None
+
+        Returns
+        -------
+        List[SupportsMultiQuery]
+            A list of the queried model.
+        """
+        ...
+
+    async def query(
+        self,
+        cls: Type[Union[SupportsSingleQuery, SupportsMultiQuery]],
+        language: Optional[Language] = None,
+    ) -> Union[SupportsSingleQuery, List[SupportsMultiQuery]]:
+        # -----
         json = await self._request(cls, language)
         return cls._from_json(json)
 
@@ -174,6 +213,9 @@ async def query_list_of(
         Optional[List[SupportsMultiQuery]]
             A list of the queried model.
         """
+        logging.getLogger(__name__).warn(
+            "Deprecation warning: `query_list_of(type)` is deprecated and will be removed in version 2.0. Use `query(type)` instead."
+        )
         json = await self._request(cls, language)
         return cls._from_json(json)