SyncTask.abort() - Code Metrics - Inspection of "Bumped version to 1.1.0.2-beta" - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — beta ( 72a57d...7d0ef0 )

by Dean

created 2016-06-12 12:37 UTC

SyncTask.abort() A

↳ Parent: SyncTask

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	14.7187

Importance

Changes

Metric	Value
dl	0
loc	13
ccs	1
cts	8
cp	0.125
rs	9.2
c	0
b	0
f	0
cc	4
crap	14.7187

from plugin.managers.exception import ExceptionManager
from plugin.models import *
from plugin.sync.core.enums import SyncData, SyncMode
from plugin.sync.core.exceptions import SyncAbort
from plugin.sync.core.task.artifacts import SyncArtifacts
from plugin.sync.core.task.configuration import SyncConfiguration
from plugin.sync.core.task.map import SyncMap
from plugin.sync.core.task.progress import SyncProgress
from plugin.sync.core.task.profiler import SyncProfiler
from plugin.sync.core.task.state import SyncState

from datetime import datetime
from peewee import JOIN_LEFT_OUTER
import logging
import time

log = logging.getLogger(__name__)


class SyncTask(object):
    def __init__(self, account, mode, data, media, result, status, **kwargs):
        self.account = account

        # Sync options
        self.mode = mode
        self.data = data
        self.media = media

        # Extra arguments
        self.kwargs = kwargs

        # Handlers/Modes for task
        self.handlers = None
        self.modes = None

        # State/Result management
        self.result = result
        self.status = status

        self.exceptions = []

        self.finished = False
        self.started = False
        self.success = None

        self._abort = False

        # Construct children
        self.artifacts = SyncArtifacts(self)
        self.configuration = SyncConfiguration(self)
        self.map = SyncMap(self)
        self.progress = SyncProgress(self)
        self.profiler = SyncProfiler(self)

        self.state = SyncState(self)

    @property
    def id(self):
        if self.result is None:
            return None

        return self.result.id

    @property
    def elapsed(self):
        if self.result is None:
            return None

        return (datetime.utcnow() - self.result.started_at).total_seconds()

    def construct(self, handlers, modes):
        log.debug('Constructing %d handlers...', len(handlers))
        self.handlers = dict(self._construct_modules(handlers, 'data'))

        log.debug('Constructing %d modes...', len(modes))
        self.modes = dict(self._construct_modules(modes, 'mode'))

    def load(self):
        # Load task configuration
        self.configuration.load(self.account)

        # Automatically determine enabled data types
        if self.data is None:
            self.data = self.get_enabled_data(self.configuration, self.mode)

        log.debug('Sync Data: %r', self.data)
        log.debug('Sync Media: %r', self.media)

        if self.data is None:
            raise ValueError('No collections enabled for sync')

        # Load children
        self.profiler.load()
        self.state.load()

    def abort(self, timeout=None):
        # Set `abort` flag, thread will abort on the next `checkpoint()`
        self._abort = True

        if timeout is None:
            return

        # Wait `timeout` seconds for task to finish
        for x in xrange(timeout):
            if self.finished:
                return

            time.sleep(1)

    def checkpoint(self):
        # Check if an abort has been requested
        if not self._abort:
            return

        raise SyncAbort()

    def finish(self):
        # Update result in database
        self.result.ended_at = datetime.utcnow()
        self.result.success = self.success
        self.result.save()

        # Store exceptions in database
        for exc_info in self.exceptions:
            try:
                self.store_exception(self.result, exc_info)
            except Exception, ex:
                log.warn('Unable to store exception: %s', str(ex), exc_info=True)

        # Flush caches to archives
        self.state.flush()

        # Display profiler report
        self.profiler.log_report()

        # Mark finished
        self.finished = True

    @staticmethod
    def store_exception(result, exc_info):
        exception, error = ExceptionManager.create.from_exc_info(exc_info)

        # Link error to result
        SyncResultError.create(
            result=result,
            error=error
        )

        # Link exception to result
        SyncResultException.create(
            result=result,
            exception=exception
        )

    @classmethod
    def create(cls, account, mode, data, media, trigger, **kwargs):
        # Get account
        if type(account) is int:
            account = cls.get_account(account)
        elif type(account) is not Account:
            raise ValueError('Unexpected value provided for the "account" parameter')

        # Get/Create sync status
        status, created = SyncStatus.get_or_create(
            account=account,
            mode=mode,
            section=kwargs.get('section', None)
        )

        # Create sync result
        result = SyncResult.create(
            status=status,
            trigger=trigger,

            started_at=datetime.utcnow()
        )

        # Create sync task
        task = SyncTask(
            account, mode,
            data, media,
            result, status,
            **kwargs
        )

        # Load sync configuration/state
        task.load()

        return task

    @classmethod
    def get_account(cls, account_id):
        # TODO Move account retrieval/join to `Account` class
        return (
            Account.select(
                Account.id,
                Account.name,

                PlexAccount.id,
                PlexAccount.key,
                PlexAccount.username,
                PlexBasicCredential.token_plex,
                PlexBasicCredential.token_server,

                TraktAccount.username,
                TraktBasicCredential.token,

                TraktOAuthCredential.access_token,
                TraktOAuthCredential.refresh_token,
                TraktOAuthCredential.created_at,
                TraktOAuthCredential.expires_in
            )
            # Plex
            .join(
                PlexAccount, JOIN_LEFT_OUTER, on=(
                    PlexAccount.account == Account.id
                ).alias('plex')
            )
            .join(
                PlexBasicCredential, JOIN_LEFT_OUTER, on=(
                    PlexBasicCredential.account == PlexAccount.id
                ).alias('basic')
            )
            # Trakt
            .switch(Account)
            .join(
                TraktAccount, JOIN_LEFT_OUTER, on=(
                    TraktAccount.account == Account.id
                ).alias('trakt')
            )
            .join(
                TraktBasicCredential, JOIN_LEFT_OUTER, on=(
                    TraktBasicCredential.account == TraktAccount.id
                ).alias('basic')
            )
            .switch(TraktAccount)
            .join(
                TraktOAuthCredential, JOIN_LEFT_OUTER, on=(
                    TraktOAuthCredential.account == TraktAccount.id
                ).alias('oauth')
            )
            .where(Account.id == account_id)
            .get()
        )

    @classmethod
    def get_enabled_data(cls, config, mode):
        # Determine accepted modes
        modes = [SyncMode.Full]

        if mode == SyncMode.Full:
            modes.extend([
                SyncMode.FastPull,
                SyncMode.Pull,
                SyncMode.Push
            ])
        elif mode == SyncMode.FastPull:
            modes.extend([
                mode,
                SyncMode.Pull
            ])
        else:
            modes.append(mode)

        # Retrieve enabled data
        enabled = []

        if config['sync.watched.mode'] in modes:
            enabled.append(SyncData.Watched)

        if config['sync.ratings.mode'] in modes:
            enabled.append(SyncData.Ratings)

        if config['sync.playback.mode'] in modes:
            enabled.append(SyncData.Playback)

        if config['sync.collection.mode'] in modes:
            enabled.append(SyncData.Collection)

        # Lists
        if config['sync.lists.watchlist.mode'] in modes:
            enabled.append(SyncData.Watchlist)

        if config['sync.lists.liked.mode'] in modes:
            enabled.append(SyncData.Liked)

        if config['sync.lists.personal.mode'] in modes:
            enabled.append(SyncData.Personal)

        # Convert to enum value
        result = None

        for data in enabled:
            if result is None:
                result = data
                continue

            result |= data

        return result

    def _construct_modules(self, modules, attribute):
        for cls in modules:
            keys = getattr(cls, attribute, None)

            if keys is None:
                log.warn('Module %r is missing a valid %r attribute', cls, attribute)
                continue

            # Convert `keys` to list
            if type(keys) is not list:
                keys = [keys]

            # Construct module
            obj = cls(self)

            # Return module with keys
            for key in keys:
                yield key, obj


