API Reference 

apply(remote_path, local_dir, local_filename)

Apply the naming changes

Parameters:

remote_path (str) – the original remote path
local_dir (str) – the local directory the file should be written to
local_filename (str) – the local filename the file should be written to

Returns:

the local_dir and local_filename after applying the desired changes

Return type:

class aioslsk.naming.DefaultNamingStrategy

The default naming strategy determines the filename using the remote_path parameter. The local_filename parameter is ignored

apply(remote_path, local_dir, local_filename)

Apply the naming changes

Parameters:

remote_path (str) – the original remote path
local_dir (str) – the local directory the file should be written to
local_filename (str) – the local filename the file should be written to

Returns:

the local_dir and local_filename after applying the desired changes

Return type:

class aioslsk.naming.KeepDirectoryStrategy

Keeps the original directory the remote file was in

apply(remote_path, local_dir, local_filename)

Apply the naming changes

Parameters:

remote_path (str) – the original remote path
local_dir (str) – the local directory the file should be written to
local_filename (str) – the local filename the file should be written to

Returns:

the local_dir and local_filename after applying the desired changes

Return type:

class aioslsk.naming.DuplicateNamingStrategy

should_be_applied(local_dir, local_filename)

Parameters:

local_dir (str)
local_filename (str)

Return type:

class aioslsk.naming.NumberDuplicateStrategy

Duplicate name strategy that appends a number to the filename in case it already exists

PATTERN = ' \\((\\d+)\\)'

apply(remote_path, local_dir, local_filename)

Apply the naming changes

Parameters:

remote_path (str) – the original remote path
local_dir (str) – the local directory the file should be written to
local_filename (str) – the local filename the file should be written to

Returns:

the local_dir and local_filename after applying the desired changes

Return type:

aioslsk.naming.chain_strategies(strategies, remote_path, local_dir)

Chains strategies together to find the target location and filename to which the file should be written.

Parameters:

strategies (list[NamingStrategy]) – list of strategies to apply
remote_path (str) – the full remote path
local_dir (str) – initial local path, this should be the initial download directory

Returns:

tuple with the path and filename

Return type:

aioslsk.exceptions

exception aioslsk.exceptions.AioSlskException

exception aioslsk.exceptions.AuthenticationError(reason, message)

Parameters:

reason (str)
message (str)

exception aioslsk.exceptions.InvalidSessionError: Raised when trying to execute a command but the user is not logged in

exception aioslsk.exceptions.NoSuchUserError

exception aioslsk.exceptions.UnknownMessageError(message_id, data, message)

Parameters:

message_id (int)
data (bytes)
message (str)

exception aioslsk.exceptions.MessageSerializationError

exception aioslsk.exceptions.MessageDeserializationError(proto_message, message)

Parameters:

proto_message (bytes)
message (str)

Return type:

None

exception aioslsk.exceptions.FileError

exception aioslsk.exceptions.FileNotFoundError

exception aioslsk.exceptions.FileNotSharedError

exception aioslsk.exceptions.SharedDirectoryError

exception aioslsk.exceptions.TransferException

exception aioslsk.exceptions.TransferNotFoundError

exception aioslsk.exceptions.InvalidStateTransition(transfer, current, desired, message='could not make the desired state transition')

Parameters:

transfer (Transfer)
current (TransferState.State)
desired (TransferState.State)
message (str)

exception aioslsk.exceptions.RequestPlaceFailedError

exception aioslsk.exceptions.NetworkError

exception aioslsk.exceptions.PeerConnectionError

exception aioslsk.exceptions.ConnectionFailedError

exception aioslsk.exceptions.ListeningConnectionFailedError

exception aioslsk.exceptions.ConnectionReadError

exception aioslsk.exceptions.ConnectionWriteError

aioslsk.settings

aioslsk.settings.translate_blocked_users(value)

Converts blocked user set to a dictionary object

Parameters:: value (Any)
Return type:: dict[str, BlockingFlag]

class aioslsk.settings.SharedDirectorySettingEntry(*, path, share_mode=DirectoryShareMode.EVERYONE, users=<factory>)

Parameters:

path (str)
share_mode (DirectoryShareMode)
users (list[str])

path: str

share_mode: DirectoryShareMode

users: list[str]

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.WishlistSettingEntry(*, query, enabled=True)

Parameters:

query (str)
enabled (bool)

query: str

enabled: bool

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.UpnpSettings(*, enabled=True, lease_duration=21600, check_interval=600, search_timeout=10)

Parameters:

enabled (bool)
lease_duration (int)
check_interval (int)
search_timeout (int)

enabled: bool

lease_duration: int

check_interval: int

search_timeout: int

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.ReconnectSettings(*, auto=False, timeout=10)

Parameters:

auto (bool)
timeout (int)

auto: bool

timeout: int

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.ServerSettings(*, hostname='server.slsknet.org', port=2416, reconnect=ReconnectSettings(auto=False, timeout=10))

Parameters:

hostname (str)
port (int)
reconnect (ReconnectSettings)

hostname: str

port: int

reconnect: ReconnectSettings

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.ListeningSettings(*, error_mode=ListeningConnectionErrorMode.CLEAR, port=60000, obfuscated_port=60001)

Parameters:

error_mode (ListeningConnectionErrorMode)
port (int)
obfuscated_port (int)

error_mode: ListeningConnectionErrorMode

port: int

obfuscated_port: int

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.PeerSettings(*, obfuscate=False, connect_mode=PeerConnectMode.RACE)

Parameters:

obfuscate (bool)
connect_mode (PeerConnectMode)

obfuscate: bool

connect_mode: PeerConnectMode

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.NetworkLimitSettings(*, upload_speed_kbps=0, download_speed_kbps=0)

Parameters:

upload_speed_kbps (int)
download_speed_kbps (int)

upload_speed_kbps: int

download_speed_kbps: int

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.NetworkSettings(*, server=<factory>, listening=<factory>, peer=<factory>, upnp=<factory>, limits=<factory>)

Parameters:

server (ServerSettings)
listening (ListeningSettings)
peer (PeerSettings)
upnp (UpnpSettings)
limits (NetworkLimitSettings)

server: ServerSettings

listening: ListeningSettings

peer: PeerSettings

upnp: UpnpSettings

limits: NetworkLimitSettings

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.UserInfoSettings(*, description=None, picture=None)

Parameters:

description (str | None)
picture (bytes | None)

description: str | None

picture: bytes | None

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.CredentialsSettings(*, username, password, info=<factory>)

Parameters:

username (str)
password (str)
info (UserInfoSettings)

username: str

password: str

info: UserInfoSettings

are_configured()

Returns whether the credentials are correctly configured

Return type:: bool

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.SearchSendSettings(*, store_results=True, request_timeout=0, wishlist_request_timeout=-1)

Parameters:

store_results (bool)
request_timeout (int)
wishlist_request_timeout (int)

store_results: bool

request_timeout: int

wishlist_request_timeout: int

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.SearchReceiveSettings(*, store_amount=500, max_results=100)

Parameters:

store_amount (int)
max_results (int)

store_amount: int

max_results: int

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.SearchSettings(*, send=<factory>, receive=<factory>, wishlist=<factory>)

Parameters:

send (SearchSendSettings)
receive (SearchReceiveSettings)
wishlist (list[WishlistSettingEntry])

send: SearchSendSettings

receive: SearchReceiveSettings

wishlist: list[WishlistSettingEntry]

classmethod handle_max_result_alias(values)

Parameters:: values (dict)

property max_results: int

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.TransferLimitSettings(*, upload_slots=2)

Parameters:: upload_slots (int)

upload_slots: int

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.TransfersSettings(*, limits=<factory>, report_interval=0.25)

Parameters:

limits (TransferLimitSettings)
report_interval (float)

limits: TransferLimitSettings

report_interval: float

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.SharesSettings(*, scan_on_start=True, download='/home/docs/checkouts/readthedocs.org/user_builds/aioslsk/checkouts/stable/docs/source', directories=<factory>)

Parameters:

scan_on_start (bool)
download (str)
directories (list[SharedDirectorySettingEntry])

scan_on_start: bool

download: str

directories: list[SharedDirectorySettingEntry]

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.RoomsSettings(*, auto_join=True, private_room_invites=True, favorites=<factory>)

Parameters:

auto_join (bool)
private_room_invites (bool)
favorites (set[str])

auto_join: bool

private_room_invites: bool

favorites: set[str]

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.UsersSettings(*, friends=<factory>, blocked=<factory>)

Parameters:

friends (set[str])
blocked (Annotated[dict[str, BlockingFlag], BeforeValidator(func=~aioslsk.settings.translate_blocked_users, json_schema_input_type=PydanticUndefined)])

friends: set[str]

blocked: Annotated[dict[str, BlockingFlag], BeforeValidator(func=translate_blocked_users, json_schema_input_type=PydanticUndefined)]

is_blocked(username, flag)

Parameters:

username (str)
flag (BlockingFlag)

Return type:

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.InterestsSettings(*, liked=<factory>, hated=<factory>)

Parameters:

liked (set[str])
hated (set[str])

liked: set[str]

hated: set[str]

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.DebugSettings(*, search_for_parent=True, ip_overrides=<factory>, log_connection_count=False)

Parameters:

search_for_parent (bool)
ip_overrides (dict[str, str])
log_connection_count (bool)

search_for_parent: bool

ip_overrides: dict[str, str]

log_connection_count: bool

