AccelByte Modular Python SDK
:warning: This accelbyte-python-modular-sdk
is not to be confused with accelbyte-python-sdk:
- The former (modular SDK) is experimental and is planned to be the sucessor for the latter (monolithic SDK).
- The modular SDK allows developers to include only the required modules in projects.
- If you are starting a new project, you may try to use modular SDK.
- If you use monolithic SDK in an existing project, a migration path is available via compatibility layer in modular SDK.
- Both monolithic and modular SDK will be maintained for some time to give time for migration until monolithic SDK is deprecated in the future.
A software development kit (SDK) for interacting with AccelByte Gaming Services (AGS) written in Python.
This SDK was generated from AGS OpenAPI spec files included in the spec directory.
Setup
This SDK requires Python 3.9 to be installed.
Install with Pip
Install the core package from PyPI
pip install accelbyte-py-sdk-core
and then install the service you need
pip install accelbyte-py-sdk-service-achievement
pip install accelbyte-py-sdk-service-ams
pip install accelbyte-py-sdk-service-basic
pip install accelbyte-py-sdk-service-cloudsave
pip install accelbyte-py-sdk-service-dslogmanager
pip install accelbyte-py-sdk-service-dsmc
pip install accelbyte-py-sdk-service-gametelemetry
pip install accelbyte-py-sdk-service-gdpr
pip install accelbyte-py-sdk-service-iam
pip install accelbyte-py-sdk-service-leaderboard
pip install accelbyte-py-sdk-service-legal
pip install accelbyte-py-sdk-service-lobby
pip install accelbyte-py-sdk-service-match2
pip install accelbyte-py-sdk-service-matchmaking
pip install accelbyte-py-sdk-service-platform
pip install accelbyte-py-sdk-service-qosm
pip install accelbyte-py-sdk-service-reporting
pip install accelbyte-py-sdk-service-seasonpass
pip install accelbyte-py-sdk-service-session
pip install accelbyte-py-sdk-service-sessionbrowser
pip install accelbyte-py-sdk-service-social
pip install accelbyte-py-sdk-service-ugc
and then install any feature you want
pip install accelbyte-py-sdk-feat-auth
pip install accelbyte-py-sdk-feat-token-validation
or install everything
pip install accelbyte-py-sdk-all
Environment Variables
The following environment variables need to be set when using EnvironmentConfigRepository
(default).
Name | Required | Example |
---|
AB_BASE_URL | Yes | https://demo/accelbyte.io |
AB_CLIENT_ID | Yes | abcdef0123456789abcdef0123456789 |
AB_CLIENT_SECRET | Yes, only if you use a private AB_CLIENT_ID | ab#c,d)ef(ab#c,d)ef(ab#c,d)ef(ab |
AB_NAMESPACE | Yes, the SDK will automatically fill up the {namespace} path parameter (overridable) | accelbyte |
AB_APP_NAME | No, the SDK will automatically fill up the User-Agent header (overridable) | MyApp |
AB_APP_VERSION | No, the SDK will automatically fill up the User-Agent header (overridable) | 1.0.0 |
Usage
Initializing
You'll have to initialize the SDK using the initialize()
function.
import accelbyte_py_sdk
if __name__ == "__main__":
accelbyte_py_sdk.initialize()
You could also pass in options like so:
from os import environ
import accelbyte_py_sdk
from accelbyte_py_sdk.core import MyConfigRepository
if __name__ == "__main__":
base_url = environ["MY_BASE_URL"]
client_id = environ["MY_CLIENT_ID"]
client_secret = environ["MY_CLIENT_SECRET"]
namespace = environ["MY_NAMESPACE"]
app_name = environ["MY_APP_NAME"]
app_version = environ["MY_APP_VERSION"]
my_config_repository = MyConfigRepository(
base_url=base_url,
client_id=client_id,
client_secret=client_secret,
namespace=namespace,
app_name=app_name,
app_version=app_version
)
options = {
"config": my_config_repository
}
accelbyte_py_sdk.initialize(options)
Logging In and Logging Out
Login using Username and Password
from os import environ
import accelbyte_py_sdk
from accelbyte_py_sdk.services.auth import login_user, logout
if __name__ == "__main__":
accelbyte_py_sdk.initialize()
username = environ["AB_USERNAME"]
password = environ["AB_PASSWORD"]
token, error = login_user(username, password)
assert error is None
_, error = logout()
assert error is None
Here login_user(username, password)
and logout()
are wrapper functions.
You can also specify the scope you want for login_user(...)
. With the scope
parameter typed as Optional[Union[str, List[str]]]
. By default the scope used is commerce account social publishing analytics
(a space seprated value string).
login_user(username, password, scope="scopeA")
Login using OAuth Client (Public or Confidential)
from os import environ
import accelbyte_py_sdk
from accelbyte_py_sdk.services.auth import login_client
if __name__ == "__main__":
accelbyte_py_sdk.initialize()
client_id = environ["AB_CLIENT_ID"]
client_secret = environ["AB_CLIENT_SECRET"]
token, error = login_client(client_id, client_secret)
assert error is not None
:bulb: The use of a Public OAuth Client is highly discouraged for this use case. Please ensure that you both set the Client ID and Client Secret.
Refreshing Tokens
:bulb: Using login_x(..., auto_refresh=True)
automatically refreshes the token once the expiration draws near.
res, err = login_client(client_id, client_secret, auto_refresh=True, refresh_rate=0.5)
res, err = login_user(username, password, auto_refresh=True, refresh_rate=0.5)
The auto refresh is only triggered when another request is fired. If you want to the refresh run automatically in the background. Use any of the LoginXTimer
classes.
from accelbyte_py_sdk.services.auth import LoginClientTimer, LoginPlatformTimer, LoginUserTimer, RefreshLoginTimer
res, err = login_user(username, password)
if err is not None:
exit(1)
interval = 1800
timer = LoginUserTimer(
interval,
username=username,
password=password,
repeats=-1,
autostart=True,
)
To manually refresh the token:
from accelbyte_py_sdk.core import get_token_repository
from accelbyte_py_sdk.services.auth import refresh_login
token_repo = get_token_repository()
refresh_token = token_repo.get_refresh_token()
token, error = refresh_login(refresh_token)
assert error is None
To use on-demand token refresh, enable the refresh_if_possible
option by setting it to True.
This configuration involves checking for the presence of an existing token, verifying its expiration status.
If the token has expired, the SDK will then examine whether a refresh token exists.
If a refresh token is available, it will be utilized to obtain a new token.
res, error = login_user(username, password, refresh_if_possible=True)
Using multiple SDK instances
The examples above demonstrates using just one instance of the Modular Python SDK (the default which is also global), but you could also instantiate multiple instances of the SDK and use them at the same time.
import accelbyte_py_sdk.services.auth as auth_service
import accelbyte_py_sdk.api.iam as iam_service
import accelbyte_py_sdk.api.iam.models as iam_models
from accelbyte_py_sdk import AccelByteSDK
from accelbyte_py_sdk.core import EnvironmentConfigRepository
from accelbyte_py_sdk.core import InMemoryTokenRepository
client_sdk = AccelByteSDK()
user_sdk1 = AccelByteSDK()
user_sdk2 = AccelByteSDK()
client_sdk.initialize(
options={
"config": EnvironmentConfigRepository(),
"token": InMemoryTokenRepository(),
}
)
user_sdk1.initialize(
options={
"config": EnvironmentConfigRepository(),
"token": InMemoryTokenRepository(),
}
)
user_sdk2.initialize(
options={
"config": user_sdk1.get_config_repository(),
"token": InMemoryTokenRepository(),
}
)
_, error = auth_service.login_client(sdk=client_sdk)
username1, password1 = ...
_, error = auth_service.login_user(username1, password1, sdk=user_sdk1)
username2, password2 = ...
_, error = auth_service.login_user(username2, password2, sdk=user_sdk2)
result1, error = iam_service.public_create_user_v4(
body=iam_models.AccountCreateUserRequestV4.create_from_dict({...}),
sdk=user_sdk1,
)
result2, error = iam_service.public_create_user_v4(
body=iam_models.AccountCreateUserRequestV4.create_from_dict({...}),
sdk=user_sdk2,
)
result, error = admin_update_user_v4(
body=iam_models.ModelUserUpdateRequestV3.create_from_dict({...}),
user_id=result1.user_id,
sdk=client_sdk,
)
client_sdk1.deintialize()
client_sdk1.deintialize()
client_sdk1.deintialize()
CLI
You can also use the Python Extend SDK to get tokens.
python -m accelbyte_py_sdk login -h
usage: python -m accelbyte_py_sdk login [-h] [-b BASE_URL] [-n NAMESPACE] [-c CLIENT_ID] [-s CLIENT_SECRET] [-u USERNAME] [-p PASSWORD] {client,user}
positional arguments:
{client,user}
options:
-h, --help show this help message and exit
-b BASE_URL, --base-url BASE_URL
-n NAMESPACE, --namespace NAMESPACE
-c CLIENT_ID, --client-id CLIENT_ID
-s CLIENT_SECRET, --client-secret CLIENT_SECRET
-u USERNAME, --username USERNAME
-p PASSWORD, --password PASSWORD
python -m accelbyte_py_sdk login client -b 'https://<environment>.accelbyte.io' -c **** -s **** | jq -r .access_token
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJuYW1lIjoiQWNjZWwgSm9lIn0.VOnZ863KSLHhgJfupKzuKZq4ACcefUlcQwi1YvxdBlw
AB_BASE_URL='https://<environment>.accelbyte.io' AB_CLIENT_ID=**** AB_CLIENT_SECRET=**** python -m accelbyte_py_sdk login user -u **** -p **** | jq -r .access_token
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJuYW1lIjoiQWNjZWwgSm9lIn0.VOnZ863KSLHhgJfupKzuKZq4ACcefUlcQwi1YvxdBlw
Interacting with a Service Endpoint
Example A
In this example we will create a new user using the POST
endpoint /iam/v3/public/namespaces/{namespace}/users
import json
import accelbyte_py_sdk
from accelbyte_py_sdk.services.auth import login_client
from accelbyte_py_sdk.api.iam import public_create_user_v3
from accelbyte_py_sdk.api.iam.models import ModelUserCreateRequestV3
from accelbyte_py_sdk.api.iam.models import ModelUserCreateResponseV3
def main():
accelbyte_py_sdk.initialize()
_, error = login_client()
result, error = public_create_user_v3(
body=ModelUserCreateRequestV3.create(
auth_type="EMAILPASSWD",
country="US",
date_of_birth="2001-01-01",
display_name="************",
email_address="******@fakemail.com",
password="******************",
)
)
if error:
exit(1)
print(json.dumps(result.to_dict(), indent=2))
if __name__ == "__main__":
main()
:bulb: All wrapper functions follow the return value format of result, error
.
:bulb: You could also write your own wrapper functions by using the models and operations in accelbyte_py_sdk.api.<service-name>
and accelbyte_py_sdk.api.<service-name>.models
respectively.
:bulb: All wrapper functions have an asynchronous counterpart that ends with _async
.
Example A (async
)
To convert Example A
asynchronously the following steps are needed.
-
Import the asyncio package.
import asyncio
-
Convert the main method into async
.
async def main():
-
Change how the main
function is invoked.
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
-
Use HttpxHttpClient
.
accelbyte_py_sdk.initialize(options={"http": "HttpxHttpClient"})
-
Use the async
version of the wrapper by appending _async
.
from accelbyte_py_sdk.api.iam import public_create_user_v3_async
-
Use the async
wrapper with the await
keyword.
result, error = await public_create_user_v3_async(
Here is the full code:
import asyncio
import json
import accelbyte_py_sdk
from accelbyte_py_sdk.services.auth import login_client
from accelbyte_py_sdk.api.iam import public_create_user_v3_async
from accelbyte_py_sdk.api.iam.models import ModelUserCreateRequestV3
from accelbyte_py_sdk.api.iam.models import ModelUserCreateResponseV3
async def main():
accelbyte_py_sdk.initialize(options={"http": "HttpxHttpClient"})
_, error = login_client()
result, error = await public_create_user_v3_async(
body=ModelUserCreateRequestV3.create(
auth_type="EMAILPASSWD",
country="US",
date_of_birth="2001-01-01",
display_name="************",
email_address="******@fakemail.com",
password="******************",
)
)
if error:
exit(1)
print(json.dumps(result.to_dict(), indent=2))
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
Configuring HTTP Retry
To use the HTTP Retry
feature, set the HttpClient
's retry_policy
and backoff_policy
.
import accelbyte_py_sdk
from accelbyte_py_sdk.core import get_http_client
accelbyte_py_sdk.initialize()
http_client = get_http_client()
from accelbyte_py_sdk.core import ConstantHttpBackoffPolicy
from accelbyte_py_sdk.core import MaxRetriesHttpRetryPolicy
http_client.retry_policy = MaxRetriesHttpRetryPolicy(3)
from accelbyte_py_sdk.core import ExponentialHttpBackoffPolicy
from accelbyte_py_sdk.core import MaxElapsedHttpRetryPolicy
http_client.backoff_policy = ExponentialHttpBackoffPolicy(initial=1.0, multiplier=2.0)
http_client.retry_policy = MaxElapsedHttpRetryPolicy(15)
from datetime import timedelta
from typing import Optional
def my_custom_retry_policy(request, response, /, *, retries: int = 0, elapsed: Optional[timedelta] = None, **kwargs) -> float:
return "Retry-After" in response.headers and retries == 1
def my_custom_backoff_policy(request, response, /, *, retries: int = 0, elapsed: Optional[timedelta] = None, **kwargs) -> float:
return response.headers.get("Retry-After", 1)
http_client.backoff_policy = my_custom_backoff_policy
http_client.retry_policy = my_custom_retry_policy
from accelbyte_py_sdk.core import CompositeHttpRetryPolicy
from accelbyte_py_sdk.core import MaxRetriesHttpRetryPolicy
from accelbyte_py_sdk.core import StatusCodesHttpRetryPolicy
http_client.retry_policy = CompositeHttpRetryPolicy(
StatusCodesHttpRetryPolicy(401, (501, 503)),
MaxRetriesHttpRetryPolicy(3)
)
Validating Tokens
You can use accelbyte_py_sdk.token_validation.caching.CachingTokenValidator
or accelbyte_py_sdk.token_validation.iam.IAMTokenValidator
.
token_validator = CachingTokenValidator(sdk)
error = token_validator.validate_token(access_token)
if error:
raise error
See tests for more usage.
Samples
Sample apps are available in the samples directory
Documentation
For documentation about AccelByte services and SDK, see docs.accelbyte.io
:bulb: Check out the index files in the docs directory if you are looking for a specific endpoint.
Misc
Utility Functions
Check if the SDK is initialized.
import accelbyte_py_sdk
is_initialized = accelbyte_py_sdk.is_initialized()
print(is_initialized)
Create a Basic Auth from a string tuple.
import accelbyte_py_sdk
basic_auth = accelbyte_py_sdk.core.create_basic_authentication("foo", "bar")
print(basic_auth)
Gets the stored access token.
import accelbyte_py_sdk
access_token, error = accelbyte_py_sdk.core.get_access_token()
print(access_token)
Get AB_*
environment configuration values.
import accelbyte_py_sdk
base_url, client_id, client_secret, namespace = accelbyte_py_sdk.core.get_env_config()
print(f"{base_url}, {client_id}, {client_secret}, {namespace}")
Get AB_*
environment user credential values.
import accelbyte_py_sdk
username, password = accelbyte_py_sdk.core.get_env_user_credentials()
print(f"{base_url}: {client_id}")
Set logger level and add logger handlers.
import logging
accelbyte_py_sdk.core.set_logger_level(logging.INFO)
accelbyte_py_sdk.core.set_logger_level(logging.INFO, "http")
accelbyte_py_sdk.core.set_logger_level(logging.INFO, "ws")
accelbyte_py_sdk.core.add_stream_handler_to_logger()
accelbyte_py_sdk.core.add_buffered_file_handler_to_logger(
filename="/path/to/sdk.log",
capacity=10,
level=logging.INFO
)
accelbyte_py_sdk.core.add_buffered_file_handler_to_logger(
filename="/path/to/http.log",
capacity=3,
level=logging.INFO,
additional_scope="http"
)
accelbyte_py_sdk.initialize(
options={
"log_files": {
"": "/path/to/sdk.log",
"http": {
"filename": "/path/to/http.log",
"capacity": 3,
"level": logging.INFO
}
}
}
)
def format_request_response_as_yaml(data: dict) -> str:
return f"---\n{yaml.safe_dump(data, sort_keys=False).rstrip()}\n..."
http_client = accelbyte_py_sdk.core.get_http_client()
http_client.request_log_formatter = format_request_response_as_yaml
http_client.response_log_formatter = format_request_response_as_yaml
In-depth Topics
Generated code
Models
Each definition in #/definitions/
is turned into a Model.
Example:
properties:
avatarLargeUrl:
type: string
avatarSmallUrl:
type: string
avatarUrl:
type: string
customAttributes:
additionalProperties:
type: object
type: object
dateOfBirth:
format: date
type: string
x-nullable: true
firstName:
type: string
language:
type: string
lastName:
type: string
namespace:
type: string
status:
enum:
- ACTIVE
- INACTIVE
type: string
timeZone:
type: string
userId:
type: string
zipCode:
type: string
type: object
class UserProfileInfo(Model):
"""User profile info (UserProfileInfo)
Properties:
avatar_large_url: (avatarLargeUrl) OPTIONAL str
avatar_small_url: (avatarSmallUrl) OPTIONAL str
avatar_url: (avatarUrl) OPTIONAL str
custom_attributes: (customAttributes) OPTIONAL Dict[str, Any]
date_of_birth: (dateOfBirth) OPTIONAL str
first_name: (firstName) OPTIONAL str
language: (language) OPTIONAL str
last_name: (lastName) OPTIONAL str
namespace: (namespace) OPTIONAL str
status: (status) OPTIONAL Union[str, StatusEnum]
time_zone: (timeZone) OPTIONAL str
user_id: (userId) OPTIONAL str
zip_code: (zipCode) OPTIONAL str
"""
avatar_large_url: str
avatar_small_url: str
avatar_url: str
custom_attributes: Dict[str, Any]
date_of_birth: str
first_name: str
language: str
last_name: str
namespace: str
status: Union[str, StatusEnum]
time_zone: str
user_id: str
zip_code: str
there are also a number of utility functions generated with each model that should help in the ease of use.
...
def with_user_id(self, value: str) -> UserProfileInfo:
self.user_id = value
return self
def to_dict(self, include_empty: bool = False) -> dict:
result: dict = {}
...
return result
@classmethod
def create(
cls,
avatar_large_url: Optional[str] = None,
avatar_small_url: Optional[str] = None,
avatar_url: Optional[str] = None,
custom_attributes: Optional[Dict[str, Any]] = None,
date_of_birth: Optional[str] = None,
first_name: Optional[str] = None,
language: Optional[str] = None,
last_name: Optional[str] = None,
namespace: Optional[str] = None,
status: Optional[Union[str, StatusEnum]] = None,
time_zone: Optional[str] = None,
user_id: Optional[str] = None,
zip_code: Optional[str] = None,
) -> UserProfileInfo:
instance = cls()
...
return instance
@classmethod
def create_from_dict(cls, dict_: dict, include_empty: bool = False) -> UserProfileInfo:
instance = cls()
...
return instance
@staticmethod
def get_field_info() -> Dict[str, str]:
return {
"avatarLargeUrl": "avatar_large_url",
"avatarSmallUrl": "avatar_small_url",
"avatarUrl": "avatar_url",
"customAttributes": "custom_attributes",
"dateOfBirth": "date_of_birth",
"firstName": "first_name",
"language": "language",
"lastName": "last_name",
"namespace": "namespace",
"status": "status",
"timeZone": "time_zone",
"userId": "user_id",
"zipCode": "zip_code",
}
...
Operations
Each path item in #/paths
is turned into an Operation.
Example:
description: 'Get user profile.<br>Other detail info: <ul><li><i>Required
permission</i>: resource=<b>"NAMESPACE:{namespace}:USER:{userId}:PROFILE"</b>,
action=2 <b>(READ)</b></li><li><i>Action code</i>:
11403</li><li><i>Returns</i>: user profile</li></ul>'
operationId: publicGetUserProfileInfo
parameters:
- description: namespace, only accept alphabet and numeric
in: path
name: namespace
required: true
type: string
- description: user's id, should follow UUID version 4 without hyphen
in: path
name: userId
required: true
type: string
produces:
- application/json
responses:
'200':
description: Successful operation
schema:
$ref: '#/definitions/UserProfileInfo'
'400':
description: <table><tr><td>errorCode</td><td>errorMessage</td></tr><tr><td>20002</td><td>validation
error</td></tr></table>
schema:
$ref: '#/definitions/ValidationErrorEntity'
'401':
description: <table><tr><td>errorCode</td><td>errorMessage</td></tr><tr><td>20001</td><td>unauthorized</td></tr></table>
schema:
$ref: '#/definitions/ErrorEntity'
'403':
description: <table><tr><td>errorCode</td><td>errorMessage</td></tr><tr><td>20013</td><td>insufficient
permission</td></tr></table>
schema:
$ref: '#/definitions/ErrorEntity'
'404':
description: '<table><tr><td>errorCode</td><td>errorMessage</td></tr><tr><td>11440</td><td>Unable
to {action}: User profile not found in namespace [{namespace}]</td></tr></table>'
schema:
$ref: '#/definitions/ErrorEntity'
security:
- authorization: []
- HasPermission:
- NAMESPACE:{namespace}:USER:{userId}:PROFILE [READ]
authorization: []
summary: Get user profile
tags:
- UserProfile
x-authorization:
action: '2'
resource: NAMESPACE:{namespace}:USER:{userId}:PROFILE
same with the models there are also a number of utility functions generated with each operation that should help in the ease of use.
from __future__ import annotations
from typing import Any, Dict, List, Optional, Tuple, Union
from accelbyte_py_sdk.core import Operation
from accelbyte_py_sdk.core import HeaderStr
from accelbyte_py_sdk.core import HttpResponse
from ...models import ErrorEntity
from ...models import UserProfileInfo
from ...models import ValidationErrorEntity
class PublicGetUserProfileInfo(Operation):
"""Get user profile (publicGetUserProfileInfo)
Get user profile.
Other detail info:
* Required permission : resource= "NAMESPACE:{namespace}:USER:{userId}:PROFILE" , action=2 (READ)
* Action code : 11403
* Returns : user profile
Required Permission(s):
- NAMESPACE:{namespace}:USER:{userId}:PROFILE [READ]
Properties:
url: /basic/v1/public/namespaces/{namespace}/users/{userId}/profiles
method: GET
tags: ["UserProfile"]
consumes: []
produces: ["application/json"]
securities: [BEARER_AUTH] or [BEARER_AUTH]
namespace: (namespace) REQUIRED str in path
user_id: (userId) REQUIRED str in path
Responses:
200: OK - UserProfileInfo (Successful operation)
400: Bad Request - ValidationErrorEntity (20002: validation error)
401: Unauthorized - ErrorEntity (20001: unauthorized)
403: Forbidden - ErrorEntity (20013: insufficient permission)
404: Not Found - ErrorEntity (11440: Unable to {action}: User profile not found in namespace [{namespace}])
"""
_url: str = "/basic/v1/public/namespaces/{namespace}/users/{userId}/profiles"
_method: str = "GET"
_consumes: List[str] = []
_produces: List[str] = ["application/json"]
_securities: List[List[str]] = [["BEARER_AUTH"], ["BEARER_AUTH"]]
_location_query: str = None
namespace: str
user_id: str
@property
def url(self) -> str:
return self._url
@property
def method(self) -> str:
return self._method
@property
def consumes(self) -> List[str]:
return self._consumes
@property
def produces(self) -> List[str]:
return self._produces
@property
def securities(self) -> List[List[str]]:
return self._securities
@property
def location_query(self) -> str:
return self._location_query
def get_all_params(self) -> dict:
return {
"path": self.get_path_params(),
}
def get_path_params(self) -> dict:
result = {}
if hasattr(self, "namespace"):
result["namespace"] = self.namespace
if hasattr(self, "user_id"):
result["userId"] = self.user_id
return result
def with_namespace(self, value: str) -> PublicGetUserProfileInfo:
self.namespace = value
return self
def with_user_id(self, value: str) -> PublicGetUserProfileInfo:
self.user_id = value
return self
def to_dict(self, include_empty: bool = False) -> dict:
result: dict = {}
if hasattr(self, "namespace") and self.namespace:
result["namespace"] = str(self.namespace)
elif include_empty:
result["namespace"] = ""
if hasattr(self, "user_id") and self.user_id:
result["userId"] = str(self.user_id)
elif include_empty:
result["userId"] = ""
return result
def parse_response(
self, code: int, content_type: str, content: Any
) -> Tuple[
Union[None, UserProfileInfo],
Union[None, ErrorEntity, HttpResponse, ValidationErrorEntity],
]:
"""Parse the given response.
200: OK - UserProfileInfo (Successful operation)
400: Bad Request - ValidationErrorEntity (20002: validation error)
401: Unauthorized - ErrorEntity (20001: unauthorized)
403: Forbidden - ErrorEntity (20013: insufficient permission)
404: Not Found - ErrorEntity (11440: Unable to {action}: User profile not found in namespace [{namespace}])
---: HttpResponse (Undocumented Response)
---: HttpResponse (Unexpected Content-Type Error)
---: HttpResponse (Unhandled Error)
"""
pre_processed_response, error = self.pre_process_response(
code=code, content_type=content_type, content=content
)
if error is not None:
return None, None if error.is_no_content() else error
code, content_type, content = pre_processed_response
if code == 200:
return UserProfileInfo.create_from_dict(content), None
if code == 400:
return None, ValidationErrorEntity.create_from_dict(content)
if code == 401:
return None, ErrorEntity.create_from_dict(content)
if code == 403:
return None, ErrorEntity.create_from_dict(content)
if code == 404:
return None, ErrorEntity.create_from_dict(content)
return self.handle_undocumented_response(
code=code, content_type=content_type, content=content
)
@classmethod
def create(cls, namespace: str, user_id: str, **kwargs) -> PublicGetUserProfileInfo:
instance = cls()
instance.namespace = namespace
instance.user_id = user_id
return instance
@classmethod
def create_from_dict(
cls, dict_: dict, include_empty: bool = False
) -> PublicGetUserProfileInfo:
instance = cls()
if "namespace" in dict_ and dict_["namespace"] is not None:
instance.namespace = str(dict_["namespace"])
elif include_empty:
instance.namespace = ""
if "userId" in dict_ and dict_["userId"] is not None:
instance.user_id = str(dict_["userId"])
elif include_empty:
instance.user_id = ""
return instance
@staticmethod
def get_field_info() -> Dict[str, str]:
return {
"namespace": "namespace",
"userId": "user_id",
}
@staticmethod
def get_required_map() -> Dict[str, bool]:
return {
"namespace": True,
"userId": True,
}
Creating
:bulb: there are 4 ways to create an instance of these models and operations.
model = ModelName()
model.param_a = "foo"
model.param_b = "bar"
model = ModelName() \
.with_param_a("foo") \
.with_param_b("bar")
model = ModelName.create(
param_a="foo",
param_b="bar",
)
model_params = {
"param_a": "foo",
"param_b": "bar",
"param_c": False,
"param_d": None,
"param_e": [],
"param_f": {},
}
model = ModelName.create_from_dict(model_params)
Wrappers
To improve ergonomics the code generator also generates wrappers around the operations.
The purpose of these wrappers is to automatically fill up parameters that the SDK already knows.
(e.g. namespace, client_id, access_token, etc.)
They are located at accelbyte_py_sdk.api.<service-name>.wrappers
but can be accessed like so: from accelbyte_py_sdk.api.<service-name> import <wrapper-name>
import accelbyte_py_sdk
from accelbyte_py_sdk.api.iam import token_grant_v3
if __name__ == "__main__":
accelbyte_py_sdk.initialize()
token, error = token_grant_v3(
grant_type="client_credentials"
)
assert error is not None
The wrapper function token_grant_v3
is a wrapper for the TokenGrantV3
operation.
It automatically passes in the information needed like the Basic Auth Headers.
The values are gotten from the current ConfigRepository
.
continuing from the previous examples (GetUserProfileInfo), its wrapper would be:
from typing import Any, Dict, List, Optional, Tuple, Union
from accelbyte_py_sdk.core import get_namespace as get_services_namespace
from accelbyte_py_sdk.core import run_request
from accelbyte_py_sdk.core import same_doc_as
from ..operations.user_profile import PublicGetUserProfileInfo
@same_doc_as(PublicGetUserProfileInfo)
def public_get_user_profile_info(
user_id: str,
namespace: Optional[str] = None,
x_additional_headers: Optional[Dict[str, str]] = None,
**kwargs
):
if namespace is None:
namespace, error = get_services_namespace()
if error:
return None, error
request = PublicGetUserProfileInfo.create(
user_id=user_id,
namespace=namespace,
)
return run_request(request, additional_headers=x_additional_headers, **kwargs)
this wrapper function automatically fills up the required path parameter namespace
.
now to use it only the user_id
is now required.
import accelbyte_py_sdk
from accelbyte_py_sdk.api.basic import public_get_user_profile_info
if __name__ == "__main__":
accelbyte_py_sdk.initialize()
user_profile_info, error = public_get_user_profile_info(user_id="lorem")
assert error is not None
print(f"Hello there {user_profile_info.first_name}!")