1	1	from plugin.managers.exception import ExceptionManager
2	1	from plugin.models import *
3	1	from plugin.sync.core.enums import SyncData, SyncMode
4	1	from plugin.sync.core.exceptions import SyncAbort
5	1	from plugin.sync.core.task.artifacts import SyncArtifacts
6	1	from plugin.sync.core.task.configuration import SyncConfiguration
7	1	from plugin.sync.core.task.map import SyncMap
8	1	from plugin.sync.core.task.progress import SyncProgress
9	1	from plugin.sync.core.task.profiler import SyncProfiler
10	1	from plugin.sync.core.task.state import SyncState
11
12	1	from datetime import datetime
13	1	from peewee import JOIN_LEFT_OUTER
14	1	import logging
15	1	import time
16
17	1	log = logging.getLogger(__name__)
18
19
20	1	class SyncTask(object):
21	1	def __init__(self, account, mode, data, media, result, status, **kwargs):
22	1	self.account = account
23
24		# Sync options
25	1	self.mode = mode
26	1	self.data = data
27	1	self.media = media
28
29		# Extra arguments
30	1	self.kwargs = kwargs
31
32		# Handlers/Modes for task
33	1	self.handlers = None
34	1	self.modes = None
35
36		# State/Result management
37	1	self.result = result
38	1	self.status = status
39
40	1	self.exceptions = []
41
42	1	self.finished = False
43	1	self.started = False
44	1	self.success = None
45
46	1	self._abort = False
47
48		# Construct children
49	1	self.artifacts = SyncArtifacts(self)
50	1	self.configuration = SyncConfiguration(self)
51	1	self.map = SyncMap(self)
52	1	self.progress = SyncProgress(self)
53	1	self.profiler = SyncProfiler(self)
54
55	1	self.state = SyncState(self)
56
57	1	@property
58		def id(self):
59		if self.result is None:
60		return None
61
62		return self.result.id
63
64	1	@property
65		def elapsed(self):
66		if self.result is None:
67		return None
68
69		return (datetime.utcnow() - self.result.started_at).total_seconds()
70
71	1	def construct(self, handlers, modes):
72		log.debug('Constructing %d handlers...', len(handlers))
73		self.handlers = dict(self._construct_modules(handlers, 'data'))
74
75		log.debug('Constructing %d modes...', len(modes))
76		self.modes = dict(self._construct_modules(modes, 'mode'))
77
78	1	def load(self):
79		# Load task configuration
80		self.configuration.load(self.account)
81
82		# Automatically determine enabled data types
83		if self.data is None:
84		self.data = self.get_enabled_data(self.configuration, self.mode)
85
86		log.debug('Sync Data: %r', self.data)
87		log.debug('Sync Media: %r', self.media)
88
89		if self.data is None:
90		raise ValueError('No collections enabled for sync')
91
92		# Load children
93		self.profiler.load()
94		self.state.load()
95
96	1	def abort(self, timeout=None):
97		# Set `abort` flag, thread will abort on the next `checkpoint()`
98		self._abort = True
99
100		if timeout is None:
101		return
102
103		# Wait `timeout` seconds for task to finish
104		for x in xrange(timeout):
105		if self.finished:
106		return
107
108		time.sleep(1)
109
110	1	def checkpoint(self):
111		# Check if an abort has been requested
112		if not self._abort:
113		return
114
115		raise SyncAbort()
116
117	1	def finish(self):
118		# Update result in database
119		self.result.ended_at = datetime.utcnow()
120		self.result.success = self.success
121		self.result.save()
122
123		# Store exceptions in database
124		for exc_info in self.exceptions:
125		try:
126		self.store_exception(self.result, exc_info)
127		except Exception, ex:
128		log.warn('Unable to store exception: %s', str(ex), exc_info=True)
129
130		# Flush caches to archives
131		self.state.flush()
132
133		# Display profiler report
134		self.profiler.log_report()
135
136		# Mark finished
137		self.finished = True
138
139	1	@staticmethod
140		def store_exception(result, exc_info):
141		exception, error = ExceptionManager.create.from_exc_info(exc_info)
142
143		# Link error to result
144		SyncResultError.create(
145		result=result,
146		error=error
147		)
148
149		# Link exception to result
150		SyncResultException.create(
151		result=result,
152		exception=exception
153		)
154
155	1	@classmethod
156		def create(cls, account, mode, data, media, trigger, **kwargs):
157		# Get account
158		if type(account) is int:
159		account = cls.get_account(account)
160		elif type(account) is not Account:
161		raise ValueError('Unexpected value provided for the "account" parameter')
162
163		# Get/Create sync status
164		status, created = SyncStatus.get_or_create(
165		account=account,
166		mode=mode,
167		section=kwargs.get('section', None)
168		)
169
170		# Create sync result
171		result = SyncResult.create(
172		status=status,
173		trigger=trigger,
174
175		started_at=datetime.utcnow()
176		)
177
178		# Create sync task
179		task = SyncTask(
180		account, mode,
181		data, media,
182		result, status,
183		**kwargs
184		)
185
186		# Load sync configuration/state
187		task.load()
188
189		return task
190
191	1	@classmethod
192		def get_account(cls, account_id):
193		# TODO Move account retrieval/join to `Account` class
194		return (
195		Account.select(
196		Account.id,
197		Account.name,
198
199		PlexAccount.id,
200		PlexAccount.key,
201		PlexAccount.username,
202		PlexBasicCredential.token_plex,
203		PlexBasicCredential.token_server,
204
205		TraktAccount.username,
206		TraktBasicCredential.token,
207
208		TraktOAuthCredential.access_token,
209		TraktOAuthCredential.refresh_token,
210		TraktOAuthCredential.created_at,
211		TraktOAuthCredential.expires_in
212		)
213		# Plex
214		.join(
215		PlexAccount, JOIN_LEFT_OUTER, on=(
216		PlexAccount.account == Account.id
217		).alias('plex')
218		)
219		.join(
220		PlexBasicCredential, JOIN_LEFT_OUTER, on=(
221		PlexBasicCredential.account == PlexAccount.id
222		).alias('basic')
223		)
224		# Trakt
225		.switch(Account)
226		.join(
227		TraktAccount, JOIN_LEFT_OUTER, on=(
228		TraktAccount.account == Account.id
229		).alias('trakt')
230		)
231		.join(
232		TraktBasicCredential, JOIN_LEFT_OUTER, on=(
233		TraktBasicCredential.account == TraktAccount.id
234		).alias('basic')
235		)
236		.switch(TraktAccount)
237		.join(
238		TraktOAuthCredential, JOIN_LEFT_OUTER, on=(
239		TraktOAuthCredential.account == TraktAccount.id
240		).alias('oauth')
241		)
242		.where(Account.id == account_id)
243		.get()
244		)
245
246	1	@classmethod
247		def get_enabled_data(cls, config, mode):
248		# Determine accepted modes
249		modes = [SyncMode.Full]
250
251		if mode == SyncMode.Full:
252		modes.extend([
253		SyncMode.FastPull,
254		SyncMode.Pull,
255		SyncMode.Push
256		])
257		elif mode == SyncMode.FastPull:
258		modes.extend([
259		mode,
260		SyncMode.Pull
261		])
262		else:
263		modes.append(mode)
264
265		# Retrieve enabled data
266		enabled = []
267
268		if config['sync.watched.mode'] in modes:
269		enabled.append(SyncData.Watched)
270
271		if config['sync.ratings.mode'] in modes:
272		enabled.append(SyncData.Ratings)
273
274		if config['sync.playback.mode'] in modes:
275		enabled.append(SyncData.Playback)
276
277		if config['sync.collection.mode'] in modes:
278		enabled.append(SyncData.Collection)
279
280		# Lists
281		if config['sync.lists.watchlist.mode'] in modes:
282		enabled.append(SyncData.Watchlist)
283
284		if config['sync.lists.liked.mode'] in modes:
285		enabled.append(SyncData.Liked)
286
287		if config['sync.lists.personal.mode'] in modes:
288		enabled.append(SyncData.Personal)
289
290		# Convert to enum value
291		result = None
292
293		for data in enabled:
294		if result is None:
295		result = data
296		continue
297
298		result \|= data
299
300		return result
301
302	1	def _construct_modules(self, modules, attribute):
303		for cls in modules:
304		keys = getattr(cls, attribute, None)
305
306		if keys is None:
307		log.warn('Module %r is missing a valid %r attribute', cls, attribute)
308		continue
309
310		# Convert `keys` to list
311		if type(keys) is not list:
312		keys = [keys]
313
314		# Construct module
315		obj = cls(self)
316
317		# Return module with keys
318		for key in keys:
319		yield key, obj
320

fuzeman / Plex-Trakt-Scrobbler

Push — beta ( 72a57d...7d0ef0 )

SyncTask.abort() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like