mirror/galaxy-integrations-python-api
1
0
Fork 0
mirror of https://github.com/gogcom/galaxy-integrations-python-api.git synced 2026-06-12 07:44:35 -04:00
Code Issues Packages Projects Releases Wiki Activity

SDK-3069: add OS compatibility import

Browse Source
This commit is contained in:
Aleksej Pawlowskij
2019-10-02 15:41:16 +02:00
parent 2e2aa8c4a0
commit 98cff9cfb8
7 changed files with 314 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/galaxy/api/consts.py
View File

@@ -111,6 +111,7 @@ class Feature(Enum):
ShutdownPlatformClient = "ShutdownPlatformClient"
LaunchPlatformClient = "LaunchPlatformClient"
ImportGameLibrarySettings = "ImportGameLibrarySettings"
ImportOSCompatibility = "ImportOSCompatibility"
class LicenseType(Enum):
@@ -129,3 +130,12 @@ class LocalGameState(Flag):
None_ = 0
Installed = 1
Running = 2
class OSCompatibility(Flag):
"""Possible game OS compatibility.
Use "bitwise or" to express multiple OSs compatibility, e.g. ``os=OSCompatibility.Windows|OSCompatibility.MacOS``
"""
Windows = 0b001
MacOS = 0b010
Linux = 0b100

92
src/galaxy/api/plugin.py
View File

