Refurbishment

examples/desktop_authenticator.py

@@ -4,7 +4,6 @@
 
 from steampy.guard import generate_confirmation_key, generate_one_time_code
 
-
 shared_secret = ''
 identity_secret = ''
 

examples/inventory.py

@@ -1,9 +1,10 @@
 import json
+import sys
+from pathlib import Path
 
-from steampy.client import SteamClient, InvalidCredentials
+from steampy.client import InvalidCredentials, SteamClient
 from steampy.models import GameOptions
 
-
 # Your Steam username
 username = ''
 
@@ -30,7 +31,7 @@
     steam_client.login(username, password, steam_guard_path)
 except (ValueError, InvalidCredentials):
     print('Your login credentials are invalid!')
-    exit(1)
+    sys.exit(1)
 else:
     print('Finished! Logged in into Steam')
 
@@ -56,6 +57,6 @@
 
 # Dump all the info to inventory_(app_id)_(context_id).json file
 print('Saving information...')
-with open(f'inventory_{app_id}_{context_id}.json', 'w') as file:
+with Path(f'inventory_{app_id}_{context_id}.json').open('w') as file:
     json.dump(item_amounts, file)
 print(f'Done! Saved to file: inventory_{app_id}_{context_id}.json')

examples/storehouse.py

@@ -2,7 +2,6 @@
 
 from steampy.client import SteamClient, TradeOfferState
 
-
 # Set API key
 api_key = ''
 # Set path to SteamGuard file
@@ -13,7 +12,7 @@
 password = ''
 
 
-def main():
+def main() -> None:
     print('This is the donation bot accepting items for free.')
 
     if not are_credentials_filled():

setup.py

@@ -1,26 +1,23 @@
-from setuptools import setup
-import sys
 
-if not sys.version_info[0] == 3 and sys.version_info[1] < 8:
-    sys.exit('Python < 3.8 is not supported')
+from setuptools import setup
 
 version = '1.1.2'
 
 setup(
     name='steampy',
-    packages=['steampy', 'test', 'examples', ],
+    packages=['steampy', 'test', 'examples' ],
     version=version,
     description='A Steam lib for trade automation',
     author='Michał Bukowski',
     author_email='gigibukson@gmail.com',
     license='MIT',
     url='https://github.com/bukson/steampy',
     download_url='https://github.com/bukson/steampy/tarball/' + version,
-    keywords=['steam', 'trade', ],
+    keywords=['steam', 'trade' ],
     classifiers=[],
     install_requires=[
         "requests",
         "beautifulsoup4",
-        "rsa"
+        "rsa",
     ],
 )

steampy/client.py

@@ -1,42 +1,42 @@
-import re
-import bs4
+from __future__ import annotations
+
 import json
+import re
 import urllib.parse as urlparse
-from typing import List, Union
 from decimal import Decimal
 
 import requests
 
 from steampy import guard
 from steampy.confirmation import ConfirmationExecutor
-from steampy.exceptions import SevenDaysHoldException, ApiException
-from steampy.login import LoginExecutor, InvalidCredentials
+from steampy.exceptions import ApiException, SevenDaysHoldException
+from steampy.login import InvalidCredentials, LoginExecutor
 from steampy.market import SteamMarket
-from steampy.models import Asset, TradeOfferState, SteamUrl, GameOptions
+from steampy.models import Asset, GameOptions, SteamUrl, TradeOfferState
 from steampy.utils import (
-    text_between,
-    texts_between,
-    merge_items_with_descriptions_from_inventory,
-    steam_id_to_account_id,
-    merge_items_with_descriptions_from_offers,
-    get_description_key,
-    merge_items_with_descriptions_from_offer,
     account_id_to_steam_id,
+    get_description_key,
     get_key_value_from_url,
-    ping_proxy,
     login_required,
+    merge_items_with_descriptions_from_inventory,
+    merge_items_with_descriptions_from_offer,
+    merge_items_with_descriptions_from_offers,
+    ping_proxy,
+    steam_id_to_account_id,
+    text_between,
+    texts_between,
 )
 
 
 class SteamClient:
     def __init__(
         self,
         api_key: str,
-        username: str = None,
-        password: str = None,
-        steam_guard: str = None,
-        login_cookies: dict = None,
-        proxies: dict = None,
+        username: str | None = None,
+        password: str | None = None,
+        steam_guard: str | None = None,
+        login_cookies: dict | None = None,
+        proxies: dict | None = None,
     ) -> None:
         self._api_key = api_key
         self._session = requests.Session()
