1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
"""
Xbox Live Client
Basic factory that stores :class:`XboxLiveLanguage`, User authorization data
and available `Providers`
"""
import logging
from typing import Any
from httpx import Response
from ms_cv import CorrelationVector
from xbox.webapi.api.language import DefaultXboxLiveLanguages, XboxLiveLanguage
from xbox.webapi.api.provider.account import AccountProvider
from xbox.webapi.api.provider.achievements import AchievementsProvider
from xbox.webapi.api.provider.catalog import CatalogProvider
from xbox.webapi.api.provider.cqs import CQSProvider
from xbox.webapi.api.provider.gameclips import GameclipProvider
from xbox.webapi.api.provider.lists import ListsProvider
from xbox.webapi.api.provider.mediahub import MediahubProvider
from xbox.webapi.api.provider.message import MessageProvider
from xbox.webapi.api.provider.people import PeopleProvider
from xbox.webapi.api.provider.presence import PresenceProvider
from xbox.webapi.api.provider.profile import ProfileProvider
from xbox.webapi.api.provider.screenshots import ScreenshotsProvider
from xbox.webapi.api.provider.smartglass import SmartglassProvider
from xbox.webapi.api.provider.titlehub import TitlehubProvider
from xbox.webapi.api.provider.usersearch import UserSearchProvider
from xbox.webapi.api.provider.userstats import UserStatsProvider
from xbox.webapi.authentication.manager import AuthenticationManager
log = logging.getLogger("xbox.api")
class Session:
def __init__(self, auth_mgr: AuthenticationManager):
self._auth_mgr = auth_mgr
self._cv = CorrelationVector()
async def request(
self,
method: str,
url: str,
include_auth: bool = True,
include_cv: bool = True,
**kwargs: Any,
) -> Response:
"""Proxy Request and add Auth/CV headers."""
headers = kwargs.pop("headers", {})
params = kwargs.pop("params", None)
data = kwargs.pop("data", None)
# Extra, user supplied values
extra_headers = kwargs.pop("extra_headers", None)
extra_params = kwargs.pop("extra_params", None)
extra_data = kwargs.pop("extra_data", None)
if include_auth:
# Ensure tokens valid
await self._auth_mgr.refresh_tokens()
# Set auth header
headers[
"Authorization"
] = self._auth_mgr.xsts_token.authorization_header_value
if include_cv:
headers["MS-CV"] = self._cv.increment()
# Extend with optionally supplied values
if extra_headers:
headers.update(extra_headers)
if extra_params:
# query parameters
params = params or {}
params.update(extra_params)
if extra_data:
# form encoded post data
data = data or {}
data.update(extra_data)
return await self._auth_mgr.session.request(
method, url, **kwargs, headers=headers, params=params, data=data
)
async def get(self, url: str, **kwargs: Any) -> Response:
return await self.request("GET", url, **kwargs)
async def options(self, url: str, **kwargs: Any) -> Response:
return await self.request("OPTIONS", url, **kwargs)
async def head(self, url: str, **kwargs: Any) -> Response:
return await self.request("HEAD", url, **kwargs)
async def post(self, url: str, **kwargs: Any) -> Response:
return await self.request("POST", url, **kwargs)
async def put(self, url: str, **kwargs: Any) -> Response:
return await self.request("PUT", url, **kwargs)
async def patch(self, url: str, **kwargs: Any) -> Response:
return await self.request("PATCH", url, **kwargs)
async def delete(self, url: str, **kwargs: Any) -> Response:
return await self.request("DELETE", url, **kwargs)
class XboxLiveClient:
def __init__(
self,
auth_mgr: AuthenticationManager,
language: XboxLiveLanguage = DefaultXboxLiveLanguages.United_States,
):
self._auth_mgr = auth_mgr
self.session = Session(auth_mgr)
self._language = language
self.cqs = CQSProvider(self)
self.lists = ListsProvider(self)
self.profile = ProfileProvider(self)
self.achievements = AchievementsProvider(self)
self.usersearch = UserSearchProvider(self)
self.gameclips = GameclipProvider(self)
self.people = PeopleProvider(self)
self.presence = PresenceProvider(self)
self.mediahub = MediahubProvider(self)
self.message = MessageProvider(self)
self.userstats = UserStatsProvider(self)
self.screenshots = ScreenshotsProvider(self)
self.titlehub = TitlehubProvider(self)
self.account = AccountProvider(self)
self.catalog = CatalogProvider(self)
self.smartglass = SmartglassProvider(self)
@property
def xuid(self) -> str:
"""
Gets the Xbox User ID
Returns: Xbox user Id
"""
return self._auth_mgr.xsts_token.xuid
@property
def language(self) -> XboxLiveLanguage:
"""
Gets the active Xbox Live Language
Returns: Active Xbox Live language
"""
return self._language
|
