/usr/lib/python3/dist-packages/transmissionrpc/session.py is in python3-transmissionrpc 0.11-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | # -*- coding: utf-8 -*-
# Copyright (c) 2008-2013 Erik Svensson <erik.public@gmail.com>
# Licensed under the MIT license.
from transmissionrpc.utils import Field
from six import iteritems, integer_types
class Session(object):
"""
Session is a class holding the session data for a Transmission daemon.
Access the session field can be done through attributes.
The attributes available are the same as the session arguments in the
Transmission RPC specification, but with underscore instead of hyphen.
``download-dir`` -> ``download_dir``.
"""
def __init__(self, client=None, fields=None):
self._client = client
self._fields = {}
if fields is not None:
self._update_fields(fields)
def __getattr__(self, name):
try:
return self._fields[name].value
except KeyError:
raise AttributeError('No attribute %s' % name)
def __str__(self):
text = ''
for key in sorted(self._fields.keys()):
text += "% 32s: %s\n" % (key[-32:], self._fields[key].value)
return text
def _update_fields(self, other):
"""
Update the session data from a Transmission JSON-RPC arguments dictionary
"""
if isinstance(other, dict):
for key, value in iteritems(other):
self._fields[key.replace('-', '_')] = Field(value, False)
elif isinstance(other, Session):
for key in list(other._fields.keys()):
self._fields[key] = Field(other._fields[key].value, False)
else:
raise ValueError('Cannot update with supplied data')
def _dirty_fields(self):
"""Enumerate changed fields"""
outgoing_keys = ['peer_port', 'pex_enabled']
fields = []
for key in outgoing_keys:
if key in self._fields and self._fields[key].dirty:
fields.append(key)
return fields
def _push(self):
"""Push changed fields to the server"""
dirty = self._dirty_fields()
args = {}
for key in dirty:
args[key] = self._fields[key].value
self._fields[key] = self._fields[key]._replace(dirty=False)
if len(args) > 0:
self._client.set_session(**args)
def update(self, timeout=None):
"""Update the session information."""
self._push()
session = self._client.get_session(timeout=timeout)
self._update_fields(session)
session = self._client.session_stats(timeout=timeout)
self._update_fields(session)
def from_request(self, data):
"""Update the session information."""
self._update_fields(data)
def _get_peer_port(self):
"""
Get the peer port.
"""
return self._fields['peer_port'].value
def _set_peer_port(self, port):
"""
Set the peer port.
"""
if isinstance(port, integer_types):
self._fields['peer_port'] = Field(port, True)
self._push()
else:
raise ValueError("Not a valid limit")
peer_port = property(_get_peer_port, _set_peer_port, None, "Peer port. This is a mutator.")
def _get_pex_enabled(self):
"""Is peer exchange enabled?"""
return self._fields['pex_enabled'].value
def _set_pex_enabled(self, enabled):
"""Enable/disable peer exchange."""
if isinstance(enabled, bool):
self._fields['pex_enabled'] = Field(enabled, True)
self._push()
else:
raise TypeError("Not a valid type")
pex_enabled = property(_get_pex_enabled, _set_pex_enabled, None, "Enable peer exchange. This is a mutator.")
|