SwgohComlink API¶

SwgohComlink is the synchronous client for a running swgoh-comlink service. Each method sends one request and returns the decoded JSON response. For async/await code, use SwgohComlinkAsync, which has the same methods and constructor arguments.

Quick Start¶

from swgoh_comlink import SwgohComlink

with SwgohComlink() as comlink:
    player = comlink.get_player(allycode=245866537)
    guild = comlink.get_guild(guild_id=player["guildId"])

Without a context manager, call close() when done:

comlink = SwgohComlink()
try:
    player = comlink.get_player(allycode=245866537)
finally:
    comlink.close()

Connecting to Comlink¶

The client connects to http://localhost:3000 by default. Point it at another service with either a full url, or a host and port:

comlink = SwgohComlink(url="http://comlink.example.com:3000")
comlink = SwgohComlink(host="192.168.1.10", port=3000)

When host is given, url and stats_url are ignored.

If your comlink service requires HMAC signing, pass access_key and secret_key. When they are not passed, the client reads them from the ACCESS_KEY and SECRET_KEY environment variables. Signing is turned on only when both keys are set.

See the API Reference below for every constructor argument.

Handling Errors¶

When comlink answers with an HTTP error status, the client raises a typed HTTP error, such as SwgohComlinkRateLimitError for a rate refusal or SwgohComlinkClientError for a rejected request. All of them subclass SwgohComlinkException.

Retries and Pacing¶

By default a client makes exactly one attempt per call and raises the typed HTTP error on failure. Pass a RetryPolicy as the keyword-only retry argument to retry the failures that mean the call was not carried out:

