Change requests to Route based

twitchio/authentication/oauth.py

@@ -52,8 +52,9 @@ async def validate_token(self, token: str, /) -> ValidateTokenPayload:
         token = token.removeprefix("Bearer ").removeprefix("OAuth ")
 
         headers: dict[str, str] = {"Authorization": f"OAuth {token}"}
-        data: ValidateTokenResponse = await self.request_json("GET", "/oauth2/validate", use_id=True, headers=headers)
+        route: Route = Route("GET", "/oauth2/validate", use_id=True, headers=headers)
 
+        data: ValidateTokenResponse = await self.request_json(route)
         return ValidateTokenPayload(data)
 
     async def refresh_token(self, refresh_token: str, /) -> RefreshTokenPayload:
@@ -66,9 +67,8 @@ async def refresh_token(self, refresh_token: str, /) -> RefreshTokenPayload:
             "client_secret": self.client_secret,
         }
 
-        data: RefreshTokenResponse = await self.request_json(
-            "POST", "/oauth2/token", use_id=True, headers=headers, params=params
-        )
+        route: Route = Route("POST", "/oauth2/token", use_id=True, headers=headers, params=params)
+        data: RefreshTokenResponse = await self.request_json(route)
 
         return RefreshTokenPayload(data)
 
@@ -88,9 +88,8 @@ async def user_access_token(self, code: str, /) -> UserTokenPayload:
             # "state": #TODO
         }
 
-        data: UserTokenResponse = await self.request_json(
-            "POST", "/oauth2/token", use_id=True, headers=headers, params=params
-        )
+        route: Route = Route("POST", "/oauth2/token", use_id=True, headers=headers, params=params)
+        data: UserTokenResponse = await self.request_json(route)
 
         return UserTokenPayload(data)
 
@@ -102,7 +101,8 @@ async def revoke_token(self, token: str, /) -> None:
             "token": token,
         }
 
-        await self.request_json("POST", "/oauth2/revoke", use_id=True, headers=headers, params=params)
+        route: Route = Route("POST", "/oauth2/revoke", use_id=True, headers=headers, params=params)
+        await self.request_json(route)
 
     async def client_credentials_token(self) -> ClientCredentialsPayload:
         headers: dict[str, str] = {"Content-Type": "application/x-www-form-urlencoded"}
@@ -113,17 +113,15 @@ async def client_credentials_token(self) -> ClientCredentialsPayload:
             "grant_type": "client_credentials",
         }
 
-        data: ClientCredentialsResponse = await self.request_json(
-            "POST", "/oauth2/token", use_id=True, headers=headers, params=params
-        )
+        route: Route = Route("POST", "/oauth2/token", use_id=True, headers=headers, params=params)
+        data: ClientCredentialsResponse = await self.request_json(route)
 
         return ClientCredentialsPayload(data)
 
     def get_authorization_url(self, scopes: Scopes, state: str = "") -> str:
         if not self.redirect_uri:
             raise ValueError("Missing redirect_uri")
 
-        route: Route = Route("GET", "/oauth2/authorize", use_id=True)
         params = {
             "client_id": self.client_id,
             "redirect_uri": urllib.parse.quote(self.redirect_uri),
@@ -132,5 +130,5 @@ def get_authorization_url(self, scopes: Scopes, state: str = "") -> str:
             "state": state,
         }
 
-        query_string = "&".join(f"{key}={value}" for key, value in params.items())
-        return f"{route}?{query_string}"
+        route: Route = Route("GET", "/oauth2/authorize", use_id=True, params=params)
+        return route.url

twitchio/authentication/scopes.py

@@ -187,8 +187,8 @@ def __contains__(self, scope: _scope_property | str, /) -> bool:
 
         return scope in self._selected
 
-    def urlsafe(self) -> str:
-        return "+".join([scope.quoted() for scope in self._selected])
+    def urlsafe(self, *, unquote: bool = False) -> str:
+        return "+".join([scope.quoted() if not unquote else scope.value for scope in self._selected])
 
     @property
     def selected(self) -> list[str]:

twitchio/http.py

@@ -25,8 +25,7 @@
 
 import logging
 import sys
-from functools import cached_property
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 import aiohttp
 
@@ -38,7 +37,7 @@
 if TYPE_CHECKING:
     from typing_extensions import Unpack
 
-    from .types_.requests import APIRequest, HTTPMethod
+    from .types_.requests import APIRequest, APIRequestKwargs, HTTPMethod
 
 
 logger: logging.Logger = logging.getLogger(__name__)
@@ -48,7 +47,7 @@ async def json_or_text(resp: aiohttp.ClientResponse) -> dict[str, Any] | str:
     text: str = await resp.text()
 
     try:
-        if resp.headers["Content-Type"] == "application/json":
+        if resp.headers["Content-Type"].startswith("application/json"):
             return _from_json(text)  # type: ignore
     except KeyError:
         pass
@@ -59,32 +58,53 @@ async def json_or_text(resp: aiohttp.ClientResponse) -> dict[str, Any] | str:
 class Route:
     # TODO: Document this class.
 