model_config = {'validate_assignment': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aioslsk.settings.Settings(_case_sensitive=None, _nested_model_default_partial_update=None, _env_prefix=None, _env_file=PosixPath('.'), _env_file_encoding=None, _env_ignore_empty=None, _env_nested_delimiter=None, _env_nested_max_split=None, _env_parse_none_str=None, _env_parse_enums=None, _cli_prog_name=None, _cli_parse_args=None, _cli_settings_source=None, _cli_parse_none_str=None, _cli_hide_none_type=None, _cli_avoid_json=None, _cli_enforce_required=None, _cli_use_class_docs_for_groups=None, _cli_exit_on_error=None, _cli_prefix=None, _cli_flag_prefix_char=None, _cli_implicit_flags=None, _cli_ignore_unknown_args=None, _cli_kebab_case=None, _cli_shortcuts=None, _secrets_dir=None, *, network=<factory>, credentials=<factory>, searches=<factory>, shares=<factory>, users=<factory>, rooms=<factory>, interests=<factory>, transfers=<factory>, debug=<factory>)

Parameters:

_case_sensitive (bool | None)
_nested_model_default_partial_update (bool | None)
_env_prefix (str | None)
_env_file (DotenvType | None)
_env_file_encoding (str | None)
_env_ignore_empty (bool | None)
_env_nested_delimiter (str | None)
_env_nested_max_split (int | None)
_env_parse_none_str (str | None)
_env_parse_enums (bool | None)
_cli_prog_name (str | None)
_cli_parse_args (bool | list[str] | tuple[str, ...] | None)
_cli_settings_source (CliSettingsSource[Any] | None)
_cli_parse_none_str (str | None)
_cli_hide_none_type (bool | None)
_cli_avoid_json (bool | None)
_cli_enforce_required (bool | None)
_cli_use_class_docs_for_groups (bool | None)
_cli_exit_on_error (bool | None)
_cli_prefix (str | None)
_cli_flag_prefix_char (str | None)
_cli_implicit_flags (bool | None)
_cli_ignore_unknown_args (bool | None)
_cli_kebab_case (bool | Literal['all', 'no_enums'] | None)
_cli_shortcuts (Mapping[str, str | list[str]] | None)
_secrets_dir (PathType | None)
network (NetworkSettings)
credentials (CredentialsSettings)
searches (SearchSettings)
shares (SharesSettings)
users (UsersSettings)
rooms (RoomsSettings)
interests (InterestsSettings)
transfers (TransfersSettings)
debug (DebugSettings)

network: NetworkSettings

credentials: CredentialsSettings

searches: SearchSettings

shares: SharesSettings

users: UsersSettings

rooms: RoomsSettings

interests: InterestsSettings

transfers: TransfersSettings

debug: DebugSettings

model_config = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_shortcuts': None, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_nested_max_split': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': '', 'extra': 'forbid', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'secrets_dir': None, 'toml_file': None, 'validate_assignment': True, 'validate_default': True, 'yaml_config_section': None, 'yaml_file': None, 'yaml_file_encoding': None}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

aioslsk.base_manager

class aioslsk.base_manager.BaseManager

Base class for manager instances. This class provides hooks that will get called during starting and stopping of the client

async load_data(): Load the data for the manager. This method gets called during client client start-up before any connection is made

async store_data(): Stores data from the manager. This gets called during the client shutdown

async start(): Optionally performs a start action on the manager. Gets called after loading the data but before connecting

async stop()

Cancel all running tasks. The implementation of this method should simply cancel the task and return, the code calling this method should be responsible for awaiting the cancellation

Returns:: list of all cancelled tasks
Return type:: list[Task]

aioslsk.constants

aioslsk.constants.PEER_CONNECT_TIMEOUT: float = 10: Direct connection timeout

aioslsk.constants.PEER_INDIRECT_CONNECT_TIMEOUT: float = 60: Indirect connection timeout

aioslsk.constants.PEER_INIT_TIMEOUT: float = 5: Timeout waiting for Peer Initialization message

aioslsk.constants.PEER_READ_TIMEOUT: float = 60: Timeout waiting for message on a peer or distributed connection

aioslsk.constants.TRANSFER_TIMEOUT: float = 180: Timeout waiting for new data on a file connection

aioslsk.constants.PATH_SEPERATOR_PATTERN = re.compile('[\\\\/]+'): Pattern for splitting/normalizing remote paths

aioslsk.constants.POTENTIAL_PARENTS_CACHE_SIZE: int = 20: Maximum amount of potential parents stored

aioslsk.distributed

aioslsk.distributed.task_counter(): Implement next(self).

class aioslsk.distributed.DistributedPeer(username, connection, branch_level=None, branch_root=None)

Represents a distributed peer and its values in the distributed network

Parameters:

username (str)
connection (PeerConnection)
branch_level (int | None)
branch_root (str | None)

username: str

connection: PeerConnection: Distributed connection (type=D) associated with the peer

branch_level: int | None

branch_root: str | None

class aioslsk.distributed.DistributedNetwork(settings, event_bus, network)

Class responsible for handling the distributed network

Parameters:

settings (Settings)
event_bus (EventBus)
network (Network)

parent: DistributedPeer | None: Distributed parent. This variable is None if we are looking for a parent

children: list[DistributedPeer]

potential_parents: deque[str]

distributed_peers: list[DistributedPeer]

parent_min_speed: int | None

parent_speed_ratio: int | None

min_parents_in_cache: int | None

parent_inactivity_timeout: int | None

distributed_alive_interval: int | None

register_listeners()

get_distributed_peer(connection)

Get the distributed peer object related to the given connection.

This method will return None if the connection was not properly initialized (no username set on the connection) or if there is no peer object associated with the connection

Parameters:: connection (PeerConnection)
Return type:: DistributedPeer | None

async reset(): Disconnects all parent and child connections

async send_messages_to_children(*messages)

Parameters:: messages (MessageDataclass | bytes)

async stop()

Cancels all pending tasks

Returns:: a list of tasks that have been cancelled so that they can be awaited
Return type:: list[Task]

aioslsk.peer

class aioslsk.peer.PeerManager(settings, event_bus, user_manager, shares_manager, upload_info_provider, network)

Peer manager is responsible for handling peer messages

Parameters:

settings (Settings)
event_bus (EventBus)
user_manager (UserManager)
shares_manager (SharesManager)
upload_info_provider (UploadInfoProvider)
network (Network)

register_listeners()

aioslsk.server

class aioslsk.server.ServerManager(settings, event_bus, network)

Class handling server state changes

Parameters:

settings (Settings)
event_bus (EventBus)
network (Network)

property connection_state: ConnectionState

register_listeners()

async send_ping(): Send ping to the server

aioslsk.session

class aioslsk.session.Session(user: aioslsk.user.model.User, ip_address: str, greeting: str, client_version: int, minor_version: int, logged_in_at: datetime.datetime = <factory>, privileges_time_left: int = 0)

Parameters:

user (User)
ip_address (str)
greeting (str)
client_version (int)
minor_version (int)
logged_in_at (datetime)
privileges_time_left (int)

user: User

ip_address: str

greeting: str

client_version: int

minor_version: int

logged_in_at: datetime

privileges_time_left: int

property privileged: bool

aioslsk.tasks

class aioslsk.tasks.BackgroundTask(interval, task_coro, preempt_wait=False, name='background-task', context=None)

Wrapper class for a background task

Variables:

interval – interval at which the task should run
task_coro – coroutine function that should be run at the provided interval
preempt_wait – whether to perform an initial wait before running the coroutine for the first time
name – optional task name
context – optional context to pass to the coroutine

Parameters:

interval (float | Callable[[], float])
task_coro (Callable[[], Coroutine[None, None, float | None]] | Callable[[Any], Coroutine[None, None, float | None]])
preempt_wait (bool)
name (str)
context (Any | None)

resolve_interval()

Return type:: float

start()

cancel()

Return type:: Task | None

is_running()

Return type:: bool

async runner()

class aioslsk.tasks.Timer(timeout, callback)

Re-schedulable timer which runs a callback after completion

Parameters:

timeout (float)
callback (Callable[[], Coroutine[None, None, float | None]] | Callable[[Any], Coroutine[None, None, float | None]])

start()

cancel()

Return type:: Task | None

async runner()

reschedule(timeout=None)

Parameters:: timeout (float | None)

aioslsk.utils

aioslsk.utils.task_counter(): Implement next(self).

aioslsk.utils.try_decoding(value)

Parameters:: value (bytes)

aioslsk.utils.split_remote_path(path)

Splits a remote path into parts. Empty parts will be filtered out

Parameters:: path (str)
Return type:: list[str]

aioslsk.utils.get_duration(attributes)

Parameters:: attributes (list[Attribute])
Return type:: str

aioslsk.utils.get_attribute_string(attributes)

Returns an attributes list to a string (excludes duration)

Parameters:: attributes (list[Attribute])
Return type:: str

aioslsk.utils.ticket_generator(initial=1)

Generator for tickets to be used in various protocol messages. The generator will be reset to the initial value once the value would exceed 2 ^ 32

Parameters:: initial (int)
Return type:: Generator[int, None, None]

async aioslsk.utils.cancel_task(task)

Parameters:: task (Task | None)

aioslsk.transfer.manager

aioslsk.transfer.manager.task_counter(): Implement next(self).

class aioslsk.transfer.manager.TransferManager(settings, event_bus, user_manager, shares_manager, network, cache=None)

Class responsible for transfer related functionality. This class stores transfer objects and handles related events.

Parameters:

settings (Settings)
event_bus (EventBus)
user_manager (UserManager)
shares_manager (SharesManager)
network (Network)
cache (Optional[TransferCache])

property transfers

register_listeners()

async read_cache(): Reads the transfers from the caches and corrects the state of those transfers

write_cache(): Write all currently stored transfers to the cache

async load_data(): Load the data for the manager. This method gets called during client client start-up before any connection is made

async store_data(): Stores data from the manager. This gets called during the client shutdown

async start(): Optionally performs a start action on the manager. Gets called after loading the data but before connecting

async stop()

Cancel all current transfer tasks

Returns:: a list of tasks that have been cancelled so that they can be awaited
Return type:: list[Task]

async start_progress_reporting_task(): Start the transfer progress reporting background task which will emit aioslsk.events.TransferProgressEvent events at an interval defined in the settings. This method is called automatically when starting the client

stop_progress_reporting_task(): Start the transfer progress reporting task

async download(username, filename, paused=False)

Requests to start a downloading the file from the given user

Parameters:

user – User from which to download the file
filename (str) – Name of the file to download. This should be the full path to the file as returned in the search results
paused (bool) – Adds the download in the paused state
username (str)

Returns:

a Transfer object from which the status of the transfer can be requested. If the transfer already exists in the client then this transfer will be returned

Return type:

Transfer

async abort(transfer)

Aborts the given transfer. This will cancel all pending transfers and remove the file (in case of download)

Parameters:

transfer (Transfer) – Transfer object to abort

Raises:

TransferNotFoundError – if the transfer has not been added to the manager first
InvalidStateTransition – When the transfer could not be transitioned to aborted

async queue(transfer)

Places given transfer (back) in the queue

This method can be called on downloads that are in the following states:

ABORTED: Re-starts the aborted download
PAUSED: Continues the paused download
COMPLETE: Re-downloads the file. By default this will download the file to a new location unless the initial file was removed
INCOMPLETE: The library should automatically attempt to retry a download in this state
FAILED: Re-attempt to download the file

Parameters:

transfer (Transfer) – Transfer object to queue

Raises:

TransferNotFoundError – if the transfer has not been added to the manager first
InvalidStateTransition – When the transfer could not be transitioned to queued

async pause(transfer)

Parameters:: transfer (Transfer)

async add(transfer)

Adds a transfer if it does not already exist, otherwise it returns the already existing transfer. This method will emit a TransferAddedEvent only if the transfer did not exist.

This method only adds a transfer and does not automatically start it. To do so call queue() on the manager

Parameters:: transfer (Transfer) – Transfer to be added
Returns:: either the transfer we have passed or the already existing transfer
Return type:: Transfer

async remove(transfer)

Remove a transfer from the list of transfers. This will attempt to abort the transfer before removing it. Emits a TransferRemovedEvent after removal

Parameters:: transfer (Transfer) – Transfer object to remove
Raises:: TransferNotFoundError – Raised when the transfer was not added to the manager

get_uploads()

Return type:: list[Transfer]

get_downloads()

Return type:: list[Transfer]

get_upload_slots()

Returns the total amount of upload slots

Return type:: int

has_slots_free()

Return type:: bool

get_free_upload_slots()

Return type:: int

get_queue_size()

Returns the amount of queued uploads

Return type:: int

get_downloading()

Returns all transfers that are currently downloading or an attempt is is made to start downloading

Return type:: list[Transfer]

get_uploading()

Returns all transfers that are currently uploading or an attempt is is made to start uploading

Return type:: list[Transfer]

get_finished_transfers()

Returns a complete list of transfers that are in a finalized state (COMPLETE, ABORTED, FAILED)

Return type:: list[Transfer]

get_unfinished_transfers()

Returns a complete list of transfers that are not in a finalized state (COMPLETE, ABORTED, FAILED)

Return type:: list[Transfer]

get_download_speed()

Return current download speed (in bytes/second)

Return type:: float

get_upload_speed()

Return current upload speed (in bytes/second)

Return type:: float

get_average_upload_speed()

Returns average upload speed (in bytes/second)

Return type:: float

get_place_in_queue(transfer)

Gets the place of the given upload in the transfer queue

Returns:: The place in the queue, 0 if not in the queue a value equal or greater than 1 indicating the position otherwise
Parameters:: transfer (Transfer)
Return type:: int

get_transfer(username, remote_path, direction)

Lookup transfer by username, remote_path and transfer direction. This method will raise an exception in case the transfer is not found

Parameters:

username (str) – Username of the transfer
remote_path (str) – Full remote path of the transfer
direction (TransferDirection) – Direction of the transfer (upload / download)

Returns:

The matching transfer object

Raises:

ValueError – If the transfer is not found

Return type:

Transfer

find_transfer(username, remote_path, direction)

Lookup transfer by username, remote_path and transfer direction. This method will return None if the transfer is not found

Parameters:

username (str) – Username of the transfer
remote_path (str) – Full remote path of the transfer
direction (TransferDirection) – Direction of the transfer (upload / download)

Returns:

The matching transfer object or None

Return type:

Transfer | None

async manage_user_tracking(): Remove or add user tracking based on the list of transfers. This method will untrack users for which there are no more unfinished transfers and start/keep tracking users for which there are unfinished transfers

request_management_cycle(flag)

Parameters:: flag (_RequestFlag)

manage_transfers(): This method analyzes the state of the current downloads/uploads and starts them up in case there are free slots available

async manage_shares_changed()

async request_place_in_queue(transfer)

Requests the place in queue for the given transfer. The method will return the value in case of success and return the value

Returns:: place in queue or None in case of error
Raises:: RequestPlaceFailedError – when the request failed to send to the peer or waiting for a response timed out
Parameters:: transfer (Transfer)
Return type:: int | None

async on_transfer_state_changed(transfer, old, new)

Parameters:

transfer (Transfer)
old (State)
new (State)

aioslsk.transfer.model

class aioslsk.transfer.model.FailReason

Definition of reasons for which an transfer queue request or transfer request was rejected

CANCELLED = 'Cancelled'

COMPLETE = 'Complete'

QUEUED = 'Queued'

FILE_NOT_SHARED = 'File not shared.'

FILE_READ_ERROR = 'File read error.'

class aioslsk.transfer.model.AbortReason

REQUESTED = 'Requested'

BLOCKED = 'Blocked'

FILE_NOT_SHARED = 'File not shared'

class aioslsk.transfer.model.TransferDirection(*values)

UPLOAD = 0

DOWNLOAD = 1

class aioslsk.transfer.model.TransferProgressSnapshot(state, bytes_transfered, speed, start_time=None, complete_time=None, fail_reason=None, abort_reason=None)

Represents the current progress of a transfer, used for reporting progress back to the user through a TransferProgressEvent

Parameters:

state (State)
bytes_transfered (int)
speed (float)
start_time (float | None)
complete_time (float | None)
fail_reason (str | None)
abort_reason (str | None)

state: State

bytes_transfered: int: Amount of bytes transfered in bytes

speed: float: Current transfer speed if the transfer is still downloading, average speed if the transfer has completed

start_time: float | None = None

complete_time: float | None = None

fail_reason: str | None = None: Optional transfer fail reason

abort_reason: str | None = None: Optional transfer abort reason

class aioslsk.transfer.model.Transfer(username, remote_path, direction)

Class representing a transfer

Parameters:

username (str)
remote_path (str)
direction (TransferDirection)

state: TransferState

direction: TransferDirection: Determines whether this transfer is an upload or a download

username: str: Username of the peer

remote_path: str: Remote path, this is the path that is shared between peers

local_path: str | None: Absolute path to the file on disk

remotely_queued: bool: Indicates whether the transfer queue message was received by the peer. This only used for downloads

place_in_queue: int | None

fail_reason: str | None: Reason for failure in case the transfer has failed

abort_reason: str | None: Reason for why transfer was aborted

filesize: int | None: Filesize in bytes

bytes_transfered: int: Amount of bytes transfered

queue_attempts: int

last_queue_attempt: float

upload_request_attempts: int

last_upload_request_attempt: float

start_time: float | None: Time at which the transfer was started. This is the time the transfer entered the DOWNLOADING or UPLOADING state

complete_time: float | None: Time at which the transfer was completed. This is the time the transfer entered the COMPLETE, INCOMPLETE, ABORTED or FAILED state

progress_snapshot: TransferProgressSnapshot: Snapshot of the transfer progress. Used internally to report progress at set intervals

state_listeners: list[TransferStateListener]

reset_local_vars(): Resets the local file variables

reset_progress_vars()

reset_queue_vars(): Reset all variables when the

reset_time_vars(): Clear all time related variables

set_start_time(): Set the start time, clear the complete time

set_complete_time(): Set the complete time only if the start time has not been set

increase_queue_attempts()

reset_queue_attempts()

increase_upload_request_attempt()

reset_upload_request_attempts()

async transition(state)

Transitions the state of the transfer and notifies the state_listeners. This is an internal method and should not be used directly.

Parameters:: state (TransferState)

get_speed()

Retrieve the speed of the transfer

Returns:: 0 if the transfer has not yet begun. The current speed if the transfer is ongoing. The transfer speed if the transfer was complete. Bytes per second
Return type:: float

add_speed_log_entry(bytes_transfered)

Logs a transfer speed entry. This is an internal method

Parameters:: bytes_transfered (int)

is_upload()

Return type:: bool

is_download()

Return type:: bool

is_finalized()

Return true if the transfer is in a finalized state

Return type:: bool

is_processing()

Return true if an attempt is being made to start transferring the file or the transfer is currently in progress.

Return type:: bool

is_transferring()

Return true if the transfer is in progress

Return type:: bool

is_transfered()

Return type:: bool

get_tasks()

Return type:: list[Task]

cancel_tasks()

Cancels all tasks for the transfer, this method returns the tasks which have been cancelled

Return type:: list[Task]

take_progress_snapshot()

Saves and returns a snapshot of the transfer progress

Return type:: TransferProgressSnapshot

aioslsk.transfer.cache

class aioslsk.transfer.cache.TransferCache(*args, **kwargs)

Abstract base class for storing shares

read()

Return type:: list[Transfer]

write(transfers)

Parameters:: transfers (list[Transfer])

class aioslsk.transfer.cache.TransferNullCache

Transfer cache object that does not perform any caching

read()

Return type:: list[Transfer]

write(transfers)

Parameters:: transfers (list[Transfer])

class aioslsk.transfer.cache.TransferShelveCache(data_directory)

Transfer cache that uses the Python built-in shelve module to store transfer objects

Parameters:: data_directory (str)

DEFAULT_FILENAME = 'transfers'

read()

Return type:: list[Transfer]

write(transfers)

Parameters:: transfers (list[Transfer])

aioslsk.shares.cache

class aioslsk.shares.cache.SharesCache(*args, **kwargs)

Abstract base class for storing shares

read()

Return type:: list[SharedDirectory]

write(shared_directories)

Parameters:: shared_directories (list[SharedDirectory])

class aioslsk.shares.cache.SharesNullCache

read()

Return type:: list[SharedDirectory]

write(shared_directories)

Parameters:: shared_directories (list[SharedDirectory])

class aioslsk.shares.cache.SharesShelveCache(data_directory, filename='shares_index')

Shares cache that uses the Python built-in shelve module

Parameters:

data_directory (str)
filename (str)

DEFAULT_FILENAME = 'shares_index'

read()

Return type:: list[SharedDirectory]

write(shared_directories)

Parameters:: shared_directories (list[SharedDirectory])

aioslsk.shares.model

class aioslsk.shares.model.DirectoryShareMode(*values)

Share mode for directories. The mode determines for who the results are locked and who can download the file

EVERYONE = 'everyone'

FRIENDS = 'friends'

USERS = 'users'

class aioslsk.shares.model.SharedDirectory(directory, absolute_path, alias, share_mode=DirectoryShareMode.EVERYONE, users=<factory>)

Class representing a directory that is explicitly shared by the user. This class holds information on how the directory is shared and is a container for shared items within this directory.

Parameters:

directory (str)
absolute_path (str)
alias (str)
share_mode (DirectoryShareMode)
users (list[str])

directory: str

absolute_path: str

alias: str

share_mode: DirectoryShareMode = 'everyone'

users: list[str]

items: set[SharedItem]

is_parent_of(directory)

Returns true if the current directory is any parent of the passed shared directory

Parameters:: directory (str | SharedDirectory) – directory to test, if this directory is of str type it should be an absolute path. If it is of SharedDirectory type the absolute_path will be taken
Return type:: bool

is_child_of(directory)

Returns true if the passed directory is a (sub)child of the current directory

Parameters:: directory (str | SharedDirectory) – directory to test, if this directory is of str type it should be an absolute path. If it is of SharedDirectory type the absolute_path will be taken
Return type:: bool

get_remote_path()

Return type:: str

get_item_by_remote_path(remote_path)

Returns the SharedItem instance belonging to the passed remote_path

Raises:: FileNotFoundError – when the item cannot be found in the set of items
Parameters:: remote_path (str)
Return type:: SharedItem

get_items_for_directory(directory)

Gets items in the current directory that are part of given directory

Parameters:: directory (SharedDirectory)
Return type:: set[SharedItem]

class aioslsk.shares.model.SharedItem(shared_directory: aioslsk.shares.model.SharedDirectory, subdir: str, filename: str, modified: float)

Parameters:

shared_directory (SharedDirectory)
subdir (str)
filename (str)
modified (float)

shared_directory: SharedDirectory

subdir: str

filename: str

modified: float

attributes: list[tuple[int, int]] | None = None

get_absolute_path()

Returns the absolute path of the shared item

Return type:: str

get_remote_path()

Returns the full remote path of this file.

Returns:: Full remote file path: Example: “@@abcdeMusicsong.mp3”
Return type:: str

get_remote_directory_path()

Returns the remote directory path this file resides in

Returns:: Remote directory path. Example: “@@abcdeMusic”
Return type:: str

get_remote_directory_path_parts()

Returns the remote directory path split into parts

Return type:: tuple[str, …]

get_query_path()

Returns the query-able part of the shared item

Returns:: Queryable path of the item. Example: “MusicMetalSong.mp3”
Return type:: str

aioslsk.shares.manager

aioslsk.shares.manager.scan_directory(shared_directory, children=None)

Scans the directory for items to share

Warning: when using ProcessPoolExecutor on this method the returned items will not have the shared directory set in some cases. The shared_directory passed here is a copy and the object will be removed once this function finishes (see https://stackoverflow.com/a/72726998/1419478). In this case you should manually assign it again

Parameters:

shared_directory (SharedDirectory) – SharedDirectory instance
children (list[SharedDirectory] | None) – list of SharedDirectory instances, the items in this list will not be returned and should be scanned individually

Returns:

set of SharedItem objects found during the scan

Return type:

set[SharedItem]

aioslsk.shares.manager.extract_attributes(filepath)

Attempts to extract attributes from the file at filepath. If there was an error attempting to extract the attributes this method will log a warning and return an empty list of attributes

Parameters:: filepath (str)
Return type:: list[tuple[int, int]]

class aioslsk.shares.manager.SharesManager(settings, event_bus, network, cache=None, executor_factory=None)

Parameters:

settings (Settings)
event_bus (EventBus)
network (Network)
cache (SharesCache | None)
executor_factory (Callable[[], Executor] | None)

register_listeners()

property shared_directories

generate_alias(path, offset=0)

Generates a directory alias for the given path, this method will be called recursively increasing the offset in case the alias is already found in the shared_directories.

The hardware address is mixed in to avoid getting the same alias for the same directory on different machines. Admittedly this is a lousy security measure but hopefully this will prevent easy leaking of files in case such issue would occur. Example: ‘abcde’ is always generated for ‘C:' so the attacker could guess where a file is located.

Parameters:

path (str) – the path for which to generate an alias
offset (int) – offset for the value of the initial 5 bytes (default=0)

Returns:

a string of 5 alphabetic characters all lowercased

Return type:

str

async start(): Optionally performs a start action on the manager. Gets called after loading the data but before connecting

async stop()

Cancel all running tasks. The implementation of this method should simply cancel the task and return, the code calling this method should be responsible for awaiting the cancellation

Returns:: list of all cancelled tasks
Return type:: list[Task]

async load_data(): Load the data for the manager. This method gets called during client client start-up before any connection is made

async store_data(): Stores data from the manager. This gets called during the client shutdown

load_from_settings(): Loads the shared directories from the settings. Existing directories will be updated, non-existing directories added, directories that no longer exist will be removed

read_cache(): Read the directories from the cache

write_cache(): Write current shared directories to the cache

get_download_directory()

Gets the absolute path the to download directory configured from the settings

Returns:: Absolute path of the value of the shares.download setting
Return type:: str

async create_directory(absolute_path)

Ensures the passed directory exists

Parameters:: absolute_path (str)

rebuild_term_map()

get_shared_item_cache(remote_path, username=None)

Gets a shared item from the cache if it exists

If a username is passed this will also check if the file is locked and raise a FileNotSharedError if the file is not accessible for that user

Parameters:

remote_path (str) – the remote_path
username (str | None) – optional username to check if the file is shared or not

Raises:

FileNotFoundError – filename was not found in shared_items
FileNotSharedError – file is found, but locked for the given username

Return type:

SharedItem

async get_shared_item(remote_path, username=None)

Same as get_shared_item_cache() but also checks if the file exists on disk

Parameters:

remote_path (str) – the remote_path
username (str | None) – optional username to check if the file is shared or not

Raises:

FileNotFoundError – filename was not found in shared_items or was found but did not exist on disk
FileNotSharedError – file is found, but locked for the given username

Return type:

SharedItem

find_shared_item_cache(remote_path, username=None)

Equivelant to get_shared_item_cache() but returns None if the shared item is not found

Parameters:

remote_path (str)
username (str | None)

Return type:

SharedItem | None

async find_shared_item(remote_path, username=None)

Equivelant to get_shared_item() but returns None if the shared item is not found

Parameters:

remote_path (str)
username (str | None)

Return type:

SharedItem | None

add_shared_directory(shared_directory, share_mode=DirectoryShareMode.EVERYONE, users=None)

Adds a shared directory. This method will call generate_alias() and add the directory to the directory map.

This method will not scan the directory, however if the directory is a (sub)child of an already registered shared directory the items from that directory will be removed from the parent directory to the new directory

For scanning see the scan(), scan_directory_files() and scan_directory_file_attributes() methods.

Parameters:

shared_directory (str) – path of the shared directory
share_mode (DirectoryShareMode) – the share mode for the directory
users (list[str] | None) – in case the share mode is USERS, a list of users to share it with

Returns:

a SharedDirectory object

Raises:

SharedDirectoryError – if the directory is already shared

Return type:

SharedDirectory

update_shared_directory(directory, share_mode=None, users=None)

Updates share_mode and users values for the given directory

Parameters:

directory (str | SharedDirectory) – if a string is given this method will attempt to find the shared directory based on the absolute path
share_mode (DirectoryShareMode | None)
users (list[str] | None)

Returns:

the updated directory

Raises:

SharedDirectoryError – if the given directory is a string and not in the list of shared directories

Return type:

SharedDirectory

remove_shared_directory(directory)

Removes the given shared directory. If the directory was a subdirectory of another shared directory its items will be moved into that directory

Example file structure:

Music
- Artist_One
  song_one.mp3
- Artist_Two
  song_two.mp3

If the user shares 2 directories:

Music: EVERYONE
MusicArtist_One: FRIENDS

And removes the MusicArtist_Oneshared directory the files of that directory get returned back to the parent directory. In this case file ``song_two.mp3` will be shared with EVERYONE again

Parameters:

shared_directory – SharedDirectory instance to remove
directory (str | SharedDirectory)

Returns:

the removed directory

Raises:

SharedDirectoryError – raised when the passed directory was not added to the manager

Return type:

SharedDirectory

get_shared_directory(directory)

Calculates the absolute path of given directory and looks for the matching SharedDirectory instance

Raises:: SharedDirectoryError – if there is no corresponding SharedDirectory instance registered in this class
Parameters:: directory (str)
Return type:: SharedDirectory

is_directory_shared(directory)

Checks if a directory is already a shared directory by checking the absolute path of that directory

Parameters:: directory (str)
Return type:: bool

async scan_directory_files(shared_directory)

Scans the files for the given shared_directory

Parameters:: shared_directory (SharedDirectory) – SharedDirectory instance to scan
Raises:: SharedDirectoryError – raised when the passed shared_directory was not added to the manager

async scan_directory_file_attributes(shared_directory)

Scans the file attributes for files in the given shared_directory only files that do not have attributes will be scanned

The results of the scan are handled internally and are automatically to the SharedItem object for which the scan was performed

Parameters:: shared_directory (SharedDirectory) – SharedDirectory instance for which the files need to be scanned

async scan()

Scan the files and their attributes for all directories currently defined in the shared_directories

This method will emit a ScanCompleteEvent on the event bus and report the shares to the server

async get_filesize(shared_item)

Parameters:: shared_item (SharedItem)
Return type:: int

query(query, username=None, excluded_search_phrases=None)

Performs a query on the shared_directories returning the matching items. If username is passed this method will return a list of visible results and list of locked results. If None the second list will always be empty.

This method makes a first pass using the built in term map and filters the remaining results using regular expressions.

Parameters:

query (str | SearchQuery) – the query to perform on the shared directories
username (str | None) – optionally the username of the user making the query. This is used to determine locked results, if not given the locked results list will be empty
excluded_search_phrases (list[str] | None) – optional list of search phrases that should be excluded from the search results

Returns:

two lists of visible results and locked results

Return type:

tuple[list[SharedItem], list[SharedItem]]

get_stats()

Gets the total amount of shared directories and files.

Returns:: directory and file count as a tuple
Return type:: tuple[int, int]

calculate_download_path(remote_path)

Calculates the local download path for a remote path returned by another peer.

Returns:: tuple of the directory and file name
Parameters:: remote_path (str)
Return type:: tuple[str, str]

get_shared_directories_for_user(username)

Parameters:: username (str)
Return type:: tuple[list[SharedDirectory], list[SharedDirectory]]

create_shares_reply(username)

Creates a complete list of the currently shared items as a reply to a PeerSharesRequest messages

Parameters:: username (str) – username of the user requesting the shares reply, this is used to determine the locked results
Returns:: tuple with two lists: public directories and locked directories
Return type:: tuple[list[DirectoryData], list[DirectoryData]]

create_directory_reply(remote_directory)

Lists directory data as a response to a directory request. This will not contain any information about subdirectories, only the files within that directory

This method differs from the official clients who don’t respond at all to the message if the directory does not exist. Instead if the directory does not exist and empty list is returned

Parameters:: remote_directory (str) – remote path of the directory
Returns:: list of directories. Empty if the directory is not shared, a list with one entry if the directory is found
Return type:: list[DirectoryData]

is_directory_locked(directory, username)

Checks if the shared directory is locked for the given username

Parameters:

directory (SharedDirectory)
username (str)

Return type:

is_item_locked(item, username)

Checks if the shared item is locked for the given username

Parameters:

item (SharedItem)
username (str)

Return type:

async report_shares(): Reports the shares amount to the server

aioslsk.user.model

class aioslsk.user.model.UserStatus(*values)

User status values, everything except the UNKNOWN status are used by the server

UNKNOWN = -1

OFFLINE = 0

AWAY = 1

ONLINE = 2

class aioslsk.user.model.UploadPermissions(*values)

Upload permissions returned / sent by PeerUserInfoReply message

UNKNOWN = -1

NOONE = 0: Don’t allow uploads from anyone

EVERYONE = 1: Allow uploads from anyone

USER_LIST = 2: Only allow uploads from users in the users list

PERMITTED_LIST = 3: Only allow uploads from users in a specific list

class aioslsk.user.model.TrackingFlag(*values)

Tracking flags hold information how the user is being tracked

REQUESTED = 1: Tracking was explicitly requested by the user

TRANSFER = 2: Tracking was requested through transfer manager

FRIEND = 4: Tracking because the user is a friend

class aioslsk.user.model.TrackingState(*values)

UNTRACKED = 'untracked'

TRACKED = 'tracked'

RETRY_PENDING = 'retry_pending'

class aioslsk.user.model.BlockingFlag(*values)

NONE = 0

PRIVATE_MESSAGES = 1: Blocks private messages sent by the user

ROOM_MESSAGES = 2: Blocks room messages sent by the user

SEARCHES = 4: Blocks incoming searches from the user

SHARES = 8: Blocks shares, directory listing requests

INFO = 16

UPLOADS = 32: Blocks all uploads and upload requests

IGNORE = 3: Ignore all messaging

ALL = 63: Fully blocks the user: includes messaging, shares requests, uploads

class aioslsk.user.model.User(name: 'str', status: 'UserStatus' = <UserStatus.UNKNOWN: -1>, privileged: 'bool' = False, description: 'Optional[str]' = None, picture: 'Optional[bytes]' = None, country: 'Optional[str]' = None, avg_speed: 'Optional[int]' = None, uploads: 'Optional[int]' = None, shared_file_count: 'Optional[int]' = None, shared_folder_count: 'Optional[int]' = None, has_slots_free: 'Optional[bool]' = None, slots_free: 'Optional[int]' = None, upload_slots: 'Optional[int]' = None, queue_length: 'Optional[int]' = None, upload_permissions: 'UploadPermissions' = <UploadPermissions.UNKNOWN: -1>, interests: 'set[str]' = <factory>, hated_interests: 'set[str]' = <factory>)

Parameters:

name (str)
status (UserStatus)
privileged (bool)
description (str | None)
picture (bytes | None)
country (str | None)
avg_speed (int | None)
uploads (int | None)
shared_file_count (int | None)
shared_folder_count (int | None)
has_slots_free (bool | None)
slots_free (int | None)
upload_slots (int | None)
queue_length (int | None)
upload_permissions (UploadPermissions)
interests (set[str])
hated_interests (set[str])

name: str

status: UserStatus

privileged: bool

description: str | None

picture: bytes | None

country: str | None

avg_speed: int | None

uploads: int | None

shared_file_count: int | None

shared_folder_count: int | None

has_slots_free: bool | None

slots_free: int | None

upload_slots: int | None

queue_length: int | None

upload_permissions: UploadPermissions

interests: set[str]

hated_interests: set[str]

update_from_user_stats(user_stats)

Parameters:: user_stats (UserStats)

clear_all(): Clears all user data except user status and privileges info and country

clear(info=False, interests=False, stats=False, upload_info=False)

Resets selected fields

Parameters:

info (bool)
interests (bool)
stats (bool)
upload_info (bool)

class aioslsk.user.model.ChatMessage(id, timestamp, user, message, is_direct)

Represents a private chat message

Parameters:

id (int)
timestamp (int)
user (User)
message (str)
is_direct (bool)

id: int

timestamp: int

user: User

message: str

is_direct: bool

is_server_message()

Return type:: bool

property is_admin: bool

Only kept to keep backward compatibility. Use is_direct

This property does not actually represent whether the message was sent by an admin. Instead it represents whether the message was sent directly or was queued on the server before being sent (example, when user was offline and came back online)

aioslsk.user.manager

class aioslsk.user.manager.TrackingRequest(operation: Callable[[aioslsk.user.model.TrackingFlag], NoneType], flag: aioslsk.user.model.TrackingFlag, handled: asyncio.locks.Event = <factory>)

Parameters:

operation (Callable[[TrackingFlag], None])
flag (TrackingFlag)
handled (Event)

operation: Callable[[TrackingFlag], None]

flag: TrackingFlag

handled: Event

class aioslsk.user.manager.TrackedUser(user: aioslsk.user.model.User, flags: aioslsk.user.model.TrackingFlag = <TrackingFlag: 0>, state: aioslsk.user.model.TrackingState = <TrackingState.UNTRACKED: 'untracked'>, queue: asyncio.queues.Queue[aioslsk.user.manager.TrackingRequest] = <factory>, task: _asyncio.Task | None = None, retry_task: _asyncio.Task | None = None)

Parameters:

user (User)
flags (TrackingFlag)
state (TrackingState)
queue (Queue[TrackingRequest])
task (Task | None)
retry_task (Task | None)

user: User

flags: TrackingFlag

state: TrackingState

queue: Queue[TrackingRequest]

task: Task | None

retry_task: Task | None

add_flag(flag)

Parameters:: flag (TrackingFlag)

remove_flag(flag)

Parameters:: flag (TrackingFlag)

class aioslsk.user.manager.UserManagementContext(friends: set[str], blocked: dict[str, aioslsk.user.model.BlockingFlag])

Parameters:

friends (set[str])
blocked (dict[str, BlockingFlag])

friends: set[str]

blocked: dict[str, BlockingFlag]

class aioslsk.user.manager.UserManager(settings, event_bus, network)

Class responsible for handling user messages and storing users

Parameters:

settings (Settings)
event_bus (EventBus)
network (Network)

register_listeners()

property users: dict[str, User]

property privileged_users: set[str]

async start(): Optionally performs a start action on the manager. Gets called after loading the data but before connecting

async stop()

Cancel all running tasks. The implementation of this method should simply cancel the task and return, the code calling this method should be responsible for awaiting the cancellation

Returns:: list of all cancelled tasks
Return type:: list[Task]

get_self()

Returns the user object for the current session

Return type:: User

get_user_object(username)

Gets a User object for given username, if the user is not stored it will be created and stored.

A user will be stored only if there is a strong reference to the User object, otherwise it will be removed

Parameters:: username (str) – Name of the user
Returns:: a User object
Return type:: User

is_tracked(username)

Returns whether a user is currently being tracked

Parameters:: username (str)
Return type:: bool

reset_users(): Performs a reset on all users. This method is called when the connection with the server is disconnected and shouldn’t be called directly

async track_user(username, flag=<TrackingFlag.REQUESTED: 1>)

Requests to start tracking a user. Tracking of a user will be handled in the background

Parameters:

username (str) – user to track
flag (TrackingFlag) – tracking flag to add from the user

async untrack_user(username, flag=<TrackingFlag.REQUESTED: 1>)

Requests to stop tracking a user. Tracking of a user will be handled in the background.

Untracking does not necesserily mean that the library stops receiving user updates, the library could internally still be wanting to track a user for example for managing transfers

Parameters:

username (str) – user to track
flag (TrackingFlag) – tracking flag to add from the user

async track_friend(username)

Request to track a friend with given username

Parameters:: username (str)

async track_friends(): Starts tracking the users defined in the friends list

async untrack_friend(username)

Request to stop tracking a friend with given username

Parameters:: username (str)

get_tracking_flags(username)

Returns current desired tracking flags of the user

Parameters:: username (str) – username of the user
Return type:: TrackingFlag

get_tracking_state(username)

Returns current tracking state of the user

Parameters:: username (str) – username of the user
Return type:: TrackingState

class aioslsk.user.manager.UserTrackingManager(settings, event_bus, network)

Keeps the user tracking state in sync with the server

Parameters:

settings (Settings)
event_bus (EventBus)
network (Network)

register_listeners()

get_tracking_flags(username)

Returns current desired tracking flags of the user with given username

Parameters:: username (str) – username of the user
Return type:: TrackingFlag

get_tracking_state(username)

Returns the current tracking state of the given user

Parameters:: username (str)
Return type:: TrackingState

track_user(user, flag=<TrackingFlag.REQUESTED: 1>)

Parameters:

user (User)
flag (TrackingFlag)

Return type:

TrackingRequest

untrack_user(user, flag=<TrackingFlag.REQUESTED: 1>)

Parameters:

user (User)
flag (TrackingFlag)

Return type:

TrackingRequest | None

stop()

Return type:: list[Task]

aioslsk.room.model

class aioslsk.room.model.Room(name: 'str', private: 'bool' = False, users: 'list[User]' = <factory>, joined: 'bool' = False, user_count: 'int' = 0, tickers: 'dict[str, str]'=<factory>, members: 'set[str]' = <factory>, owner: 'Optional[str]' = None, operators: 'set[str]' = <factory>)

Parameters:

name (str)
private (bool)
users (list[User])
joined (bool)
user_count (int)
tickers (dict[str, str])
members (set[str])
owner (str | None)
operators (set[str])

name: str

private: bool

users: list[User]: Current list of joined users

joined: bool

user_count: int

tickers: dict[str, str]: Room tickers (room wall)

members: set[str]: For private rooms, names of members of the room (excludes owner)

owner: str | None: For private rooms, name of the room owner

operators: set[str]: For private rooms, names of operators

add_user(user)

Parameters:: user (User)

remove_user(user)

Parameters:: user (User)

class aioslsk.room.model.RoomMessage(timestamp: 'int', user: 'User', message: 'str', room: 'Room')

Parameters:

timestamp (int)
user (User)
message (str)
room (Room)

timestamp: int

user: User

message: str

room: Room

aioslsk.room.manager

class aioslsk.room.manager.RoomManager(settings, event_bus, user_manager, network)

Class handling rooms

Parameters:

settings (Settings)
event_bus (EventBus)
user_manager (UserManager)
network (Network)

property rooms: dict[str, Room]

get_joined_rooms()

Returns a list of all rooms currently joined

Return type:: list[Room]

get_private_rooms()

Returns a list of all private rooms of which we are a member or owner. This includes rooms that are not joined

Return type:: list[Room]

get_public_rooms()

Returns the list of public rooms

Return type:: list[Room]

get_owned_rooms()

Returns a list of which we are the owner

Return type:: list[Room]

get_operated_rooms()

Returns a list of rooms in which we have operator privileges

Return type:: list[Room]

get_or_create_room(room_name, private=False)

Parameters:

room_name (str)
private (bool)

Return type:

Room

reset_rooms(): Performs a reset on all users and rooms

register_listeners()

async auto_join_rooms(): Automatically joins rooms stored in the settings

aioslsk.search.model

class aioslsk.search.model.SearchType(*values)

NETWORK = 1

USER = 2

ROOM = 3

WISHLIST = 4

class aioslsk.search.model.ReceivedSearch(username, query, result_count)

Used for keeping track of searches received from the distributed parent or server

Parameters:

username (str)
query (str)
result_count (int)

username: str

query: str

result_count: int

class aioslsk.search.model.SearchQuery(query: str, include_terms: set[str] = <factory>, exclude_terms: set[str] = <factory>, wildcard_terms: set[str] = <factory>)

Parameters:

query (str)
include_terms (set[str])
exclude_terms (set[str])
wildcard_terms (set[str])

query: str

include_terms: set[str]

exclude_terms: set[str]

wildcard_terms: set[str]

classmethod parse(query)

Parses the given query string into terms and creates a new SearchQuery object

Parameters:: query (str)
Return type:: SearchQuery

matchers_iter()

Generator for term matchers

Return type:: Generator[Callable[[str], bool], None, None]

has_inclusion_terms()

Return whether this query has any valid inclusion terms

Return type:: bool

class aioslsk.search.model.SearchResult(ticket, username, has_free_slots=False, avg_speed=0, queue_size=0, shared_items=<factory>, locked_results=<factory>)

Search result received from a user

Parameters:

ticket (int)
username (str)
has_free_slots (bool)
avg_speed (int)
queue_size (int)
shared_items (list[FileData])
locked_results (list[FileData])

ticket: int

username: str

has_free_slots: bool

avg_speed: int

queue_size: int

shared_items: list[FileData]

locked_results: list[FileData]

class aioslsk.search.model.SearchRequest(ticket, query, search_type=SearchType.NETWORK, room=None, username=None, results=<factory>, started=<factory>, timer=None)

Search request we have made

Parameters:

ticket (int)
query (str)
search_type (SearchType)
room (str | None)
username (str | None)
results (list[SearchResult])
started (datetime)
timer (Timer | None)

ticket: int

query: str

search_type: SearchType

room: str | None

username: str | None

results: list[SearchResult]

started: datetime

timer: Timer | None

aioslsk.search.manager

aioslsk.search.manager.task_counter(): Implement next(self).

class aioslsk.search.manager.SearchManager(settings, event_bus, shares_manager, upload_info_provider, network)

Handler for searches requests

Parameters:

settings (Settings)
event_bus (EventBus)
shares_manager (SharesManager)
upload_info_provider (UploadInfoProvider)
network (Network)

register_listeners()

remove_request(request)

Removes the search request from the client. Incoming results after the request has been removed will be ignored

Parameters:: request (SearchRequest | int) – SearchRequest object or ticket number to remove

async search(query)

Performs a global search. The results generated by this query will stored in the returned object or can be listened to through the SearchResultEvent event

Parameters:: query (str) – The search query
Returns:: An object containing the search request details and results
Return type:: SearchRequest

async search_room(room, query)

Performs a search request on the specific room. The results generated by this query will stored in the returned object or can be listened to through the SearchResultEvent event

Parameters:

room (str | Room) – Room object or name to query
query (str) – The search query

Returns:

An object containing the search request details and results

Return type:

SearchRequest

async search_user(username, query)

Performs a search request on the specific user. The results generated by this query will stored in the returned object or can be listened to through the SearchResultEvent event

Parameters:

user – User object or name to query
query (str) – The search query
username (str)

Returns:

An object containing the search request details and results

Return type:

SearchRequest

async stop()

Cancels all pending tasks

Returns:: a list of tasks that have been cancelled so that they can be awaited
Return type:: list[Task]

aioslsk.interest.manager

class aioslsk.interest.manager.InterestManager(settings, event_bus, user_manager, network)

Class handling interests and recommendations

Parameters:

settings (Settings)
event_bus (EventBus)
user_manager (UserManager)
network (Network)

register_listeners()

async advertise_interests(): Advertises all interests and hated interests defined in the settings to the server

aioslsk.network.connection

aioslsk.network.connection.task_counter(): Implement next(self).

class aioslsk.network.connection.PeerConnectionType

FILE = 'F'

PEER = 'P'

DISTRIBUTED = 'D'

class aioslsk.network.connection.ConnectionState(*values)

UNINITIALIZED = 1

CONNECTING = 2

CONNECTED = 3

CLOSING = 4

CLOSED = 5

class aioslsk.network.connection.CloseReason(*values)

UNKNOWN = 1

CONNECT_FAILED = 2

REQUESTED = 3

READ_ERROR = 4

WRITE_ERROR = 5

TIMEOUT = 6

EOF = 7

class aioslsk.network.connection.PeerConnectionState(*values)

AWAITING_INIT = 1: No init message has been received yet (PeerInit, PeerPierceFirewall)

ESTABLISHED = 2: Message connections: ready to receive / send data

NEGOTIATING_TRANSFER = 3: Transfer connections: negotiating the transfer ticket / offset

TRANSFERRING = 4: Transfer connections: ready to receive / send data

class aioslsk.network.connection.Connection(hostname, port, network)

Parameters:

hostname (str)
port (int)
network (Network)

hostname: str

port: int

network: Network

state: ConnectionState

abstractmethod async disconnect(reason=CloseReason.UNKNOWN)

Parameters:: reason (CloseReason)

async set_state(state, close_reason=CloseReason.UNKNOWN)

Parameters:

state (ConnectionState)
close_reason (CloseReason)

class aioslsk.network.connection.ListeningConnection(hostname, port, network, obfuscated=False)

A listening connection, objects of this class are responsible for accepting incoming connections from peers

Parameters:

hostname (str)
port (int)
network (Network)
obfuscated (bool)

obfuscated: bool

async disconnect(reason=CloseReason.UNKNOWN)

Parameters:: reason (CloseReason)

async connect()

Open a listening connection on the current hostname and port

Raises:: ConnectionFailedError – raised when binding failed

async accept(reader, writer)

Parameters:

reader (StreamReader)
writer (StreamWriter)

class aioslsk.network.connection.DataConnection(hostname, port, network, obfuscated=False, read_timeout=60)

Connection for message and data transfer

Parameters:

hostname (str)
port (int)
network (Network)
obfuscated (bool)
read_timeout (float)

read_timeout: float

get_connecting_ip()

Gets the IP address being used to connect to the server/peer.

The connection needs to be established for this method to work

Return type:: str

async connect(timeout=30)

Opens a TCP connection the hostname:ip provided in this object. The state of the connection will be changed to CONNECTING before the attempt is made

Upon success the state will be set to CONNECTED and the message reader loop will be started. Upon failure the state will be set to CLOSED with CONNECT_FAILED as its reason

Parameters:: timeout (float) – timeout in seconds before giving up (default: 30)
Raises:: ConnectionFailedError – raised when connection failed or timed out

async disconnect(reason=CloseReason.UNKNOWN)

Disconnects the TCP connection. The method will not raise an exception in case the connection wasn’t yet connected

Parameters:: reason (CloseReason) – optional reason for the disconnect. This parameter is purely informational

start_reader_task(): Starts the message reader task

stop_reader_task(): Stops the message reader task if it exists

async receive_message()

Receives a single raw message

Returns:: The message as a bytes object. None if the connection returned EOF
Return type:: bytes | None

async receive_message_object()

Receives a single message and parses it to a MessageDataclass

Returns:: The parsed message, None if the connection returned EOF
Return type:: MessageDataclass | None

async receive_until_eof(raise_exception=True)

Receives data until the other end closes the connection. If raise_exception parameter is set to True other read errors are treated as EOF as well

Parameters:: raise_exception (bool)
Return type:: bytes | None

queue_message(message)

Parameters:: message (bytes | MessageDataclass)
Return type:: Task

queue_messages(*messages)

Parameters:: messages (bytes | MessageDataclass)
Return type:: list[Task]

async send_message(message)

Sends a message or a set of bytes over the connection. In case an object of MessageDataClass is provided the object will first be serialized. If the obfuscated flag is set for the connection the message or bytes will first be obfuscated

Parameters:: message (bytes | MessageDataclass) – message to be sent over the connection
Raises:: ConnectionWriteError – error or timeout occured during writing

encode_message_data(message)

Serializes the MessageDataclass or bytes and obfuscates the contents. See serialize_message()

Returns:: bytes object
Raises:: MessageSerializationError – raised when serialization failed
Parameters:: message (bytes | MessageDataclass)
Return type:: bytes

decode_message_data(data)

De-obfuscates and deserializes message data into a MessageDataclass object. See deserialize_message()

Parameters:: data (bytes) – message bytes to decode
Returns:: object of MessageDataclass
Raises:: MessageDeserializationError – raised when deserialization failed
Return type:: MessageDataclass

abstractmethod deserialize_message(message_data)

Should be called after a full message has been received. This method should parse the message

Parameters:: message_data (bytes)
Return type:: MessageDataclass

serialize_message(message)

Parameters:: message (bytes | MessageDataclass)
Return type:: bytes

class aioslsk.network.connection.ServerConnection(hostname, port, network, obfuscated=False, read_timeout=600)

Parameters:

hostname (str)
port (int)
network (Network)
obfuscated (bool)
read_timeout (float)

async connect(timeout=30)

Opens a TCP connection the hostname:ip provided in this object. The state of the connection will be changed to CONNECTING before the attempt is made

Upon success the state will be set to CONNECTED and the message reader loop will be started. Upon failure the state will be set to CLOSED with CONNECT_FAILED as its reason

Parameters:: timeout (float) – timeout in seconds before giving up (default: 30)
Raises:: ConnectionFailedError – raised when connection failed or timed out

deserialize_message(message_data)

Should be called after a full message has been received. This method should parse the message

Parameters:: message_data (bytes)
Return type:: MessageDataclass

class aioslsk.network.connection.PeerConnection(hostname, port, network, obfuscated=False, username=None, connection_type='P', incoming=False, read_timeout=60, transfer_read_timeout=180)

Parameters:

hostname (str)
port (int)
network (Network)
obfuscated (bool)
username (Optional[str])
connection_type (str)
incoming (bool)
read_timeout (float)
transfer_read_timeout (float)

transfer_read_timeout: float

incoming: bool

connection_type: str

username: str | None

download_rate_limiter: RateLimiter

upload_rate_limiter: RateLimiter

async connect(timeout=10)

Opens a TCP connection the hostname:ip provided in this object. The state of the connection will be changed to CONNECTING before the attempt is made

Upon success the state will be set to CONNECTED and the message reader loop will be started. Upon failure the state will be set to CLOSED with CONNECT_FAILED as its reason

Parameters:: timeout (float) – timeout in seconds before giving up (default: 30)
Raises:: ConnectionFailedError – raised when connection failed or timed out

set_connection_state(state)

Sets the current connection state.

If the current state is AWAITING_INIT and the connection type is a distributed or file connection the obfuscated flag for this connection will be set to False

If the state goes to the ESTABLISHED state the message reader task will be started. In all other cases it will be stopped (in case the task was running)

Parameters:: state (PeerConnectionState) – The new state of the connection

async receive_transfer_ticket()

Receive the transfer ticket from the connection

Return type:: int

async receive_transfer_offset()

Receive the transfer offset from the connection

Return type:: int

async receive_data(n_bytes)

Receives data on the connection. The n_bytes indicates the max amount of bytes that should be received on the connection, less bytes can be returned.

In case of error the socket will be disconnected. If no data is received it is assumed to be EOF and the connection will be disconnected.

Parameters:: n_bytes (int) – max amount of bytes to receive
Raises:: ConnectionReadError – in case timeout occured on an error on the socket
Returns:: bytes object containing the received data or None in case the connection reached EOF
Return type:: bytes | None

async receive_file(file_handle, filesize, callback=None)

Receives a file on the current connection and writes it to the given file_handle

This method will attempt to keep reading data until EOF is received from the connection or the amount of received bytes has reached the filesize

Parameters:

file_handle (aiofiles.threadpool.binary.AsyncBufferedIOBase) – a file handle to write the received data to
filesize (int) – expected filesize
callback (Callable[[bytes], None] | None) – optional callback that gets called each time a chunk of data is received

async send_data(data)

Parameters:: data (bytes)

async send_file(file_handle, callback=None)

Sends a file over the connection. This method makes use of the upload_rate_limiter to limit how many bytes are being sent at a time

Parameters:

file_handle (aiofiles.threadpool.binary.AsyncBufferedReader) – binary opened file handle
callback (Callable[[bytes], None] | None) – progress callback that gets called each time a chunk of data was sent

deserialize_message(message_data)

Should be called after a full message has been received. This method should parse the message

Parameters:: message_data (bytes)
Return type:: MessageDataclass

aioslsk.network.network

aioslsk.network.network.task_counter(): Implement next(self).

class aioslsk.network.network.ListeningConnectionErrorMode(*values)

Error mode for listening connections. During initialization of the network the connect_listening_connections method will raise an error depending on the error mode

ALL = 'all': Raise an exception if all listening connections failed to connect

ANY = 'any': Raise an exception if any listening connections failed to connect

CLEAR = 'clear': Raise an exception only if the non-obfuscated connection failed to connect

class aioslsk.network.network.PeerConnectMode(*values)

Connection mode to use when connecting to peers

FALLBACK = 'fallback': Attempt direct connection first, fallback on indirect connection

RACE = 'race': Attempt both direct and indirect connection at the same time. First to connect will be used and the other will be disconnected

class aioslsk.network.network.ExpectedResponse(connection_class, message_class, peer=None, fields=None, loop=None)

Future for an expected response message

Parameters:

connection_class (type[Union[PeerConnection, ServerConnection]])
message_class (type[MessageDataclass])
peer (Optional[str])
fields (Optional[dict[str, Any]])
loop (Optional[asyncio.AbstractEventLoop])

connection_class: type[PeerConnection | ServerConnection]

message_class: type[MessageDataclass]

peer: str | None

fields: dict[str, Any]

matches(connection, response)

Parameters:

connection (DataConnection)
response (MessageDataclass)

Return type:

class aioslsk.network.network.PeerFuture(ticket, username, typ, loop=None)

Parameters:

ticket (int)
username (str)
typ (str)
loop (Optional[asyncio.AbstractEventLoop])

ticket: int

username: str

typ: str

class aioslsk.network.network.WatchdogContext(last_state: 'ConnectionState')

Parameters:: last_state (ConnectionState)

last_state: ConnectionState

class aioslsk.network.network.Network(settings, event_bus)

Parameters:

settings (Settings)
event_bus (EventBus)

server_connection: ServerConnection

listening_connections: ListeningConnections

peer_connections: list[PeerConnection]

register_listeners()

create_server_connection()

Return type:: ServerConnection

create_listening_connections()

Return type:: tuple[ListeningConnection | None, ListeningConnection | None]

async initialize(): Initializes the server and listening connections

async connect_listening_ports()

This method will attempt to connect both listening ports (if configured) but will raise an exception depending on the settings network.listening.error_mode setting. All other listening connections will be disconnected before raising the error

Raises:: ListeningConnectionFailedError – if an error occurred connecting the listening ports

async disconnect_listening_ports(): Disconnects all listening ports

async connect_server()

async disconnect_server()

async disconnect(): Cancels all connection creation tasks and disconnects all current open connections

get_listening_ports()

Gets the currently connected listening ports

Returns:: tuple with 2 elements: the non-obfuscated- and obfuscated ports. If either is not connected or not configured 0 will be returned for this port
Return type:: tuple[int, int]

async advertise_listening_ports(): Notifies the server of our current listening ports

load_speed_limits(): (Re)loads the speed limits from the settings

set_upload_speed_limit(limit_kbps)

Modifies the upload speed limit. Passing 0 will set the upload speed to unlimited

Parameters:: limit_kbps (int) – the new upload limit

set_download_speed_limit(limit_kbps)

Modifies the download speed limit. Passing 0 will set the download speed to unlimited

Parameters:: limit_kbps (int) – the new download limit

async start_server_connection_watchdog(): Starts the server connection watchdog if it is not already running

stop_server_connection_watchdog(): Stops the server connection watchdog if it is running

async start_upnp_job()

stop_upnp_job()

select_port(port, obfuscated_port)

Selects the port used for making a connection. This attempts to take into account the network.peer.obfuscate parameter however falls back on using whatever port is available if necessary

Parameters:

port – available non-obfuscated port
obfuscated_port – available obfuscated port

Returns:

a tuple with the port number and whether the port is obfuscated

Return type:

tuple[int, bool]

async create_peer_connection(username, typ, ip=None, port=None, obfuscate=False)

Creates a new peer connection to the given username and connection type.

Optionally this method takes ip, port and obfuscate parameters. If not given it will be requested to the server (GetPeerAddress) before attempting connection

Parameters:

username (str) – username of the peer
typ (str) – connection type (‘P’, ‘D’, ‘F’)
ip (str | None) – optional ip address of the peer
port (int | None) – optional port of the peer
obfuscate (bool) – whether to obfuscate the connection. Should be used in combination with the ip and port

Returns:

an object of PeerConnection

Raises:

PeerConnectionError – when establishing a peer connection has failed

Return type:

PeerConnection

async get_peer_connection(username, typ='P')

Gets a peer connection for the given username. It will first try to re-use an existing connection, otherwise it will create a new connection

Parameters:

username (str) – username of the peer to connect to
typ (str) – Type of connection to create or reuse

Returns:

A PeerConnection instance

Return type:

PeerConnection

create_server_response_future(message_class, fields=None)

Creates a future for a server message to arrive, the message must match the message_class and fields defined in the keyword arguments.

The future will be stored by the current class and will be removed once the future is complete or cancelled.

Parameters:

message_class (type[T]) – The expected MessageDataClass
fields (dict[str, Any] | None)

Returns:

ExpectedResponse object

Return type:

ExpectedResponse

register_response_future(expected_response)

Registers an expected response future

Parameters:: expected_response (ExpectedResponse)

async wait_for_server_message(message_class, fields=None, timeout=10)

Waits for a message from the server

Parameters:

message_class (type[T]) – Class of the expected server message
fields (dict[str, Any] | None) – Optional matchers for the message fields
timeout (float)

Returns:

The MessageData object corresponding to the response

Raise:

TimeoutError

Return type:

T

create_peer_response_future(peer, message_class, fields=None)

Creates a future for a peer message to arrive, the message must match the message_class and fields defined in the keyword arguments and must be coming from a connection by peer.

The future will be stored by the current class and will be removed once the future is complete or cancelled.

Parameters:

peer (str) – name of the peer for which the message should be received
message_class (type[T])
fields (dict[str, Any] | None)

Returns:

ExpectedResponse object

Return type:

ExpectedResponse

async wait_for_peer_message(peer, message_class, fields=None, timeout=60)

Parameters:

peer (str)
message_class (type[T])
fields (dict[str, Any] | None)
timeout (float)

Return type:

T

get_peer_connections(username, typ)

Returns all connections for peer with given username and peer connection types.

Parameters:

username (str) – username of the peer
typ (str) – type of the connection (from PeerConnectionType)

Returns:

list of connections for that matches the parameters

Return type:

list[PeerConnection]

get_active_peer_connections(username, typ)

Return a list of currently active messaging connections for given peer connection type.

Parameters:

username (str) – username for which to get the active peer
typ (str) – peer connection type

Returns:

list of PeerConnection instances

Return type:

list[PeerConnection]

remove_peer_connection(connection)

Parameters:: connection (PeerConnection)

async send_peer_messages(username, *messages, raise_on_error=True)

Sends a list of messages to the peer with given username. This uses get_peer_connection and will attempt to re-use a connection or create a new peer (P) connection

Parameters:

username (str) – Peer username to send the messages to
messages (bytes | MessageDataclass) – List of messages to send
raise_on_error (bool) – When True an exception is raised when a message failed to send, if False a list of tuples with the result for each message will be returned

Returns:

a list of tuples containing the sent messages and the result of the sent messages. If None is returned for a message it was successfully sent, otherwise the result will contain the exception. None if the raise_on_error is True

queue_server_messages(*messages)

Queues server messages

Parameters:: messages (bytes | MessageDataclass) – list of messages to queue
Return type:: list[Task]

async send_server_messages(*messages, raise_on_error=True)

Sends a list of messages to the server

Parameters:

messages (bytes | MessageDataclass) – List of messages to send
raise_on_error (bool) – When True an exception is raised when a message failed to send, if False a list of tuples with the result for each message will be returned

Returns:

a list of tuples containing the sent messages and the result of the sent messages. If None is returned for a message it was successfully sent, otherwise the result will contain the exception. None if the raise_on_error is True

async on_state_changed(state, connection, close_reason=CloseReason.UNKNOWN)

Called when the state of a connection changes. This method calls 3 private method based on the type of connection that was passed

Parameters:

state (ConnectionState) – the new state the connection has received
connection (Connection) – the connection for which the state changed
close_reason (CloseReason) – in case ConnectionState.CLOSED is passed a reason will be given as well

async on_peer_accepted(connection)

Called when a connection has been accepted on one of the listening connections. This method will wait for the peer initialization message: either PeerInit or PeerPierceFirewall.

In case of PeerInit the method will perform the initialization of the connection.

In case of PeerPierceFirewall we look for an associated future object for the ticket provided in the message; there should be a task waiting for it to be completed (see _make_indirect_connection). Full initialization of the connection should be done in that method.

In any other case we disconnect the connection.

Parameters:: connection (PeerConnection) – the accepted PeerConnection

async on_message_received(message, connection)

Method called by connection instances when a message is received

Parameters:

message (MessageDataclass) – Received message instance
connection (DataConnection) – Connection on which the message was received

aioslsk.protocol.obfuscation

aioslsk.protocol.obfuscation.KEY_SIZE = 4

Variables:: KEY_SIZE – Amount of bytes in the key

aioslsk.protocol.obfuscation.generate_key()

Return type:: bytes

aioslsk.protocol.obfuscation.rotate_key(key, rot_bits=31)

Rotate the key to the right by const bits

Parameters:

key (bytes) – Key to rotate
rot_bits (int) – Amount of bits to rotate

Returns:

The rotated key

Return type:

bytes

aioslsk.protocol.obfuscation.decode(data)

De-obfuscate given data, the key should be the first 4 bytes of the data

Parameters:: data (bytes) – Data to be de-obfuscated
Returns:: De-obfuscated data
Return type:: bytes

aioslsk.protocol.obfuscation.encode(data, key=None)

Obfuscate the given data with the provided key, if no key is given it will be automatically generated

Parameters:

data (bytes)
key (bytes | None)

Return type:

bytes

aioslsk.protocol.primitives

Module defining all data primitives used in the protocol messages

Field metadata:

During (de)serialization the metadata parameter of the dataclasses.field function to control how to perform (de)serialization.

These metadata keys are implemented:

‘type’: <type_class>
- defines the primary type of the data
‘subtype’: <type_class>
- used for arrays : the type of the elements contained in the array
‘if_true’: <field_name>
- serialization : only pack this field if the value of field with name <field_name> evaluates to True
- deserialization : only parse this field if the value of field with name <field_name> evaluates to True
‘if_false’: <field_name>
- serialization : only pack this field if the value of field with name <field_name> evaluates to False
- deserialization : only parse this field if the value of field with name <field_name> evaluates to False
‘optional’: True
- serialization : only pack this field if its value is anything other than None
- deserialization : during deserialization the code will determine if the message has been fully parsed. If not it will parse this field

class aioslsk.protocol.primitives.Serializable(*args, **kwargs)

Parameters:

args (Any)
kwargs (Any)

classmethod deserialize(pos, data)

Parameters:

pos (int)
data (bytes)

Return type:

tuple[int, Self]

serialize(*args, **kwargs)

Return type:: bytes

serialize_into(buffer, *args, **kwargs)

Parameters:: buffer (bytearray)

class aioslsk.protocol.primitives.AttributeKey(*values)

BITRATE = 0

DURATION = 1

VBR = 2

SAMPLE_RATE = 4

BIT_DEPTH = 5

aioslsk.protocol.primitives.decode_string(value)

Parameters:: value (bytes)
Return type:: str

class aioslsk.protocol.primitives.uint8

STRUCT = <_struct.Struct object>

serialize()

Return type:: bytes

serialize_into(buffer)

Parameters:: buffer (bytearray)

classmethod deserialize(pos, data)

Parameters:

pos (int)
data (bytes)

Return type:

class aioslsk.protocol.primitives.uint16

STRUCT = <_struct.Struct object>

serialize()

Return type:: bytes

serialize_into(buffer)

Parameters:: buffer (bytearray)

classmethod deserialize(pos, data)

Parameters:

pos (int)
data (bytes)

Return type:

class aioslsk.protocol.primitives.uint32

STRUCT = <_struct.Struct object>

serialize()

Return type:: bytes

serialize_into(buffer)

Parameters:: buffer (bytearray)

classmethod deserialize(pos, data)

Parameters:

pos (int)
data (bytes)

Return type:

class aioslsk.protocol.primitives.uint64

STRUCT = <_struct.Struct object>

serialize()

Return type:: bytes

serialize_into(buffer)

Parameters:: buffer (bytearray)

classmethod deserialize(pos, data)

Parameters:

pos (int)
data (bytes)

Return type:

class aioslsk.protocol.primitives.int32

STRUCT = <_struct.Struct object>

serialize()

Return type:: bytes

serialize_into(buffer)

Parameters:: buffer (bytearray)

classmethod deserialize(pos, data)

Parameters:

pos (int)
data (bytes)

Return type:

class aioslsk.protocol.primitives.string

serialize(encoding='utf-8')

Parameters:: encoding (str)
Return type:: bytes

serialize_into(buffer, encoding='utf-8')

Parameters:

buffer (bytearray)
encoding (str)

classmethod deserialize(pos, data)

Parameters:

pos (int)
data (bytes)

Return type:

tuple[int, str]

class aioslsk.protocol.primitives.bytearr

serialize()

Return type:: bytes

serialize_into(buffer)

Parameters:: buffer (bytearray)

classmethod deserialize(pos, data)

Parameters:

pos (int)
data (bytes)

Return type:

tuple[int, bytes]

class aioslsk.protocol.primitives.ipaddr

STRUCT = <_struct.Struct object>

serialize()

Return type:: bytes

serialize_into(buffer)

Parameters:: buffer (bytearray)

classmethod deserialize(pos, data)

Parameters:

pos (int)
data (bytes)

Return type:

tuple[int, str]

class aioslsk.protocol.primitives.boolean

STRUCT = <_struct.Struct object>

serialize()

Return type:: bytes

serialize_into(buffer)

Parameters:: buffer (bytearray)

classmethod deserialize(pos, data)

Parameters:

pos (int)
data (bytes)

Return type:

tuple[int, bool]

class aioslsk.protocol.primitives.array(iterable=(), /)

serialize(element_type)

Parameters:: element_type (type[Serializable])
Return type:: bytes

serialize_into(buffer, element_type)

Parameters:

buffer (bytearray)
element_type (type[Serializable])

classmethod deserialize(pos, data, element_type)

Parameters:

pos (int)
data (bytes)
element_type (type[T])

Return type:

tuple[int, list[T]]

class aioslsk.protocol.primitives.ProtocolDataclass

The ProtocolDataclass defines a collection of primitives that can be serialized or deserialized. Classes inheriting from this class should use the @dataclass(order=True) decorator. The order needs to be kept as the fields definitions will be evaluated during (de)serialization.

Example definition:

@dataclass(order=True)
class CustomDataclass(ProtocolDataclass):
    username: str = field(metadata={'type': string})
    password: str = field(metadata={'type': string})
    has_privileges: bool = field(metadata={'type': boolean})

serialize()

Return type:: bytes

serialize_into(buffer)

Parameters:: buffer (bytearray)

classmethod deserialize(pos, message)

Parameters:

pos (int)
message (bytes)

class aioslsk.protocol.primitives.MessageDataclass

Message data class for which protocol messages should inherit from. This takes all behaviour from the ProtocolDataclass class but adds:

Prepending the message with length and MESSAGE_ID
Optionally the message data will (de)compressed

MESSAGE_ID: ClassVar[uint8 | uint32] = 0

serialize(compress=False)

Serializes the current MessageDataClass object and prepends the message length and MESSAGE_ID

In case the message needs to be compressed just override this method in the subclass and simply call the super method with compress=True

Parameters:: compress (bool) – use gzip compression on the message contents
Return type:: bytes

serialize_into(buffer, compress=False)

Parameters:

buffer (bytearray)
compress (bool)

classmethod deserialize(pos, message, decompress=False)

Deserializes the passed message into an object of the current type

In case the message needs to be decompressed just override this method in the subclass and simply call the super method with decompress=True

Parameters:

decompress (bool) – use gzip decompression on the message contents
pos (int)
message (bytes)

Raises:

ValueError – if the message_id found the data does not match the MESSAGE_ID defined in the current class

Returns:

an object of the current class

Return type:

Self

class aioslsk.protocol.primitives.Attribute(key: int, value: int)

Parameters:

key (int)
value (int)

key: int

value: int

classmethod deserialize(pos, message)

Parameters:

pos (int)
message (bytes)

serialize()

Return type:: bytes

serialize_into(buffer)

Parameters:: buffer (bytearray)

class aioslsk.protocol.primitives.SimilarUser(username: str, score: int)

Parameters:

username (str)
score (int)

username: str

score: int

class aioslsk.protocol.primitives.Recommendation(recommendation: str, score: int)

Parameters:

recommendation (str)
score (int)

recommendation: str

score: int

class aioslsk.protocol.primitives.RoomTicker(username: str, ticker: str)

Parameters:

username (str)
ticker (str)

username: str

ticker: str

class aioslsk.protocol.primitives.PotentialParent(username: str, ip: str, port: int)

Parameters:

username (str)
ip (str)
port (int)

username: str

ip: str

port: int

class aioslsk.protocol.primitives.UserStats(avg_speed: int, uploads: int, shared_file_count: int, shared_folder_count: int)

Parameters:

avg_speed (int)
uploads (int)
shared_file_count (int)
shared_folder_count (int)

avg_speed: int

uploads: int

shared_file_count: int

shared_folder_count: int

class aioslsk.protocol.primitives.FileData(unknown: int, filename: str, filesize: int, extension: str, attributes: list[aioslsk.protocol.primitives.Attribute])

Parameters:

unknown (int)
filename (str)
filesize (int)
extension (str)
attributes (list[Attribute])

unknown: int

filename: str

filesize: int

extension: str

attributes: list[Attribute]

classmethod deserialize(pos, message)

Parameters:

pos (int)
message (bytes)

serialize()

Return type:: bytes

serialize_into(buffer)

Parameters:: buffer (bytearray)

get_attribute_map()

Converts the attribute list to a dictionary. The resulting dictionary will only contain known attributes and attributes present in the list

Return type:: dict[AttributeKey, int]

class aioslsk.protocol.primitives.DirectoryData(name: str, files: list[aioslsk.protocol.primitives.FileData])

Parameters:

name (str)
files (list[FileData])

name: str

files: list[FileData]

classmethod deserialize(pos, message)

Parameters:

pos (int)
message (bytes)

serialize()

Return type:: bytes

aioslsk.protocol.primitives.has_unparsed_bytes(pos, message)

Parameters:

pos (int)
message (bytes)

Return type: