Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
| Max length: ``16``
Raises
------
ValueError
The nickname contains too few/many characters or contains invalid
characters.
HTTPException
An error occured while requesting.
"""
if not (3 <= len(nickname) <= 16):
raise ValueError('Invalid nickname length')
try:
await self.client.http.friends_set_nickname(self.id, nickname)
except HTTPException as e:
ignored = ('errors.com.epicgames.common.unsupported_media_type',
'errors.com.epicgames.validation.validation_failed')
if e.message_code in ignored:
raise ValueError('Invalid nickname')
raise
self._nickname = nickname
async def event_ping_received(self, ctx: EventContext) -> None:
body = ctx.body
pinger = body['pinger_id']
try:
data = (await self.client.http.party_lookup_ping(pinger))[0]
except IndexError:
return
except HTTPException as exc:
m = 'errors.com.epicgames.social.party.ping_not_found'
if exc.message_code == m:
return
raise
for inv in data['invites']:
if inv['sent_by'] == pinger and inv['status'] == 'SENT':
invite = inv
break
else:
invite = self._create_invite(pinger, {**body, **data})
if 'urn:epic:cfg:build-id_s' not in invite['meta']:
pres = self.client.get_presence(pinger)
if (pres is not None and pres.party is not None
config: Optional[dict] = None) -> ClientParty:
aquiring = not self.auth._refresh_lock.locked()
try:
if aquiring:
await self._join_party_lock.acquire()
if isinstance(config, dict):
cf = {**self.default_party_config.config, **config}
else:
cf = self.default_party_config.config
while True:
try:
data = await self.http.party_create(cf)
break
except HTTPException as exc:
if exc.message_code != ('errors.com.epicgames.social.'
'party.user_has_party'):
raise
data = await self.http.party_lookup_user(self.user.id)
async with self._leave_lock:
try:
await self.http.party_leave(
data['current'][0]['id']
)
except HTTPException as e:
m = ('errors.com.epicgames.social.'
'party.party_not_found')
if e.message_code != m:
raise
if not is_multiple:
graphql = (graphql,)
kwargs['json'] = [gql_query.as_multiple_payload()
for gql_query in graphql]
kwargs['headers'] = headers
raw = kwargs.pop('raw', False)
r, data = await self.request(method, url, **kwargs)
if raw:
return r
if 'errorCode' in data:
raise HTTPException(r, data)
if graphql is not None:
error_data = None
for child_data in data:
if 'errors' in child_data:
error_data = child_data['errors']
if error_data is not None:
selected = error_data[0]
obj = {'errorMessage': selected['message']}
service_response = selected['serviceResponse']
if service_response == '':
error_payload = {}
else:
async def fn_request(self, method: str,
route: Union[Route, List[Route]],
auth: Optional[str] = None,
graphql: Union[Route, List[Route]] = None,
**kwargs: Any) -> Any:
try:
return await self._fn_request(method, route, auth, graphql,
**kwargs)
except HTTPException as exc:
catch = (
'errors.com.epicgames.common.oauth.invalid_token',
('errors.com.epicgames.common.authentication.'
'token_verification_failed')
)
if exc.message_code in catch and not self.client._closing:
await self.client.restart()
return await self.fn_request(method, route, auth, graphql,
**kwargs)
elif exc.message_code in ('errors.com.epicgames.common.'
'server_error',):
await asyncio.sleep(0.5)
return await self._fn_request(method, route, auth, graphql,
**kwargs)
code = self.two_factor_code
if code is None:
async with _prompt_lock:
code = await ainput(
'Please enter the 2fa code:\n',
loop=self.client.loop
)
try:
await self.client.http.epicgames_mfa_login(
e.raw['metadata']['twoFactorMethod'],
code,
token
)
except HTTPException as exc:
m = (
'errors.com.epicgames.accountportal.mfa_code_invalid',
'errors.com.epicgames.accountportal.validation'
)
if exc.message_code in m:
raise AuthException(
'Invalid 2fa code passed.',
exc
) from exc
raise
await self.client.http.epicgames_redirect(token)
token = await self.fetch_xsrf_token()
log.info('Fetching exchange code.')
error_data = None
for child_data in data:
if 'errors' in child_data:
error_data = child_data['errors']
if error_data is not None:
selected = error_data[0]
obj = {'errorMessage': selected['message']}
service_response = selected['serviceResponse']
if service_response == '':
error_payload = {}
else:
error_payload = json.loads(service_response)
raise HTTPException(r, {**obj, **error_payload})
def get_payload(d):
return next(iter(d['data'].values()))
if len(data) == 1:
return get_payload(data[0])
return [get_payload(d) for d in data]
return data
async def ios_authenticate(self) -> dict:
data = None
prompt_message = ''
if self.device_auth_ready():
try:
return await self.run_device_authenticate()
except AuthException as exc:
original = exc.original
if not self.prompt_enabled() or not self.prompt_code_if_invalid: # noqa
raise
if isinstance(original, HTTPException):
m = 'errors.com.epicgames.account.invalid_account_credentials' # noqa
if original.message_code != m:
raise
prompt_message = 'Invalid device auth details passed. '
elif self.email_and_password_ready():
try:
data = await self.run_email_and_password_authenticate()
except HTTPException as e:
m = {
'errors.com.epicgames.accountportal.captcha_invalid': 'Captcha was enforced. ' # noqa
}
if self.prompt_enabled() and self.prompt_code_if_throttled:
m['errors.com.epicgames.common.throttled'] = 'Account was throttled. ' # noqa
async def _leave(self, ignore_not_found: bool = True) -> None:
await self.client.xmpp.leave_muc()
try:
await self.client.http.party_leave(self.id)
except HTTPException as e:
m = 'errors.com.epicgames.social.party.party_not_found'
if ignore_not_found and e.message_code == m:
return
raise