Skip to content

Commit

Permalink
Initial Scopes class implementation.
Browse files Browse the repository at this point in the history
  • Loading branch information
EvieePy committed Feb 10, 2024
1 parent a77cff5 commit b22a91c
Show file tree
Hide file tree
Showing 4 changed files with 210 additions and 6 deletions.
1 change: 1 addition & 0 deletions twitchio/authentication/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,3 +24,4 @@

from .oauth import OAuth as OAuth
from .payloads import *
from .scopes import Scopes as Scopes
14 changes: 8 additions & 6 deletions twitchio/authentication/oauth.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@
import urllib.parse
from typing import TYPE_CHECKING

from ..http import HTTPClient
from ..http import HTTPClient, Route
from .payloads import *


Expand All @@ -37,10 +37,11 @@
UserTokenResponse,
ValidateTokenResponse,
)
from .scopes import Scopes


class OAuth(HTTPClient):
def __init__(self, *, client_id: str, client_secret: str, redirect_uri: str | None) -> None:
def __init__(self, *, client_id: str, client_secret: str, redirect_uri: str | None = None) -> None:
super().__init__()

self.client_id = client_id
Expand Down Expand Up @@ -118,17 +119,18 @@ async def client_credentials_token(self) -> ClientCredentialsPayload:

return ClientCredentialsPayload(data)

def get_authorization_url(self, scopes: list[str], state: str = "") -> str:
def get_authorization_url(self, scopes: Scopes, state: str = "") -> str:
if not self.redirect_uri:
raise ValueError("Missing redirect_uri")

base_url = "https://id.twitch.tv/oauth2/authorize"
route: Route = Route("GET", "/oauth2/authorize", use_id=True)
params = {
"client_id": self.client_id,
"redirect_uri": urllib.parse.quote(self.redirect_uri),
"response_type": "code",
"scope": "+".join(urllib.parse.quote(scope) for scope in scopes),
"scope": scopes.urlsafe(),
"state": state,
}

query_string = "&".join(f"{key}={value}" for key, value in params.items())
return f"{base_url}?{query_string}"
return f"{route}?{query_string}"
199 changes: 199 additions & 0 deletions twitchio/authentication/scopes.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,199 @@
"""
MIT License
Copyright (c) 2017 - Present PythonistaGuild
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
"""
from __future__ import annotations

import urllib.parse
from typing import TYPE_CHECKING


if TYPE_CHECKING:
from collections.abc import Iterable, Iterator


class _scope_property:
def __set_name__(self, owner: type[Scopes], name: str) -> None:
self._name = name

def __get__(self, *_: object) -> _scope_property:
return self

def __set__(self, instance: Scopes, value: bool) -> None:
if value is True:
instance._selected.add(self)
elif value is False:
instance._selected.discard(self)
else:
raise TypeError(f"Expected bool for scope, got {type(value).__name__}")

def __str__(self) -> str:
return self._name.replace("_", ":")

def quoted(self) -> str:
return urllib.parse.quote(str(self))

@property
def name(self) -> str:
return self._name

@property
def value(self) -> str:
return str(self)

def __hash__(self) -> int:
return hash(self._name)

def __eq__(self, other: object, /) -> bool:
if not isinstance(other, (_scope_property, str)):
return NotImplemented

return str(self) == str(other)


class _ScopeMeta(type):
def __setattr__(self, name: str, value: object, /) -> None:
raise AttributeError("Cannot set the value of a Scope property.")

def __delattr__(self, name: str, /) -> None:
raise AttributeError("Cannot delete the value of a Scope property.")


class Scopes(metaclass=_ScopeMeta):
__slots__ = ("_selected",)