@@ -62,7 +62,7 @@ def set_proxies(self, proxies: dict) -> dict:
         if not isinstance(proxies, dict):
             raise TypeError(
                 'Proxy must be a dict. Example: '
-                '\{"http": "http://login:password@host:port"\, "https": "http://login:password@host:port"\}'
+                r'\{"http": "http://login:password@host:port"\, "https": "http://login:password@host:port"\}',
             )
 
         if ping_proxy(proxies):
@@ -88,13 +88,13 @@ def get_steam_id(self) -> int:
         else:
             raise ValueError(f'Invalid steam_id: {steam_id}')
 
-    def login(self, username: str = None, password: str = None, steam_guard: str = None) -> None:
+    def login(self, username: str | None = None, password: str | None = None, steam_guard: str | None = None) -> None:
         invalid_client_credentials_is_present = None in (self.username, self._password, self.steam_guard_string)
         invalid_login_credentials_is_present = None in (username, password, steam_guard)
 
         if invalid_client_credentials_is_present and invalid_login_credentials_is_present:
             raise InvalidCredentials(
-                'You have to pass username, password and steam_guard parameters when using "login" method'
+                'You have to pass username, password and steam_guard parameters when using "login" method',
             )
 
         if invalid_client_credentials_is_present:
@@ -136,9 +136,9 @@ def is_session_alive(self) -> bool:
         return steam_login.lower() in main_page_response.text.lower()
 
     def api_call(
-        self, method: str, interface: str, api_method: str, version: str, params: dict = None
+        self, method: str, interface: str, api_method: str, version: str, params: dict | None = None,
     ) -> requests.Response:
-        url = '/'.join((SteamUrl.API_URL, interface, api_method, version))
+        url = f'{SteamUrl.API_URL}/{interface}/{api_method}/{version}'
         response = self._session.get(url, params=params) if method == 'GET' else self._session.post(url, data=params)
 
         if self.is_invalid_api_key(response):