-    BASE: str = "https://api.twitch.tv/helix/"
-    ID_BASE: str = "https://id.twitch.tv/"
+    BASE: ClassVar[str] = "https://api.twitch.tv/helix/"
+    ID_BASE: ClassVar[str] = "https://id.twitch.tv/"
 
     def __init__(
-        self, method: HTTPMethod, endpoint: str, *, use_id: bool = False, **kwargs: Unpack[APIRequest]
+        self, method: HTTPMethod, path: str, *, use_id: bool = False, **kwargs: Unpack[APIRequestKwargs]
     ) -> None:
-        self.method = method
+        params: dict[str, str] = kwargs.pop("params", {})
+        self._url = self.build_url(path, use_id=use_id, params=params)
 
-        endpoint = endpoint.removeprefix("/")
-        self.endpoint = endpoint
+        self.use_id = use_id
+        self.method = method
+        self.path = path
 
-        if use_id:
-            self.url: str = self.ID_BASE + endpoint
-        else:
-            self.url: str = self.BASE + endpoint
+        self.params: dict[str, str] = params
+        self.data: dict[str, Any] = kwargs.get("data", {})
+        self.json: dict[str, Any] = kwargs.get("json", {})
+        self.headers: dict[str, str] = kwargs.get("headers", {})
 
-        self.params: dict[str, str] = kwargs.pop("params", {})
-        self.data: dict[str, Any] = kwargs.pop("data", {})
-        self.json: dict[str, Any] = kwargs.pop("json", {})
-        self.headers: dict[str, str] = kwargs.pop("headers", {})
+        self.packed: APIRequest = kwargs
 
     def __str__(self) -> str:
-        return self.url
+        return str(self._url)
 
     def __repr__(self) -> str:
-        return f"{self.method} /{self.endpoint}"
+        return f"{self.method}({self.path})"
+
+    @classmethod
+    def build_url(cls, path: str, use_id: bool = False, params: dict[str, str] = {}) -> str:
+        path_: str = path.lstrip("/")
+
+        url: str = f"{cls.ID_BASE if use_id else cls.BASE}{path_}{cls.build_query(params)}"
+        return url
+
+    def update_query(self, params: dict[str, str]) -> str:
+        self.params.update(params)
+        self.build_url(self.path, use_id=self.use_id, params=self.params)
+
+        return self._url
+
+    @property
+    def url(self) -> str:
+        return str(self._url)
+
+    @classmethod
+    def build_query(cls, params: dict[str, str]) -> str:
+        joined: str = "&".join(f"{key}={value}" for key, value in params.items())
+        return f"?{joined}" if joined else ""
 
 
 class HTTPClient:
@@ -98,7 +118,7 @@ def __init__(self) -> None:
         ua = "TwitchioClient (https://github.com/PythonistaGuild/TwitchIO {0}) Python/{1} aiohttp/{2}"
         self.user_agent: str = ua.format(__version__, pyver, aiohttp.__version__)
 
-    @cached_property
+    @property
     def headers(self) -> dict[str, str]:
         return {"User-Agent": self.user_agent}
 
@@ -126,16 +146,13 @@ async def close(self) -> None:
             self.clear()
             logger.debug("%s session closed successfully.", self.__class__.__qualname__)
 
-    async def request(
-        self, method: HTTPMethod, endpoint: str, *, use_id: bool = False, **kwargs: Unpack[APIRequest]
-    ) -> Any:
+    async def request(self, route: Route) -> Any:
         await self._init_session()
         assert self.__session is not None
 
-        route: Route = Route(method, endpoint, use_id=use_id, **kwargs)
         logger.debug("Attempting a request to %r with %s.", route, self.__class__.__qualname__)
 
-        async with self.__session.request(method, route.url, **kwargs) as resp:
+        async with self.__session.request(route.method, route.url, **route.packed) as resp:
             data: dict[str, Any] | str = await json_or_text(resp)
 
             if resp.status >= 400:
@@ -147,10 +164,8 @@ async def request(
         # TODO: This method is not complete. This is purely for testing purposes.
         return data
 
-    async def request_json(
-        self, method: HTTPMethod, endpoint: str, *, use_id: bool = False, **kwargs: Unpack[APIRequest]
-    ) -> Any:
-        data = await self.request(method, endpoint, use_id=use_id, **kwargs)
+    async def request_json(self, route: Route) -> Any:
+        data = await self.request(route)
 
         if isinstance(data, str):
             # TODO: Add a TwitchioHTTPException here.

twitchio/types_/requests.py

@@ -25,14 +25,20 @@
 from typing import Any, Literal, TypeAlias, TypedDict
 
 
-__all__ = ("HTTPMethod", "APIRequest")
+__all__ = ("HTTPMethod", "APIRequestKwargs", "APIRequest")
 
 
 HTTPMethod: TypeAlias = Literal["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD", "CONNECT", "TRACE"]
 
 
-class APIRequest(TypedDict, total=False):
+class APIRequestKwargs(TypedDict, total=False):
     headers: dict[str, str]
     data: dict[str, Any]
     params: dict[str, str]
     json: dict[str, Any]
+
+
+class APIRequest(TypedDict, total=False):
+    headers: dict[str, str]
+    data: dict[str, Any]
+    json: dict[str, Any]