From c51c9cb8ae91ff973290e31c3dfbc91028458290 Mon Sep 17 00:00:00 2001 From: Programmer-Timmy Date: Mon, 2 Mar 2026 08:45:05 +0100 Subject: [PATCH 1/3] Add async client support and SSL certificate handling - Introduced AsyncSatisfactoryAPI for asynchronous operations. - Enhanced SatisfactoryAPI to support SSL certificate pinning. - Updated README with usage examples for both sync and async clients. --- README.md | 194 ++++--- requirements.txt | 3 +- satisfactory_api_client/__init__.py | 1 + satisfactory_api_client/api_client.py | 45 +- satisfactory_api_client/async_api_client.py | 546 ++++++++++++++++++++ setup.py | 3 +- 6 files changed, 728 insertions(+), 64 deletions(-) create mode 100644 satisfactory_api_client/async_api_client.py diff --git a/README.md b/README.md index 71b66a9..ae4a94e 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,8 @@ This Python package provides a client for interacting with the Satisfactory Dedicated Server API. The client allows for managing various aspects of the server, including querying server state, logging in, adjusting game settings, handling save files, and issuing administrative commands. +Both a **synchronous** (`SatisfactoryAPI`) and an **asynchronous** (`AsyncSatisfactoryAPI`) client are provided. + ## Features - Perform health checks on the server @@ -11,6 +13,8 @@ This Python package provides a client for interacting with the Satisfactory Dedi - Create, load, save, and delete game sessions - Set client and admin passwords - Run server commands and shut down the server +- SSL certificate pinning for self-signed server certificates +- Full async support via `AsyncSatisfactoryAPI` ## Installation @@ -24,29 +28,43 @@ pip install satisfactory-api-client - Python 3.10+ - `requests` library +- `aiohttp` library (for async client) ## Usage ### Initializing the Client -The `SatisfactoryAPI` class is the main entry point for interacting with the server API. - ```python from satisfactory_api_client import SatisfactoryAPI -# Initialize the API client +# Basic initialization api = SatisfactoryAPI(host='your-server-ip') -# You can also specify a custom port +# Custom port api = SatisfactoryAPI(host='your-server-ip', port=15000) -# You can also give a token directly without login +# With an existing auth token api = SatisfactoryAPI(host='your-server-ip', auth_token='your-token') + +# Skip SSL verification (not recommended for production) +api = SatisfactoryAPI(host='your-server-ip', skip_ssl_verification=True) ``` -### Login +### SSL Certificate Pinning -You can log in to the server using passwordless or password-based methods. +Satisfactory dedicated servers use self-signed certificates. You can pin the server's certificate so that requests are verified against it instead of skipping SSL entirely: + +```python +api = SatisfactoryAPI(host='your-server-ip') + +# Fetches and saves the certificate to certs/_.pem +# All subsequent requests will be verified against it +api.init_certificate() +``` + +The certificate is saved locally and reused on future runs. If `skip_ssl_verification=True` is set, calling `init_certificate()` raises a `RuntimeError`. + +### Login ```python from satisfactory_api_client.data import MinimumPrivilegeLevel @@ -57,58 +75,51 @@ response = api.passwordless_login(MinimumPrivilegeLevel.ADMINISTRATOR) # Password login response = api.password_login(MinimumPrivilegeLevel.ADMINISTRATOR, password='your-admin-password') -# You can check if the token is valid by +# Verify the stored token response = api.verify_authentication_token() print(response.data) ``` #### Minimum Privilege Levels -The `MinimumPrivilegeLevel` enum is used to specify the type of token you want to obtain. The following levels are available: -- NOT_AUTHENTICATED -- CLIENT -- ADMINISTRATOR -- INITIAL_ADMIN -- API_TOKEN +The `MinimumPrivilegeLevel` enum specifies the type of token to obtain: +| Level | Description | +|---|---| +| `NOT_AUTHENTICATED` | No authentication required | +| `CLIENT` | Standard client access | +| `ADMINISTRATOR` | Full administrative access | +| `INITIAL_ADMIN` | Initial setup admin access | +| `API_TOKEN` | API token access | ### Health Check -To verify that the server is running and responsive, you can perform a health check. This will return the server's current state. You dont need a token to perform a health check. - ```python response = api.health_check() print(response.data) ``` -### Querying server state - -You can query the server's current state. This return information about the server and the current game session. +### Querying Server State ```python -# Get server state response = api.query_server_state() print(response.data) ``` ### Server Options -You can query the server's current options and apply new ones: - ```python # Get server options response = api.get_server_options() print(response.data) # Apply new server options -new_options = {"server_name": "New Server Name"} -response = api.apply_server_options(new_options) +from satisfactory_api_client.data import ServerOptions +response = api.apply_server_options(ServerOptions(...)) ``` ### Advanced Game Settings -Fetch and apply advanced game settings: - ```python from satisfactory_api_client.data import AdvancedGameSettings @@ -117,20 +128,16 @@ response = api.get_advanced_game_settings() print(response.data) # Apply advanced game settings -new_settings = AdvancedGameSettings(your_custom_settings) -response = api.apply_advanced_game_settings(new_settings) +response = api.apply_advanced_game_settings(AdvancedGameSettings(...)) ``` ### Managing Game Sessions -You can create, load, save, and delete game sessions. The `NewGameData` class is used to specify the parameters for creating a new game. - ```python from satisfactory_api_client.data import NewGameData # Create a new game -new_game_data = NewGameData(save_name="MyNewGame", ...) -response = api.create_new_game(new_game_data) +response = api.create_new_game(NewGameData(save_name="MyNewGame", ...)) # Load a saved game response = api.load_game("MySaveGame") @@ -140,64 +147,133 @@ response = api.save_game("MySaveGame") # Delete a save file response = api.delete_save_file("MySaveGame") -``` -### Running Commands and Managing the Server +# List all sessions (requires admin) +response = api.enumerate_sessions() +``` -You can run commands on the server or shut it down using the API. The `run_command` method is used to execute a server command. The `shutdown` method is used to shut down the server. +### Running Commands and Shutdown ```python -# Run a server command response = api.run_command("SomeCommand") - -# Shutdown the server response = api.shutdown() ``` -## Methods +--- + +## Async Client + +`AsyncSatisfactoryAPI` mirrors the full API of `SatisfactoryAPI` but uses `async`/`await` via `aiohttp`. All methods, including `init_certificate`, are async. + +### Initializing the Async Client + +```python +from satisfactory_api_client import AsyncSatisfactoryAPI + +api = AsyncSatisfactoryAPI(host='your-server-ip') + +# Skip SSL verification +api = AsyncSatisfactoryAPI(host='your-server-ip', skip_ssl_verification=True) +``` + +### SSL Certificate Pinning (async) + +```python +api = AsyncSatisfactoryAPI(host='your-server-ip') +await api.init_certificate() +``` + +### Example + +```python +import asyncio +from satisfactory_api_client import AsyncSatisfactoryAPI +from satisfactory_api_client.data import MinimumPrivilegeLevel + +async def main(): + api = AsyncSatisfactoryAPI(host='your-server-ip') + await api.init_certificate() + + await api.password_login(MinimumPrivilegeLevel.ADMINISTRATOR, password='your-password') + + state = await api.query_server_state() + print(state.data) + +asyncio.run(main()) +``` + +All methods on `AsyncSatisfactoryAPI` are `async def` and must be awaited. + +--- + +## Methods Reference ### Authentication -- `passwordless_login(minimum_privilege_level: MinimumPrivilegeLevel)`: Log in without a password to obtain a token that is automatically saved. -- `password_login(minimum_privilege_level: MinimumPrivilegeLevel, password: str)`: Log in using a password to obtain a token that is automatically saved. -- `verify_authentication_token()`: Verify that the current token is valid. +| Method | Description | +|---|---| +| `passwordless_login(minimum_privilege_level)` | Log in without a password | +| `password_login(minimum_privilege_level, password)` | Log in with a password | +| `verify_authentication_token()` | Verify the stored token is valid | -### Server Management +### SSL -- `health_check(client_custom_data: str = '')`: Perform a health check on the server. This will return the server's current state. -- `query_server_state()`: Query the server's current state. This includes information about the server and the current game session. -- `shutdown()`: Shut down the server. This will stop the server process. +| Method | Description | +|---|---| +| `init_certificate()` | Fetch and pin the server's SSL certificate | -### Game Management +### Server Management -- `create_new_game(game_data: NewGameData)`: Create a new game session. This will start a new game with the specified settings. -- `load_game(save_name: str, enable_advanced_game_settings: bool = False)`: Load a saved game. This will load a previously saved game session. -- `save_game(save_name: str)`: Save the current game session. This will save the current game state to a file. -- `delete_save_file(save_name: str)`: Delete a saved game. This will delete a previously saved game session. -- `enumerate_sessions()`: List all available game sessions. This will return a list of saved game sessions. +| Method | Description | +|---|---| +| `health_check(client_custom_data='')` | Check server health (no token required) | +| `query_server_state()` | Get the current server and session state | +| `claim_server(server_name, admin_password)` | Claim an unclaimed server | +| `rename_server(server_name)` | Rename the server | +| `run_command(command)` | Execute a console command | +| `shutdown()` | Shut down the server | ### Server Settings -- `get_server_options()`: Get current server settings. This includes the server name, description, and other options. -- `apply_server_options(options: ServerOptions)`: Apply new server settings. This will update the server options with the specified values. -- `get_advanced_game_settings()`: Get advanced game settings. This includes settings such as resource settings, enemy settings, and other advanced options. -- `apply_advanced_game_settings(settings: AdvancedGameSettings)`: Apply new advanced game settings. This will update the advanced game settings with the specified values. +| Method | Description | +|---|---| +| `get_server_options()` | Get current server options | +| `apply_server_options(options)` | Apply new server options | +| `get_advanced_game_settings()` | Get advanced game settings | +| `apply_advanced_game_settings(settings)` | Apply advanced game settings | +| `set_client_password(password)` | Set the client password | +| `set_admin_password(password, auth_token)` | Set the admin password | +| `set_auto_load_session_name(session_name)` | Set the session to auto-load on start | + +### Game Management -### Commands +| Method | Description | +|---|---| +| `create_new_game(game_data)` | Start a new game session | +| `load_game(save_name, enable_advanced_game_settings)` | Load a saved game | +| `save_game(save_name)` | Save the current game | +| `delete_save_file(save_name)` | Delete a save file | +| `delete_save_session(session_name)` | Delete all saves for a session | +| `enumerate_sessions()` | List all saved sessions (admin required) | +| `download_save_game(save_name)` | Download a save file as bytes | -- `run_command(command: str)`: Run a server command. This will execute the specified command on the server. +--- ## Error Handling -Errors returned by the API will raise an `APIError` exception, which contains the error message from the server. You can catch and handle these errors in your code. For example: +All API errors raise `APIError`: ```python +from satisfactory_api_client import APIError + try: response = api.some_method() except APIError as e: print(f"Error: {e}") ``` +--- + ## Contributing Contributions are welcome! If you find a bug or have a feature request, please create an issue on the GitHub repository. diff --git a/requirements.txt b/requirements.txt index 591807e..5876b99 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ python-dotenv~=1.0.1 -requests~=2.32.3 \ No newline at end of file +requests~=2.32.3 +aiohttp~=3.9 diff --git a/satisfactory_api_client/__init__.py b/satisfactory_api_client/__init__.py index 0c765c4..57b9d4d 100644 --- a/satisfactory_api_client/__init__.py +++ b/satisfactory_api_client/__init__.py @@ -1,5 +1,6 @@ import urllib3 from .api_client import SatisfactoryAPI +from .async_api_client import AsyncSatisfactoryAPI from .exceptions import APIError, InvalidParameterError diff --git a/satisfactory_api_client/api_client.py b/satisfactory_api_client/api_client.py index 73ebad8..c7670c8 100644 --- a/satisfactory_api_client/api_client.py +++ b/satisfactory_api_client/api_client.py @@ -1,3 +1,6 @@ +import os +import ssl + import requests from .data.advanced_game_settings import AdvancedGameSettings @@ -11,7 +14,7 @@ class SatisfactoryAPI: """ A client for the Satisfactory Dedicated Server API """ - def __init__(self, host: str, port: int = 7777, auth_token: str = None): + def __init__(self, host: str, port: int = 7777, auth_token: str = None, skip_ssl_verification: bool = False): """ Initialize the API client @@ -24,6 +27,9 @@ def __init__(self, host: str, port: int = 7777, auth_token: str = None): auth_token : str, optional The authentication token, by default None. You can use the `password_login` or `passwordless_login` methods to request a token from the server. + skip_ssl_verification : bool, optional + Disable SSL certificate verification entirely, by default False. + When True, ``init_certificate`` has no effect and all requests skip verification. Raises ------ @@ -33,11 +39,43 @@ def __init__(self, host: str, port: int = 7777, auth_token: str = None): self.host: str = host self.port: int = port self.auth_token: str | None = auth_token + self.skip_ssl_verification: bool = skip_ssl_verification + self.cert_path: str | None = None if self.auth_token: self.verify_authentication_token() - def _post(self, function, data=None, files=None): + def init_certificate(self) -> None: + """ + Fetch and cache the server's SSL certificate for verified HTTPS requests. + + Downloads the server's self-signed certificate and saves it to a local + ``certs/`` directory. Once called, all subsequent requests will verify + against this certificate instead of skipping SSL verification. + + Raises + ------ + ssl.SSLError + If the certificate cannot be retrieved from the server. + RuntimeError + If ``skip_ssl_verification`` is True. + """ + if self.skip_ssl_verification: + raise RuntimeError("Cannot initialise certificate while skip_ssl_verification is enabled.") + certs_dir = os.path.join(os.path.dirname(__file__), 'certs') + os.makedirs(certs_dir, exist_ok=True) + + cert_path = os.path.join(certs_dir, f"{self.host.replace('.', '_')}_{self.port}.pem") + + if not os.path.exists(cert_path): + pem_cert = ssl.get_server_certificate((self.host, self.port)) + with open(cert_path, 'w') as f: + f.write(pem_cert) + print(f"Certificate saved to {cert_path}") + + self.cert_path = cert_path + + """ Post a request to the API @@ -57,7 +95,8 @@ def _post(self, function, data=None, files=None): payload = {'function': function, 'data': data} if data else {'function': function} - response = requests.post(url, json=payload, headers=headers, files=files, verify=False, stream=True) + verify = False if self.skip_ssl_verification else (self.cert_path or False) + response = requests.post(url, json=payload, headers=headers, files=files, verify=verify, stream=True) if response.status_code != 200 and response.status_code != 204: raise APIError( error_code=response.json().get('errorCode'), diff --git a/satisfactory_api_client/async_api_client.py b/satisfactory_api_client/async_api_client.py new file mode 100644 index 0000000..6822bec --- /dev/null +++ b/satisfactory_api_client/async_api_client.py @@ -0,0 +1,546 @@ +import asyncio +import os +import ssl + +import aiohttp + +from .data.advanced_game_settings import AdvancedGameSettings +from .data.minimum_privilege_level import MinimumPrivilegeLevel +from .data.new_game_save import NewGameData +from .data.response import Response +from .data.server_options import ServerOptions +from .exceptions import APIError + + +class AsyncSatisfactoryAPI: + """ An async client for the Satisfactory Dedicated Server API """ + + def __init__(self, host: str, port: int = 7777, auth_token: str = None, skip_ssl_verification: bool = False): + """ + Initialize the async API client + + Parameters + ---------- + host : str + The hostname or IP address of the server + port : int, optional + The port to connect to, by default 7777 + auth_token : str, optional + The authentication token, by default None. + You can use the `password_login` or `passwordless_login` methods to request a token from the server. + skip_ssl_verification : bool, optional + Disable SSL certificate verification entirely, by default False. + When True, ``init_certificate`` has no effect and all requests skip verification. + """ + self.host: str = host + self.port: int = port + self.auth_token: str | None = auth_token + self.skip_ssl_verification: bool = skip_ssl_verification + self.cert_path: str | None = None + self._ssl_context: ssl.SSLContext | None = None + + def _get_ssl(self) -> ssl.SSLContext | bool: + if self.skip_ssl_verification: + return False + return self._ssl_context or False + + async def init_certificate(self) -> None: + """ + Fetch and cache the server's SSL certificate for verified HTTPS requests. + + Downloads the server's self-signed certificate and saves it to a local + ``certs/`` directory. Once called, all subsequent requests will verify + against this certificate instead of skipping SSL verification. + + Raises + ------ + ssl.SSLError + If the certificate cannot be retrieved from the server. + RuntimeError + If ``skip_ssl_verification`` is True. + """ + if self.skip_ssl_verification: + raise RuntimeError("Cannot initialise certificate while skip_ssl_verification is enabled.") + + certs_dir = os.path.join(os.path.dirname(__file__), 'certs') + os.makedirs(certs_dir, exist_ok=True) + + cert_path = os.path.join(certs_dir, f"{self.host.replace('.', '_')}_{self.port}.pem") + + if not os.path.exists(cert_path): + pem_cert = await asyncio.to_thread(ssl.get_server_certificate, (self.host, self.port)) + with open(cert_path, 'w') as f: + f.write(pem_cert) + print(f"Certificate saved to {cert_path}") + + self.cert_path = cert_path + ctx = ssl.create_default_context() + ctx.load_verify_locations(cert_path) + self._ssl_context = ctx + + async def _post(self, function, data=None, files=None): + """ + Post a request to the API + + :param function: The function to call + :param data: The data to send + :param files: The files to send + :return: The response data + :raises APIError: If the API returns an error + """ + url = f"https://{self.host}:{self.port}/api/v1" + headers = {'Content-Type': 'application/json'} + + if self.auth_token: + headers['Authorization'] = f'Bearer {self.auth_token}' + + payload = {'function': function, 'data': data} if data else {'function': function} + + async with aiohttp.ClientSession() as session: + async with session.post(url, json=payload, headers=headers, ssl=self._get_ssl()) as response: + if response.status not in (200, 204): + error_data = await response.json(content_type=None) + raise APIError( + error_code=error_data.get('errorCode'), + message=error_data.get('errorMessage') + ) + + if response.status == 204: + return {} + + content_type = response.headers.get('Content-Type', '') + if 'application/json' in content_type: + result = await response.json(content_type=None) + if result.get('errorCode'): + raise APIError(result.get('errorMessage')) + return result.get('data') + elif content_type == 'application/octet-stream': + return await response.read() + else: + return await response.text() + + async def health_check(self, client_custom_data: str = '') -> Response: + """ + Perform a health check on the server. + + Parameters + ---------- + client_custom_data : str + Custom data to send to the server. Defaults to an empty string. + + Returns + ------- + Response + A Response containing the data returned by the API. + + Raises + ------ + APIError + If the API returns an error + """ + response = await self._post('HealthCheck', {'ClientCustomData': client_custom_data}) + return Response(success=True, data=response) + + async def verify_authentication_token(self) -> Response: + """ + Verify the authentication token. + + Returns + ------- + Response + A Response containing the message 'Token is valid'. + + Raises + ------ + APIError + If the API returns an error or if the token is invalid. + """ + await self._post('VerifyAuthenticationToken') + return Response(success=True, data={'message': 'Token is valid'}) + + async def passwordless_login(self, minimum_privilege_level: MinimumPrivilegeLevel) -> Response: + """ + Perform a passwordless login and store the authentication token. + + Parameters + ---------- + minimum_privilege_level : MinimumPrivilegeLevel + The minimum privilege level required for the login. + + Returns + ------- + Response + A Response containing the message 'Successfully logged in, the token is now stored'. + + Raises + ------ + APIError + If the API returns an error or if the login is unsuccessful. + """ + response = await self._post('PasswordlessLogin', {'MinimumPrivilegeLevel': minimum_privilege_level.value}) + self.auth_token = response['authenticationToken'] + return Response(success=True, data={'message': 'Successfully logged in, the token is now stored'}) + + async def password_login(self, minimum_privilege_level: MinimumPrivilegeLevel, password: str) -> Response: + """ + Perform a password login and store the authentication token. + + Parameters + ---------- + minimum_privilege_level : MinimumPrivilegeLevel + The minimum privilege level required for the login. + password : str + The password associated with the account. Must not be empty. + + Returns + ------- + Response + A Response containing the message 'Successfully logged in, the token is now stored'. + + Raises + ------ + APIError + If the API returns an error or if the login is unsuccessful. + """ + response = await self._post('PasswordLogin', { + 'MinimumPrivilegeLevel': minimum_privilege_level.value, + 'Password': password + }) + self.auth_token = response['authenticationToken'] + return Response(success=True, data={'message': 'Successfully logged in, the token is now stored'}) + + async def query_server_state(self) -> Response: + """ + Query the server state. + + Returns + ------- + Response + A Response containing the server state data. + + Raises + ------ + APIError + If the API returns an error. + """ + response = await self._post('QueryServerState') + return Response(success=True, data=response) + + async def get_server_options(self) -> Response: + """ + Get the server options. + + Returns + ------- + Response + A Response containing the server options data. + + Raises + ------ + APIError + If the API returns an error. + """ + response = await self._post('GetServerOptions') + return Response(success=True, data=response) + + async def get_advanced_game_settings(self) -> Response: + """ + Fetch advanced game settings. + + Returns + ------- + Response + A Response containing the advanced game settings. + """ + response = await self._post('GetAdvancedGameSettings') + return Response(success=True, data=response) + + async def apply_advanced_game_settings(self, settings: AdvancedGameSettings) -> Response: + """ + Apply advanced game settings. + + Parameters + ---------- + settings : AdvancedGameSettings + The new advanced game settings to apply. + + Returns + ------- + Response + A Response indicating the success of the operation. + """ + await self._post('ApplyAdvancedGameSettings', {'AdvancedGameSettings': settings.to_dict()}) + return Response(success=True, data={ + 'message': 'Successfully applied advanced game settings to the server.', + 'settings': settings.to_dict() + }) + + async def claim_server(self, server_name: str, admin_password: str) -> Response: + """ + Claim the server. + + Parameters + ---------- + server_name : str + The name of the server. + admin_password : str + The administrator password. + + Returns + ------- + Response + A Response containing the server claim result. + """ + response = await self._post('ClaimServer', { + 'ServerName': server_name, + 'AdminPassword': admin_password + }) + return Response(success=True, data=response) + + async def rename_server(self, server_name: str) -> Response: + """ + Rename the server. + + Parameters + ---------- + server_name : str + The new name for the server. + + Returns + ------- + Response + A Response indicating the success of the operation. + """ + response = await self._post('RenameServer', {'ServerName': server_name}) + return Response(success=True, data=response) + + async def set_client_password(self, password: str) -> Response: + """ + Set the client password. + + Parameters + ---------- + password : str + The new client password. + + Returns + ------- + Response + A Response indicating the success of the operation. + """ + response = await self._post('SetClientPassword', {'Password': password}) + return Response(success=True, data=response) + + async def set_admin_password(self, password: str, auth_token: str) -> Response: + """ + Set the admin password. + + Parameters + ---------- + password : str + The new admin password. + auth_token : str + The authentication token. + + Returns + ------- + Response + A Response indicating the success of the operation. + """ + response = await self._post('SetAdminPassword', { + 'Password': password, + 'AuthenticationToken': auth_token + }) + return Response(success=True, data=response) + + async def set_auto_load_session_name(self, session_name: str) -> Response: + """ + Set the auto-load session name. + + Parameters + ---------- + session_name : str + The session name to auto-load. + + Returns + ------- + Response + A Response indicating the success of the operation. + """ + response = await self._post('SetAutoLoadSessionName', {'SessionName': session_name}) + return Response(success=True, data=response) + + async def run_command(self, command: str) -> Response: + """ + Run a server command. + + Parameters + ---------- + command : str + The command to run. + + Returns + ------- + Response + A Response containing the result of the command. + """ + response = await self._post('RunCommand', {'Command': command}) + return Response(success=True, data=response) + + async def shutdown(self) -> Response: + """ + Shut down the server. + + Returns + ------- + Response + A Response indicating the success of the operation. + """ + await self._post('Shutdown') + return Response(success=True, data={ + 'message': "Server is shutting down... Note: If the server is configured as a service and the restart " + "policy is set to 'always', it will restart automatically." + }) + + async def apply_server_options(self, options: ServerOptions) -> Response: + """ + Apply server options. + + Parameters + ---------- + options : ServerOptions + The server options to apply. + + Returns + ------- + Response + A Response indicating the success of the operation. + """ + await self._post('ApplyServerOptions', {'UpdatedServerOptions': options.to_dict()}) + return Response(success=True, data={ + 'message': 'Successfully applied server options to the server.', + 'options': options.to_dict() + }) + + async def create_new_game(self, game_data: NewGameData) -> Response: + """ + Create a new game. + + Parameters + ---------- + game_data : NewGameData + The data for the new game. + + Returns + ------- + Response + A Response indicating the success of the operation. + """ + response = await self._post('CreateNewGame', {'NewGameData': game_data.__dict__}) + return Response(success=True, data=response) + + async def save_game(self, save_name: str) -> Response: + """ + Save the game. + + Parameters + ---------- + save_name : str + The name of the save file. + + Returns + ------- + Response + A Response indicating the success of the save operation. + """ + response = await self._post('SaveGame', {'SaveName': save_name}) + return Response(success=True, data=response) + + async def delete_save_file(self, save_name: str) -> Response: + """ + Delete a save file. + + Parameters + ---------- + save_name : str + The name of the save file to delete. + + Returns + ------- + Response + A Response indicating the success of the operation. + """ + response = await self._post('DeleteSaveFile', {'SaveName': save_name}) + return Response(success=True, data=response) + + async def delete_save_session(self, session_name: str) -> Response: + """ + Delete a save session. + + Parameters + ---------- + session_name : str + The name of the session to delete. + + Returns + ------- + Response + A Response indicating the success of the operation. + """ + response = await self._post('DeleteSaveSession', {'SessionName': session_name}) + return Response(success=True, data=response) + + async def enumerate_sessions(self) -> Response: + """ + Enumerate available sessions. You need admin privileges to call this function. + + Returns + ------- + Response + A Response containing the available sessions. + """ + response = await self._post('EnumerateSessions') + return Response(success=True, data=response) + + async def load_game(self, save_name: str, enable_advanced_game_settings: bool = False) -> Response: + """ + Load a saved game. + + Parameters + ---------- + save_name : str + The name of the save file to load. + enable_advanced_game_settings : bool, optional + Whether to enable advanced game settings (default is False). + + Returns + ------- + Response + A Response indicating the success of the load operation. + """ + response = await self._post('LoadGame', { + 'SaveName': save_name, + 'EnableAdvancedGameSettings': enable_advanced_game_settings + }) + return Response(success=True, data=response) + + async def upload_save_game(self, save_name: str, load_save_game: bool = False, + enable_advanced_game_settings: bool = False) -> Response: + raise NotImplementedError('This method is not implemented yet') + + async def download_save_game(self, save_name: str) -> Response: + """ + Download a save game file. + + Parameters + ---------- + save_name : str + The name of the save file to download. + + Returns + ------- + Response + A Response indicating the success and the save game in bytes. + """ + response = await self._post('DownloadSaveGame', {'SaveName': save_name}) + return Response(success=True, data=response) diff --git a/setup.py b/setup.py index 21a9055..4cca9c5 100644 --- a/setup.py +++ b/setup.py @@ -12,7 +12,8 @@ packages=find_packages(exclude=['tests', 'examples']), install_requires=[ "python-dotenv~=1.0.1", - "requests~=2.32.3" + "requests~=2.32.3", + "aiohttp~=3.9", ], description='A Python Package for interacting with the Satisfactory Dedicated Server API', long_description=open(readme_path).read(), From 6fb375cd898bd6283785a9b3b4ff267014bcb851 Mon Sep 17 00:00:00 2001 From: Programmer-Timmy Date: Mon, 2 Mar 2026 08:54:20 +0100 Subject: [PATCH 2/3] Refactor _post method parameter naming and documentation Updated the parameter name from 'function' to 'func' for consistency across both synchronous and asynchronous API client methods. Enhanced the docstring to provide clearer descriptions of parameters and return types, improving code readability and maintainability. --- satisfactory_api_client/api_client.py | 27 ++++++++++++++------- satisfactory_api_client/async_api_client.py | 25 +++++++++++++------ 2 files changed, 36 insertions(+), 16 deletions(-) diff --git a/satisfactory_api_client/api_client.py b/satisfactory_api_client/api_client.py index c7670c8..624433a 100644 --- a/satisfactory_api_client/api_client.py +++ b/satisfactory_api_client/api_client.py @@ -75,17 +75,26 @@ def init_certificate(self) -> None: self.cert_path = cert_path - + def _post(self, func, data=None, files=None): """ Post a request to the API - :param function: The function to call - :param data: The data to send - :param files: The files to send - :return: The response - :rtype: dict - - :raises APIError: If the API returns an error + Parameters + ---------- + func : str + The API function to call + data : dict, optional + The data to send in the request body, by default None + files : dict, optional + The files to send in the request, by default None + Returns + ------- + dict or bytes or str + The data returned by the API, which can be a dictionary (for JSON responses), bytes (for binary responses), or a string (for plain text responses). + Raises + ------ + APIError + If the API returns an error (non-200/204 status code) or if the response contains an error message. """ url = f"https://{self.host}:{self.port}/api/v1" headers = {'Content-Type': 'application/json'} @@ -93,7 +102,7 @@ def init_certificate(self) -> None: if self.auth_token: headers['Authorization'] = f'Bearer {self.auth_token}' - payload = {'function': function, 'data': data} if data else {'function': function} + payload = {'function': func, 'data': data} if data is not None else {'function': func} verify = False if self.skip_ssl_verification else (self.cert_path or False) response = requests.post(url, json=payload, headers=headers, files=files, verify=verify, stream=True) diff --git a/satisfactory_api_client/async_api_client.py b/satisfactory_api_client/async_api_client.py index 6822bec..790925f 100644 --- a/satisfactory_api_client/async_api_client.py +++ b/satisfactory_api_client/async_api_client.py @@ -78,15 +78,26 @@ async def init_certificate(self) -> None: ctx.load_verify_locations(cert_path) self._ssl_context = ctx - async def _post(self, function, data=None, files=None): + async def _post(self, func, data=None, files=None): """ Post a request to the API - :param function: The function to call - :param data: The data to send - :param files: The files to send - :return: The response data - :raises APIError: If the API returns an error + Parameters + ---------- + func : str + The API function to call + data : dict, optional + The data to send in the request body, by default None + files : dict, optional + The files to send in the request, by default None + Returns + ------- + dict or bytes or str + The data returned by the API, which can be a dictionary (for JSON responses), bytes (for binary responses), or a string (for plain text responses). + Raises + ------ + APIError + If the API returns an error (non-200/204 status code) or if the response contains an error message. """ url = f"https://{self.host}:{self.port}/api/v1" headers = {'Content-Type': 'application/json'} @@ -94,7 +105,7 @@ async def _post(self, function, data=None, files=None): if self.auth_token: headers['Authorization'] = f'Bearer {self.auth_token}' - payload = {'function': function, 'data': data} if data else {'function': function} + payload = {'function': func, 'data': data} if data is not None else {'function': func} async with aiohttp.ClientSession() as session: async with session.post(url, json=payload, headers=headers, ssl=self._get_ssl()) as response: From 7be457a3c7311317dd1557953db03199b323f681 Mon Sep 17 00:00:00 2001 From: Programmer-Timmy Date: Mon, 2 Mar 2026 08:58:21 +0100 Subject: [PATCH 3/3] Add Python 3.13 to CI matrix Updated the CI configuration to include Python version 3.13, ensuring compatibility with the latest release. --- .github/workflows/python-package.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index 5e77f08..33334f5 100644 --- a/.github/workflows/python-package.yml +++ b/.github/workflows/python-package.yml @@ -16,7 +16,7 @@ jobs: strategy: fail-fast: false matrix: - python-version: ["3.10", "3.11", "3.12"] + python-version: ["3.10", "3.11", "3.12", "3.13"] steps: - uses: actions/checkout@v4