Merge pull request #110 from anxdpanic/pr_isengard

add oauth2 token validation used with helix api

Author: anxdpanic

resources/lib/twitch/api/helix/streams.py

@@ -38,6 +38,17 @@ def get_streams(game_id=list(), user_id=list(),
     return q
 
 
+# required scope: user:read:follows
+@query
+def get_followed(user_id, after='MA==', first=20, use_app_token=False):
+    q = Qry('streams', use_app_token=use_app_token)
+    q.add_param(keys.AFTER, Cursor.validate(after), 'MA==')
+    q.add_param(keys.FIRST, IntRange(1, 100).validate(first), 20)
+    q.add_param(keys.USER_ID, user_id)
+
+    return q
+
+
 # required scope: none
 @query
 def get_metadata(game_id=list(), user_id=list(),

resources/lib/twitch/oauth/__init__.py

@@ -9,9 +9,10 @@
     See LICENSES/GPL-3.0-only for more information.
 """
 
-__all__ = ['v5', 'default', 'helix', 'clients']
+__all__ = ['v5', 'default', 'helix', 'clients', 'validation']
 
 from . import v5  # V5 is deprecated and will be removed entirely on TBD
 from . import helix
 from . import v5 as default
 from . import clients
+from . import validation

resources/lib/twitch/oauth/validation.py

@@ -0,0 +1,19 @@
+# -*- encoding: utf-8 -*-
+"""
+
+    Copyright (C) 2022 script.module.python.twitch
+
+    This file is part of script.module.python.twitch
+
+    SPDX-License-Identifier: GPL-3.0-only
+    See LICENSES/GPL-3.0-only for more information.
+"""
+
+from ..queries import OAuthValidationQuery as Qry
+from ..queries import query
+
+
+@query
+def validate(token=None):
+    q = Qry(token)
+    return q

resources/lib/twitch/queries.py

@@ -28,6 +28,7 @@
 _uploads_baseurl = 'https://uploads.twitch.tv/'
 _gql_baseurl = 'https://gql.twitch.tv/gql'
 _oauth_baseurl = 'https://api.twitch.tv/kraken/oauth2/'
+_oauthid_baseurl = 'https://id.twitch.tv/oauth2/'
 
 
 class _Query(object):
@@ -196,6 +197,17 @@ def __init__(self, path, headers={}, data={}, method=methods.GET):
         self.add_path(path)
 
 
+class OAuthValidationQuery(JsonQuery):
+    def __init__(self, token=None):
+        _headers = {}
+        if token:
+            _headers['Authorization'] = token
+        if 'Authorization' not in _headers:
+            _headers.setdefault('Authorization', 'OAuth {access_token}'.format(access_token=OAUTH_TOKEN))
+        super(JsonQuery, self).__init__(_oauthid_baseurl, _headers, {}, methods.GET)
+        self.add_path('validate')
+
+
 class ClipsQuery(JsonQuery):
     def __init__(self, path='', headers={}, data={}, method=methods.POST):
         _headers = deepcopy(headers)