utils

Utility package with various useful functions

steam.utils.ip4_from_int(ip)

Convert int to IPv4 string

Parameters:: ip (int) – int representing an IPv4
Returns:: IP in dot-decimal notation
Return type:: str

steam.utils.ip4_to_int(ip)

Convert IPv4 string to int

Parameters:: ip (str) – IPv4 in dot-decimal notation
Return type:: int

steam.utils.ip_to_int(ip)

Convert IPv4 string to int

Parameters:: ip (str) – IPv4 in dot-decimal notation
Return type:: int

steam.utils.ip_from_int(ip)

Convert int to IPv4 string

Parameters:: ip (int) – int representing an IPv4
Returns:: IP in dot-decimal notation
Return type:: str

steam.utils.ip6_from_bytes(ip)

Convert bytes to IPv6 string

Parameters:: ip (bytes) – IPv6 in dot-decimal notation
Return type:: str

steam.utils.ip6_to_bytes(ip)

Convert IPv6 string to bytes

Parameters:: ip (str) – IPv6 in dot-decimal notation
Return type:: bytes

steam.utils.chunks(arr, size)

Splits a list into chunks

Parameters:

arr (list) – list to split
size (int) – number of elements in each chunk

Returns:

generator object

Return type:

generator

class steam.utils.WeakRefKeyDict

Bases: object

Pretends to be a dictionary. Use any object (even unhashable) as key and store a value. Once the object is garbage collected, the entry is destroyed automatically.

class steam.utils.WeakRefCallback(refs, key): Bases: object

utils.appache

Appache file parsing examples:

>>> from steam.utils.appcache import parse_appinfo, parse_packageinfo

>>> header, apps = parse_appinfo(open('/d/Steam/appcache/appinfo.vdf', 'rb'))
>>> header
{'magic': b"(DV\x07", 'universe': 1}
>>> next(apps)
{'appid': 5,
 'size': 79,
 'info_state': 1,
 'last_updated': 1484735377,
 'access_token': 0,
 'sha1': b'\x87\xfaCg\x85\x80\r\xb4\x90Im\xdc}\xb4\x81\xeeQ\x8b\x825',
 'change_number': 4603827,
 'data_sha1': b'\x87\xfaCg\x85\x80\r\xb4\x90Im\xdc}\xb4\x81\xeeQ\x8b\x825',
 'data': {'appinfo': {'appid': 5, 'public_only': 1}}}

>>> header, pkgs = parse_packageinfo(open('/d/Steam/appcache/packageinfo.vdf', 'rb'))
>>> header
{'magic': b"'UV\x06", 'universe': 1}

>>> next(pkgs)
{'packageid': 7,
 'sha1': b's\x8b\xf7n\t\xe5 k#\xb6-\x82\xd2 \x14k@\xfeDQ',
 'change_number': 7469765,
 'data': {'7': {'packageid': 7,
   'billingtype': 1,
   'licensetype': 1,
   'status': 0,
   'extended': {'requirespreapproval': 'WithRedFlag'},
   'appids': {'0': 10, '1': 80, '2': 100, '3': 254430},
   'depotids': {'0': 0, '1': 95, '2': 101, '3': 102, '4': 103, '5': 254431},
   'appitems': {}}}}

steam.utils.appcache.parse_appinfo(fp)

Parse appinfo.vdf from the Steam appcache folder

Parameters:: fp – file-like object
Raises:: SyntaxError
Return type:: (dict, Generator)
Returns:: (header, apps iterator)

steam.utils.appcache.parse_packageinfo(fp)

Parse packageinfo.vdf from the Steam appcache folder

Parameters:: fp – file-like object
Raises:: SyntaxError
Return type:: (dict, Generator)
Returns:: (header, packages iterator)

utils.binary

class steam.utils.binary.StructReader(data)

Bases: object

Simplifies parsing of struct data from bytes

Parameters:: data (bytes) – data bytes

rlen()

Number of remaining bytes that can be read

Returns:: number of remaining bytes
Return type:: int

read(n=1)

Return n bytes

Parameters:: n (int) – number of bytes to return
Returns:: bytes
Return type:: bytes

read_cstring(terminator=b'\x00')

Reads a single null termianted string

Returns:: string without bytes
Return type:: bytes

unpack(format_text)

Unpack bytes using struct modules format

Parameters:: format_text (str) – struct’s module format
Return data:: result from struct.unpack_from()
Return type:: tuple

skip(n)

Skips the next n bytes

Parameters:: n (int) – number of bytes to skip

utils.proto

utils.throttle

class steam.utils.throttle.ConstantRateLimit(times, seconds, exit_wait=False, sleep_func=<built-in function sleep>)

Bases: object

Context manager for enforcing constant rate on code inside the block .

rate = seconds / times

Parameters:

times (int) – times to execute per…
seconds (int) – …seconds
exit_wait (bool) – whether to automatically call wait() before exiting the block
sleep_func (bool) – Sleep function in seconds. Default: time.sleep()

Example:

with RateLimiter(1, 5) as r:
    # code taking 1s
    r.wait()  # blocks for 4s
    # code taking 7s
    r.wait()  # doesn't block
    # code taking 1s
    r.wait()  # blocks for 4s

wait(): Blocks until the rate is met

utils.web

steam.utils.web.make_requests_session(headers: dict = None) → Session

Returns:: requests session
Return type:: requests.Session

steam.utils.web.generate_session_id()

Returns:: session id
Return type:: str