@@ -158,9 +158,9 @@ def get_my_inventory(self, game: GameOptions, merge: bool = True, count: int = 5
 
     @login_required
     def get_partner_inventory(
-        self, partner_steam_id: str, game: GameOptions, merge: bool = True, count: int = 5000
+        self, partner_steam_id: str, game: GameOptions, merge: bool = True, count: int = 5000,
     ) -> dict:
-        url = '/'.join((SteamUrl.COMMUNITY_URL, 'inventory', partner_steam_id, game.app_id, game.context_id))
+        url = f'{SteamUrl.COMMUNITY_URL}/inventory/{partner_steam_id}/{game.app_id}/{game.context_id}'
         params = {'l': 'english', 'count': count}
 
         response_dict = self._session.get(url, params=params).json()
@@ -198,10 +198,10 @@ def _filter_non_active_offers(offers_response):
         offers_sent = offers_response['response'].get('trade_offers_sent', [])
 
         offers_response['response']['trade_offers_received'] = list(
-            filter(lambda offer: offer['trade_offer_state'] == TradeOfferState.Active, offers_received)
+            filter(lambda offer: offer['trade_offer_state'] == TradeOfferState.Active, offers_received),
         )
         offers_response['response']['trade_offers_sent'] = list(
-            filter(lambda offer: offer['trade_offer_state'] == TradeOfferState.Active, offers_sent)
+            filter(lambda offer: offer['trade_offer_state'] == TradeOfferState.Active, offers_sent),
         )
 
         return offers_response
@@ -237,14 +237,12 @@ def get_trade_history(
             'include_failed': include_failed,
             'include_total': include_total,
         }
-        response = self.api_call('GET', 'IEconService', 'GetTradeHistory', 'v1', params).json()
-        return response
+        return self.api_call('GET', 'IEconService', 'GetTradeHistory', 'v1', params).json()
 
     @login_required
     def get_trade_receipt(self, trade_id: str):
         html = self._session.get(f'https://steamcommunity.com/trade/{trade_id}/receipt').content.decode()
-        items = [json.loads(item) for item in texts_between(html, 'oItem = ', ';\r\n\toItem')]
-        return items
+        return [json.loads(item) for item in texts_between(html, 'oItem = ', ';\r\n\toItem')]
 
     @login_required
     def accept_trade_offer(self, trade_offer_id: str) -> dict:
@@ -282,23 +280,21 @@ def _fetch_trade_partner_id(self, trade_offer_id: str) -> str:
 
     def _confirm_transaction(self, trade_offer_id: str) -> dict:
         confirmation_executor = ConfirmationExecutor(
-            self.steam_guard['identity_secret'], self.steam_guard['steamid'], self._session
+            self.steam_guard['identity_secret'], self.steam_guard['steamid'], self._session,
         )
         return confirmation_executor.send_trade_allow_request(trade_offer_id)
 
     def decline_trade_offer(self, trade_offer_id: str) -> dict:
         url = f'https://steamcommunity.com/tradeoffer/{trade_offer_id}/decline'
-        response = self._session.post(url, data={'sessionid': self._get_session_id()}).json()
-        return response
+        return self._session.post(url, data={'sessionid': self._get_session_id()}).json()
 
     def cancel_trade_offer(self, trade_offer_id: str) -> dict:
         url = f'https://steamcommunity.com/tradeoffer/{trade_offer_id}/cancel'
-        response = self._session.post(url, data={'sessionid': self._get_session_id()}).json()
-        return response
+        return self._session.post(url, data={'sessionid': self._get_session_id()}).json()
 
     @login_required
     def make_offer(
-        self, items_from_me: List[Asset], items_from_them: List[Asset], partner_steam_id: str, message: str = ''
+        self, items_from_me: list[Asset], items_from_them: list[Asset], partner_steam_id: str, message: str = '',
     ) -> dict:
         offer = self._create_offer_dict(items_from_me, items_from_them)
         session_id = self._get_session_id()
@@ -338,7 +334,7 @@ def get_friend_list(self, steam_id: str, relationship_filter: str = 'all') -> di
         return data['friendslist']['friends']
 
     @staticmethod
-    def _create_offer_dict(items_from_me: List[Asset], items_from_them: List[Asset]) -> dict:
+    def _create_offer_dict(items_from_me: list[Asset], items_from_them: list[Asset]) -> dict:
         return {
             'newversion': True,
             'version': 4,
@@ -362,8 +358,8 @@ def get_escrow_duration(self, trade_offer_url: str) -> int:
     @login_required
     def make_offer_with_url(
         self,
-        items_from_me: List[Asset],
-        items_from_them: List[Asset],
+        items_from_me: list[Asset],
+        items_from_them: list[Asset],
         trade_offer_url: str,
         message: str = '',
         case_sensitive: bool = True,
@@ -403,7 +399,7 @@ def _get_trade_offer_url(trade_offer_id: str) -> str:
 
     @login_required
     # If convert_to_decimal = False, the price will be returned WITHOUT a decimal point.
-    def get_wallet_balance(self, convert_to_decimal: bool = True, on_hold: bool = False) -> Union[str, Decimal]:
+    def get_wallet_balance(self, convert_to_decimal: bool = True, on_hold: bool = False) -> str | Decimal:
         response = self._session.get(f'{SteamUrl.COMMUNITY_URL}/market')
         wallet_info_match = re.search(r'var g_rgWalletInfo = (.*?);', response.text)
         if wallet_info_match:
@@ -414,5 +410,4 @@ def get_wallet_balance(self, convert_to_decimal: bool = True, on_hold: bool = Fa
         balance_dict_key = 'wallet_delayed_balance' if on_hold else 'wallet_balance'
         if convert_to_decimal:
             return Decimal(balance_dict[balance_dict_key]) / 100
-        else:
-            return balance_dict[balance_dict_key]
+        return balance_dict[balance_dict_key]

steampy/confirmation.py

@@ -1,19 +1,23 @@
+from __future__ import annotations
+
 import enum
 import json
 import time
-from typing import List
 from http import HTTPStatus
+from typing import TYPE_CHECKING
 
-import requests
 from bs4 import BeautifulSoup
 
 from steampy import guard
 from steampy.exceptions import ConfirmationExpected
 from steampy.login import InvalidCredentials
 
+if TYPE_CHECKING:
+    import requests
+
 
 class Confirmation:
-    def __init__(self, data_confid, nonce):
+    def __init__(self, data_confid, nonce) -> None:
         self.data_confid = data_confid
         self.nonce = nonce
 
@@ -52,7 +56,7 @@ def _send_confirmation(self, confirmation: Confirmation) -> dict:
         headers = {'X-Requested-With': 'XMLHttpRequest'}
         return self._session.get(f'{self.CONF_URL}/ajaxop', params=params, headers=headers).json()
 
-    def _get_confirmations(self) -> List[Confirmation]:
+    def _get_confirmations(self) -> list[Confirmation]:
         confirmations = []
         confirmations_page = self._fetch_confirmations_page()
         if confirmations_page.status_code == HTTPStatus.OK:
@@ -93,15 +97,15 @@ def _create_confirmation_params(self, tag_string: str) -> dict:
             'tag': tag_string,
         }
 
-    def _select_trade_offer_confirmation(self, confirmations: List[Confirmation], trade_offer_id: str) -> Confirmation:
+    def _select_trade_offer_confirmation(self, confirmations: list[Confirmation], trade_offer_id: str) -> Confirmation:
         for confirmation in confirmations:
             confirmation_details_page = self._fetch_confirmation_details_page(confirmation)
             confirmation_id = self._get_confirmation_trade_offer_id(confirmation_details_page)
             if confirmation_id == trade_offer_id:
                 return confirmation
         raise ConfirmationExpected
 
-    def _select_sell_listing_confirmation(self, confirmations: List[Confirmation], asset_id: str) -> Confirmation:
+    def _select_sell_listing_confirmation(self, confirmations: list[Confirmation], asset_id: str) -> Confirmation:
         for confirmation in confirmations:
             confirmation_details_page = self._fetch_confirmation_details_page(confirmation)
             confirmation_id = self._get_confirmation_sell_listing_id(confirmation_details_page)

steampy/guard.py

@@ -1,14 +1,15 @@
-import os
+from __future__ import annotations
+
 import hmac
 import json
 import struct
-from time import time
-from typing import Dict
+from base64 import b64decode, b64encode
 from hashlib import sha1
-from base64 import b64encode, b64decode
+from pathlib import Path
+from time import time
 
 
-def load_steam_guard(steam_guard: str) -> Dict[str, str]:
+def load_steam_guard(steam_guard: str) -> dict[str, str]:
     """Load Steam Guard credentials from json (file or string).
 
     Arguments:
@@ -17,14 +18,14 @@ def load_steam_guard(steam_guard: str) -> Dict[str, str]:
     Returns:
         Dict[str, str]: Parsed json data as a dictionary of strings (both key and value).
     """
-    if os.path.isfile(steam_guard):
-        with open(steam_guard, 'r') as f:
+    if Path(steam_guard).is_file():
+        with Path(steam_guard).open() as f:
             return json.loads(f.read(), parse_int=str)
     else:
         return json.loads(steam_guard, parse_int=str)
 
 
-def generate_one_time_code(shared_secret: str, timestamp: int = None) -> str:
+def generate_one_time_code(shared_secret: str, timestamp: int | None = None) -> str:
     if timestamp is None:
         timestamp = int(time())
     time_buffer = struct.pack('>Q', timestamp // 30)  # pack as Big endian, uint64