rate refusals (SwgohComlinkRateLimitError: HTTP 429, or the game's Rate exceeded!)
HTTP 503 (SwgohComlinkUnavailableError)

Before each retry the client waits for the Retry-After header when the response has one (capped at max_retry_after), otherwise for the next value of backoff (5, 10, 20 and 40 seconds by default), plus a random extra of up to jitter times that wait (20% by default) so that calls refused together do not all retry at once. Jitter only adds time: no retry starts before the backoff or Retry-After wait. Other errors are raised at once.

A rate refusal also holds back the endpoint it came from. Until the retry wait has passed, no call to that endpoint starts on the client, whether it is a retry or a new call, so concurrent callers back off together rather than adding to the refusal. Endpoints listed in unpaced_endpoints are never held.

min_interval also spaces calls to the same endpoint. The game refuses calls with Rate exceeded! at roughly 15 calls a second per RPC on one session, so 0.4 seconds keeps a burst well under that limit.

from swgoh_comlink import RetryPolicy, SwgohComlink

comlink = SwgohComlink(retry=RetryPolicy(attempts=3, min_interval=0.4))
player = comlink.get_player(allycode=245866537)

SwgohComlinkAsync accepts the same argument; see its retries section for how waits behave with concurrent tasks.

`RetryPolicy` `dataclass` ¶

How a client retries refused calls and paces calls to the same endpoint.

Pass an instance as the retry keyword argument of SwgohComlink or SwgohComlinkAsync. Without one (the default) the clients make exactly one attempt per call and never wait.

Only two failures are retried, both of which mean the call was not carried out: rate refusals (:class:~swgoh_comlink.exceptions.SwgohComlinkRateLimitError) and HTTP 503 (:class:~swgoh_comlink.exceptions.SwgohComlinkUnavailableError). Client errors, other server errors and transport failures are raised at once.

The game meters calls per session and per RPC, and refuses with Rate exceeded! at roughly 15 calls a second to one RPC. A min_interval of 0.4 s keeps a burst to one endpoint well under that.

A rate refusal also holds back the endpoint it came from: until the retry wait has passed, no call to that endpoint starts on the client, so concurrent callers back off together instead of adding to the refusal. Each retry then waits a random extra of up to jitter times its wait, so callers refused together do not all retry at the same moment.

Attributes:

Name	Type	Description
`attempts`	`int`	Total attempts per call, including the first. `1` disables retrying while keeping pacing. [Default: 5]
`backoff`	`tuple[float, ...]`	Seconds to wait before each retry when the response has no usable `Retry-After`; the last value repeats once the schedule runs out. An empty schedule retries at once, without waiting. [Default: (5.0, 10.0, 20.0, 40.0)]
`respect_retry_after`	`bool`	Wait for the server's `Retry-After` instead of the backoff value when the header is present. [Default: True]
`max_retry_after`	`float`	Upper bound, in seconds, on a `Retry-After` wait. [Default: 60.0]
`retry_rate_limited`	`bool`	Retry rate refusals. [Default: True]
`retry_unavailable`	`bool`	Retry HTTP 503 responses. [Default: True]
`min_interval`	`float`	Minimum seconds between the starts of two calls to the same endpoint on one client, across threads or tasks. `0` disables pacing. [Default: 0.0]
`unpaced_endpoints`	`frozenset[str]`	Endpoint names (such as `"player"`) exempt from `min_interval` and from being held back after a rate refusal. [Default: empty]
`jitter`	`float`	Largest random extra added to a retry wait, as a fraction of that wait: `0.2` turns a 5 s wait into 5 to 6 s. Extra time only, so a retry never starts before the backoff or `Retry-After` wait. `0` disables it. [Default: 0.2]

Raises:

Type	Description
`SwgohComlinkValueError`	If a numeric field is negative, non-finite, or `attempts` is less than 1.
`SwgohComlinkTypeError`	If `attempts` is not an integer, a numeric field is not an int or float (`bool` included), `backoff` or `unpaced_endpoints` is a string or not iterable, or an endpoint name is not a string.

Examples:

Retry rate refusals and 503s, and space calls to one endpoint 0.4 s apart:

>>> from swgoh_comlink import SwgohComlink, RetryPolicy
>>> comlink = SwgohComlink(retry=RetryPolicy(min_interval=0.4))

`retry_delay(error, attempt)` ¶

Return how long to wait before retrying a failed attempt, or None to give up.

Parameters:

Name	Type	Description	Default
`error`	`SwgohComlinkHTTPError`	The exception raised by the attempt that failed.	required
`attempt`	`int`	The 1-based number of the attempt that failed.	required

Returns:

Type	Description
`float \| None`	Seconds to wait before the next attempt, or `None` when the error is
`float \| None`	not retryable or no attempts remain.

Examples:

>>> import httpx
>>> from swgoh_comlink.exceptions import SwgohComlinkHTTPError
>>> error = SwgohComlinkHTTPError.from_response(httpx.Response(429))
>>> RetryPolicy().retry_delay(error, attempt=1)
5.0
>>> RetryPolicy(attempts=1).retry_delay(error, attempt=1) is None
True

`jitter_extra(delay)` ¶

Return a random extra, from 0 up to jitter * delay seconds, to add to a retry wait.

Stat Calculation¶

To add stats and Galactic Power to the units in a player profile, pass the roster to StatCalc, which runs the calculation locally:

from swgoh_comlink import StatCalc, SwgohComlink

with SwgohComlink() as comlink:
    calc = StatCalc()
    player = comlink.get_player(allycode=245866537)
    calc.calc_roster_stats(player["rosterUnit"])

See the StatCalc API for full details.

API Reference¶

`SwgohComlink` ¶

Bases: SwgohComlinkBase

Synchronous interface for the swgoh-comlink service.

Instances of this class are used to query the Star Wars Galaxy of Heroes game servers for exposed endpoints via the swgoh-comlink proxy library running on the same host.

Parameters:

Name	Type	Description	Default
`url`	`str`	The URL where swgoh-comlink is running. Defaults to 'http://localhost:3000'	required
`access_key`	`str`	The HMAC public key. Default to None which indicates HMAC is not used.	required
`secret_key`	`str`	The HMAC private key. Default to None which indicates HMAC is not used.	required
`stats_url`	`str`	The url of the swgoh-stats service (if used), such as 'http://localhost:3223'	required
`host`	`str`	IP address or DNS name of server where the swgoh-comlink service is running	required
`port`	`int`	TCP port number where the swgoh-comlink service is running [Default: 3000]	required
`stats_port`	`int`	TCP port number of where the comlink-stats service is running [Default: 3223]	required
`verify_ssl`	`bool`	Whether to verify TLS certificates. [Default: True]	required
`version_cache_ttl`	`float`	Seconds to cache the game/localization versions fetched from /metadata. 0 disables caching; `math.inf` caches for the client lifetime. [Default: 3600]	required
`retry`	`RetryPolicy \| None`	Keyword-only. Retry rate refusals and HTTP 503 responses, and optionally space calls to the same endpoint, as described by a :class:`~swgoh_comlink.retry.RetryPolicy`. [Default: None, one attempt per call and no waiting]	required

Notes

If the 'host' and 'port' parameters are provided, the 'url' and 'stats_url' parameters are ignored.

Examples:

Basic usage::

comlink = SwgohComlink()
player = comlink.get_player(allycode=123456789)

As a context manager::

with SwgohComlink() as comlink:
    player = comlink.get_player(allycode=123456789)

`close()` ¶

Close the underlying HTTP client connections.

`get_unit_stats(request_payload, flags=None, language=None)` ¶

Calculate unit stats using the swgoh-stats service interface to swgoh-comlink.

Parameters:

Name	Type	Description	Default
`request_payload`	`dict[str, Any] \| list[Any]`	Dictionary or list of dictionaries containing units for which to calculate stats.	required
`flags`	`list[str] \| None`	List of strings specifying which flags to include in the request URI.	`None`
`language`	`str \| None`	String indicating the desired localized language. Default "eng_us".	`None`

Returns:

Type	Description
`list[Any] \| dict[str, Any]`	Input object with the calculated stats for the specified units included.

`get_enums(refresh=False)` ¶

Get an object containing the game data enums.

Unlike most endpoints, /enums uses a GET request.

The response is cached on the instance as enums, with the game data version it was fetched under as enums_version. Later calls return the cached response while the game data version is unchanged and fetch it again once it changes. The version is read through the instance version cache (see version_cache_ttl), so a check usually makes no request. When the version cannot be determined, the enums are fetched and returned without being cached.

Parameters:

Name	Type	Description	Default
`refresh`	`bool`	When True, fetch the game data version and the enums again even if they are cached.	`False`

Returns:

Type	Description
`dict[str, Any]`	A dictionary containing the game data enums. A cached response is the same object as `enums`:
`dict[str, Any]`	copy it before changing it.

`get_events(enums=False)` ¶

Get an object containing the events game data.

Parameters:

Name	Type	Description	Default
`enums`	`bool`	Boolean flag to indicate whether enum value should be converted in response. [Default is False]	`False`

Returns:

Type	Description
`dict[str, list[Any]]`	A single element dictionary containing a list of the events game data.

`get_game_data(version='', include_pve_units=True, request_segment=0, enums=False, items=None, device_platform='Android')` ¶

Get game data.

Parameters:

Name	Type	Description	Default
`version`	`str`	string (found in metadata key value 'latestGamedataVersion')	`''`
`include_pve_units`	`bool`	boolean [Defaults to True]	`True`
`request_segment`	`int`	integer >=0 [Defaults to 0]	`0`
`enums`	`bool`	boolean [Defaults to False]	`False`
`items`	`str \| DataItems \| int \| None`	string [Defaults to None] bitwise value indicating the collections to retrieve from game. NOTE: this parameter is mutually exclusive with request_segment.	`None`
`device_platform`	`str`	string [Defaults to "Android"]	`'Android'`

Returns:

Type	Description
`dict[str, Any]`	A dictionary containing the game data.

`get_localization(localization_id=None, locale=None, unzip=False, enums=False)` ¶

Get localization data from game.

Parameters:

Name	Type	Description	Default
`localization_id`	`str \| None`	latestLocalizationBundleVersion found in game metadata.	`None`
`locale`	`str \| None`	string Specify only a specific locale to retrieve [for example "ENG_US"]	`None`
`unzip`	`bool`	boolean [Defaults to False]	`False`
`enums`	`bool`	boolean [Defaults to False]	`False`

Returns:

Type	Description
`dict[str, Any]`	A dictionary containing the localization data.

`get_game_metadata(client_specs=None, enums=False)` ¶

Get the game metadata. Game metadata contains the current game and localization versions.

Parameters:

Name	Type	Description	Default
`client_specs`	`dict[str, Any] \| None`	Optional dictionary containing client specifications.	`None`
`enums`	`bool`	Boolean signifying whether enums in response should be translated to text. [Default: False]	`False`

Returns:

Type	Description
`dict[str, Any]`	A dictionary containing the game metadata.

`get_player(allycode=None, player_id=None, enums=False)` ¶

Get player information from game. Either allycode or player_id must be provided.

Parameters:

Name	Type	Description	Default
`allycode`	`str \| int \| None`	integer or string representing player allycode	`None`
`player_id`	`str \| None`	string representing player game ID	`None`
`enums`	`bool`	boolean [Defaults to False]	`False`

Returns:

Type	Description
`dict[str, Any]`	A dictionary containing the player information.

`get_player_arena(allycode=None, player_id=None, player_details_only=False, enums=False)` ¶

Get player arena information from game. Either allycode or player_id must be provided.

Parameters:

Name	Type	Description	Default
`allycode`	`str \| int \| None`	integer or string representing player allycode	`None`
`player_id`	`str \| None`	string representing player game ID	`None`
`player_details_only`	`bool`	filter results to only player details [Defaults to False]	`False`
`enums`	`bool`	boolean [Defaults to False]	`False`

Returns:

Type	Description
`dict[str, Any]`	A dictionary containing the player arena information.

`get_guild(guild_id, include_recent_guild_activity_info=False, enums=False)` ¶

Get guild information for a specific Guild ID.

Parameters:

Name	Type	Description	Default
`guild_id`	`str`	String ID of guild to retrieve.	required
`include_recent_guild_activity_info`	`bool`	boolean [Default: False]	`False`
`enums`	`bool`	Should enums in response be translated to text. [Default: False]	`False`

Returns:

Type	Description
`dict[str, Any]`	A dictionary containing the guild information.

`get_guilds_by_name(name, start_index=0, count=10, enums=False)` ¶

Search for guild by name and return match.

Parameters:

Name	Type	Description	Default
`name`	`str`	string for guild name search	required
`start_index`	`int`	integer for where in the result list to begin	`0`
`count`	`int`	integer for max matches to return [Default: 10]	`10`
`enums`	`bool`	Whether to translate enums in response to text [Default: False]	`False`

Returns:

Type	Description
`dict[str, Any]`	A dictionary containing the guild search results.

`get_guilds_by_criteria(search_criteria, start_index=0, count=10, enums=False)` ¶

Search for guild by guild criteria and return matches.

Parameters:

Name	Type	Description	Default
`search_criteria`	`dict[str, Any]`	Dictionary of search criteria	required
`start_index`	`int`	integer for where in the result list to begin	`0`
`count`	`int`	integer for max matches to return	`10`
`enums`	`bool`	Whether to translate enum values to text [Default: False]	`False`

Returns:

Type	Description
`dict[str, Any]`	A dictionary containing the guild search results.

`get_leaderboard(leaderboard_type, *, league=None, division=None, event_instance_id=None, group_id=None, enums=False)` ¶

Retrieve Grand Arena Championship leaderboard information.

Parameters:

Name	Type	Description	Default
`leaderboard_type`	`int`	Type 4 for GAC brackets, Type 6 for global leaderboards.	required
`league`	`int \| str \| None`	Enum value or string name for league.	`None`
`division`	`int \| str \| None`	Enum value or string name for division.	`None`
`event_instance_id`	`str \| None`	Event+instance ID (for type 4).	`None`
`group_id`	`str \| None`	Group ID (for type 4).	`None`
`enums`	`bool`	Whether to translate enum values to text [Default: False]	`False`

Returns:

Type	Description
`dict[str, Any]`	A dictionary containing the leaderboard data.

`get_guild_leaderboard(leaderboard_id, count=200, enums=False)` ¶

Fetches the guild leaderboard data for given leaderboard ID.

Parameters:

Name	Type	Description	Default
`leaderboard_id`	`list[dict[str, Any]]`	A list containing one leaderboard ID dictionary.	required
`count`	`int`	The number of leaderboard entries to retrieve. [Default: 200]	`200`
`enums`	`bool`	Whether or not to translate enum values. [Default: False]	`False`

Returns:

Type	Description
`dict[str, Any]`	A dictionary containing the guild leaderboard data.

`get_name_spaces(only_compatible=False, enums=False)` ¶

Fetches namespaces based on the specified filter criteria.

`get_segmented_content(content_name_space='current', accept_language='ENG_US', enums=False)` ¶

Retrieves segmented content from a specified namespace.

`get_latest_game_data_version(refresh=False)` ¶

Get the latest game data and language bundle versions.

Results are served from the instance version cache (see version_cache_ttl).

Parameters:

Name	Type	Description	Default
`refresh`	`bool`	When True, bypass the cache and fetch fresh values from /metadata.	`False`

Returns:

Type	Description
`dict[str, Any]`	Dictionary with 'game' and 'language' version strings.

SwgohComlink API¶

Quick Start¶

Connecting to Comlink¶

Handling Errors¶

Retries and Pacing¶

RetryPolicy dataclass ¶

retry_delay(error, attempt) ¶

jitter_extra(delay) ¶

Stat Calculation¶

API Reference¶

SwgohComlink ¶

close() ¶

get_unit_stats(request_payload, flags=None, language=None) ¶

get_enums(refresh=False) ¶

get_events(enums=False) ¶

get_game_data(version='', include_pve_units=True, request_segment=0, enums=False, items=None, device_platform='Android') ¶

get_localization(localization_id=None, locale=None, unzip=False, enums=False) ¶

get_game_metadata(client_specs=None, enums=False) ¶

get_player(allycode=None, player_id=None, enums=False) ¶

get_player_arena(allycode=None, player_id=None, player_details_only=False, enums=False) ¶

get_guild(guild_id, include_recent_guild_activity_info=False, enums=False) ¶

get_guilds_by_name(name, start_index=0, count=10, enums=False) ¶

get_guilds_by_criteria(search_criteria, start_index=0, count=10, enums=False) ¶

get_leaderboard(leaderboard_type, *, league=None, division=None, event_instance_id=None, group_id=None, enums=False) ¶

get_guild_leaderboard(leaderboard_id, count=200, enums=False) ¶

get_name_spaces(only_compatible=False, enums=False) ¶

get_segmented_content(content_name_space='current', accept_language='ENG_US', enums=False) ¶

get_latest_game_data_version(refresh=False) ¶

`RetryPolicy` `dataclass` ¶

`retry_delay(error, attempt)` ¶

`jitter_extra(delay)` ¶

`SwgohComlink` ¶

`close()` ¶

`get_unit_stats(request_payload, flags=None, language=None)` ¶

`get_enums(refresh=False)` ¶

`get_events(enums=False)` ¶

`get_game_data(version='', include_pve_units=True, request_segment=0, enums=False, items=None, device_platform='Android')` ¶

`get_localization(localization_id=None, locale=None, unzip=False, enums=False)` ¶

`get_game_metadata(client_specs=None, enums=False)` ¶

`get_player(allycode=None, player_id=None, enums=False)` ¶

`get_player_arena(allycode=None, player_id=None, player_details_only=False, enums=False)` ¶

`get_guild(guild_id, include_recent_guild_activity_info=False, enums=False)` ¶

`get_guilds_by_name(name, start_index=0, count=10, enums=False)` ¶

`get_guilds_by_criteria(search_criteria, start_index=0, count=10, enums=False)` ¶

`get_leaderboard(leaderboard_type, *, league=None, division=None, event_instance_id=None, group_id=None, enums=False)` ¶

`get_guild_leaderboard(leaderboard_id, count=200, enums=False)` ¶

`get_name_spaces(only_compatible=False, enums=False)` ¶

`get_segmented_content(content_name_space='current', accept_language='ENG_US', enums=False)` ¶

`get_latest_game_data_version(refresh=False)` ¶