@@ -7,12 +7,13 @@ import sys
from enum import Enum
from typing import Any, Dict, List, Optional, Set, Union
from galaxy.api.consts import Feature
from galaxy.api.consts import Feature, OSCompatibility
from galaxy.api.errors import ImportInProgress, UnknownError
from galaxy.api.jsonrpc import ApplicationError, NotificationClient, Server
from galaxy.api.types import Achievement, Authentication, FriendInfo, Game, GameTime, LocalGame, NextStep, GameLibrarySettings
from galaxy.task_manager import TaskManager
class JSONEncoder(json.JSONEncoder):
def default(self, o): # pylint: disable=method-hidden
if dataclasses.is_dataclass(o):
@@ -47,6 +48,7 @@ class Plugin:
self._achievements_import_in_progress = False
self._game_times_import_in_progress = False
self._game_library_settings_import_in_progress = False
self._os_compatibility_import_in_progress = False
self._persistent_cache = dict()
@@ -113,6 +115,9 @@ class Plugin:
self._register_method("start_game_library_settings_import", self._start_game_library_settings_import)
self._detect_feature(Feature.ImportGameLibrarySettings, ["get_game_library_settings"])
self._register_method("start_os_compatibility_import", self._start_os_compatibility_import)
self._detect_feature(Feature.ImportOSCompatibility, ["get_os_compatibility"])
async def __aenter__(self):
return self
@@ -173,6 +178,7 @@ class Plugin:
def _wrap_external_method(self, handler, name: str):
async def wrapper(*args, **kwargs):
return await self._external_task_manager.create_task(handler(*args, **kwargs), name, False)
return wrapper
async def run(self):
@@ -420,6 +426,27 @@ class Plugin:
def _game_library_settings_import_finished(self) -> None:
self._notification_client.notify("game_library_settings_import_finished", None)
def _os_compatibility_import_success(self, game_id: str, os_compatibility: Optional[OSCompatibility]) -> None:
self._notification_client.notify(
"os_compatibility_import_success",
{
"game_id": game_id,
"os_compatibility": os_compatibility
}
)
def _os_compatibility_import_failure(self, game_id: str, error: ApplicationError) -> None:
self._notification_client.notify(
"os_compatibility_import_failure",
{
"game_id": game_id,
"error": error.json()
}
)
def _os_compatibility_import_finished(self) -> None:
self._notification_client.notify("os_compatibility_import_finished", None)
def lost_authentication(self) -> None:
"""Notify the client that integration has lost authentication for the
current user and is unable to perform actions which would require it.
@@ -805,7 +832,7 @@ class Plugin:
This allows for optimizations like batch requests to platform API.
Default implementation returns None.
:param game_ids: the ids of the games for which game time are imported
:param game_ids: the ids of the games for which game library settings are imported
:return: context
"""
return None
@@ -815,17 +842,74 @@ class Plugin:
identified by the provided game_id.
This method is called by import task initialized by GOG Galaxy Client.
:param game_id: the id of the game for which the game time is returned
:param game_id: the id of the game for which the game library settings are imported
:param context: the value returned from :meth:`prepare_game_library_settings_context`
:return: GameLibrarySettings object
"""
raise NotImplementedError()
def game_library_settings_import_complete(self) -> None:
"""Override this method to handle operations after game times import is finished
"""Override this method to handle operations after game library settings import is finished
(like updating cache).
"""
async def _start_os_compatibility_import(self, game_ids: List[str]) -> None:
if self._os_compatibility_import_in_progress:
raise ImportInProgress()
context = await self.prepare_os_compatibility_context(game_ids)
async def import_os_compatibility(game_id, context_):
try:
os_compatibility = await self.get_os_compatibility(game_id, context_)
self._os_compatibility_import_success(game_id, os_compatibility)
except ApplicationError as error:
self._os_compatibility_import_failure(game_id, error)
except Exception:
logging.exception("Unexpected exception raised in import_os_compatibility")
self._os_compatibility_import_failure(game_id, UnknownError())
async def import_os_compatibility_set(game_ids_, context_):
try:
await asyncio.gather(*[
import_os_compatibility(game_id, context_) for game_id in game_ids_
])
finally:
self._os_compatibility_import_finished()
self._os_compatibility_import_in_progress = False
self.os_compatibility_import_complete()
self._external_task_manager.create_task(
import_os_compatibility_set(game_ids, context),
"game OS compatibility import",
handle_exceptions=False
)
self._os_compatibility_import_in_progress = True
async def prepare_os_compatibility_context(self, game_ids: List[str]) -> Any:
"""Override this method to prepare context for get_os_compatibility.
This allows for optimizations like batch requests to platform API.
Default implementation returns None.
:param game_ids: the ids of the games for which game os compatibility is imported
:return: context
"""
return None
async def get_os_compatibility(self, game_id: str, context: Any) -> Optional[OSCompatibility]:
"""Override this method to return the OS compatibility for the game with the provided game_id.
This method is called by import task initialized by GOG Galaxy Client.
:param game_id: the id of the game for which the game os compatibility is imported
:param context: the value returned from :meth:`prepare_os_compatibility_context`
:return: OSCompatibility flags indicating compatible OSs, or None if compatibility is not know
"""
raise NotImplementedError()
def os_compatibility_import_complete(self) -> None:
"""Override this method to handle operations after OS compatibility import is finished (like updating cache)."""
def create_and_run_plugin(plugin_class, argv):
"""Call this method as an entry point for the implemented integration.

35
src/galaxy/api/types.py
View File

@@ -1,10 +1,11 @@
from dataclasses import dataclass
from typing import List, Dict, Optional
from typing import Dict, List, Optional
from galaxy.api.consts import LicenseType, LocalGameState
@dataclass
class Authentication():
class Authentication:
"""Return this from :meth:`.authenticate` or :meth:`.pass_login_credentials`
to inform the client that authentication has successfully finished.
@@ -14,8 +15,9 @@ class Authentication():
user_id: str
user_name: str
@dataclass
class Cookie():
class Cookie:
"""Cookie
:param name: name of the cookie
@@ -28,8 +30,9 @@ class Cookie():
domain: Optional[str] = None
path: Optional[str] = None
@dataclass
class NextStep():
class NextStep:
"""Return this from :meth:`.authenticate` or :meth:`.pass_login_credentials` to open client built-in browser with given url.
For example:
@@ -67,8 +70,9 @@ class NextStep():
cookies: Optional[List[Cookie]] = None
js: Optional[Dict[str, List[str]]] = None
@dataclass
class LicenseInfo():
class LicenseInfo:
"""Information about the license of related product.
:param license_type: type of license
@@ -77,8 +81,9 @@ class LicenseInfo():
license_type: LicenseType
owner: Optional[str] = None
@dataclass
class Dlc():
class Dlc:
"""Downloadable content object.
:param dlc_id: id of the dlc
@@ -89,8 +94,9 @@ class Dlc():
dlc_title: str
license_info: LicenseInfo
@dataclass
class Game():
class Game:
"""Game object.
:param game_id: unique identifier of the game, this will be passed as parameter for methods such as launch_game
@@ -103,8 +109,9 @@ class Game():
dlcs: Optional[List[Dlc]]
license_info: LicenseInfo
@dataclass
class Achievement():
class Achievement:
"""Achievement, has to be initialized with either id or name.
:param unlock_time: unlock time of the achievement
@@ -119,8 +126,9 @@ class Achievement():
assert self.achievement_id or self.achievement_name, \
"One of achievement_id or achievement_name is required"
@dataclass
class LocalGame():
class LocalGame:
"""Game locally present on the authenticated user's computer.
:param game_id: id of the game
@@ -129,8 +137,9 @@ class LocalGame():
game_id: str
local_game_state: LocalGameState
@dataclass
class FriendInfo():
class FriendInfo:
"""Information about a friend of the currently authenticated user.
:param user_id: id of the user
@@ -139,8 +148,9 @@ class FriendInfo():
user_id: str
user_name: str
@dataclass
class GameTime():
class GameTime:
"""Game time of a game, defines the total time spent in the game
and the last time the game was played.
@@ -152,8 +162,9 @@ class GameTime():
time_played: Optional[int]
last_played_time: Optional[int]
@dataclass
class GameLibrarySettings():
class GameLibrarySettings:
"""Library settings of a game, defines assigned tags and visibility flag.
:param game_id: id of the related game