analytics_read_extensions = _scope_property()
analytics_read_games = _scope_property()
bits_read = _scope_property()
channel_manage_ads = _scope_property()
channel_read_ads = _scope_property()
channel_manage_broadcast = _scope_property()
channel_read_charity = _scope_property()
channel_edit_commercial = _scope_property()
channel_read_editors = _scope_property()
channel_manage_extensions = _scope_property()
channel_read_goals = _scope_property()
channel_read_guest_star = _scope_property()
channel_manage_guest_star = _scope_property()
channel_read_hype_train = _scope_property()
channel_manage_moderators = _scope_property()
channel_read_polls = _scope_property()
channel_manage_polls = _scope_property()
channel_read_predictions = _scope_property()
channel_manage_predictions = _scope_property()
channel_manage_raids = _scope_property()
channel_read_redemptions = _scope_property()
channel_manage_redemptions = _scope_property()
channel_manage_schedule = _scope_property()
channel_read_stream_key = _scope_property()
channel_read_subscriptions = _scope_property()
channel_manage_videos = _scope_property()
channel_read_vips = _scope_property()
channel_manage_vips = _scope_property()
clips_edit = _scope_property()
moderation_read = _scope_property()
moderator_manage_announcements = _scope_property()
moderator_manage_automod = _scope_property()
moderator_read_automod_settings = _scope_property()
moderator_manage_automod_settings = _scope_property()
moderator_manage_banned_users = _scope_property()
moderator_read_blocked_terms = _scope_property()
moderator_manage_blocked_terms = _scope_property()
moderator_manage_chat_messages = _scope_property()
moderator_read_chat_settings = _scope_property()
moderator_manage_chat_settings = _scope_property()
moderator_read_chatters = _scope_property()
moderator_read_followers = _scope_property()
moderator_read_guest_star = _scope_property()
moderator_manage_guest_star = _scope_property()
moderator_read_shield_mode = _scope_property()
moderator_manage_shield_mode = _scope_property()
moderator_read_shoutouts = _scope_property()
moderator_manage_shoutouts = _scope_property()
user_edit = _scope_property()
user_edit_follows = _scope_property()
user_read_blocked_users = _scope_property()
user_manage_blocked_users = _scope_property()
user_read_broadcast = _scope_property()
user_manage_chat_color = _scope_property()
user_read_email = _scope_property()
user_read_follows = _scope_property()
user_read_moderated_channels = _scope_property()
user_read_subscriptions = _scope_property()
user_manage_whispers = _scope_property()
channel_bot = _scope_property()
channel_moderate = _scope_property()
chat_edit = _scope_property()
chat_read = _scope_property()
user_bot = _scope_property()
user_read_chat = _scope_property()
user_write_chat = _scope_property()
whispers_read = _scope_property()
whispers_edit = _scope_property()

def __init__(self, scopes: Iterable[str | _scope_property] = [], /, **kwargs: bool) -> None:
self._selected: set[_scope_property] = set()

prop: _scope_property

for scope in scopes:
if isinstance(scope, str):
prop = getattr(self, scope.replace(":", "_"))
elif isinstance(scope, _scope_property): # type: ignore
prop = scope
else:
raise TypeError(f"Invalid scope provided: {type(scope)} is not a valid scope.")

self._selected.add(prop)

for key, value in kwargs.items():
prop = getattr(self, key)

if value is True:
self._selected.add(prop)
elif value is False:
self._selected.discard(prop)
else:
raise TypeError(f'Expected bool for scope kwarg "{key}", got {type(value).__name__}')

def __iter__(self) -> Iterator[str]:
return iter([str(scope) for scope in self._selected])

def __repr__(self) -> str:
return f"<Scopes selected={list(self)}>"

def __contains__(self, scope: _scope_property | str, /) -> bool:
if isinstance(scope, str):
return any(s.value == scope for s in self._selected)

return scope in self._selected

def urlsafe(self) -> str:
return "+".join([scope.quoted() for scope in self._selected])

@property
def selected(self) -> list[str]:
return list(self)

@classmethod
def all(cls) -> Scopes:
return cls([scope for scope in cls.__dict__.values() if isinstance(scope, _scope_property)])
2 changes: 2 additions & 0 deletions twitchio/types_/responses.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ class RefreshTokenResponse(TypedDict):
scope: str | list[str]
token_type: str


class UserTokenResponse(TypedDict):
access_token: str
refresh_token: str
Expand All @@ -55,6 +56,7 @@ class ValidateTokenResponse(TypedDict):
user_id: str
expires_in: int


class ClientCredentialsResponse(TypedDict):
access_token: str
expires_in: int
Expand Down

0 comments on commit b22a91c

Please sign in to comment.