chaoswg.models.Task.set_todo() - Code Metrics - Inspection of "Schedule tasks" - Obihoernchen/ChaosWG-Manager - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#6)

by Markus

created 2018-02-08 11:43 UTC

chaoswg.models.Task.set_todo() A

↳ Parent: chaoswg.models

Complexity

Conditions

Size

Total Lines	8
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	8
nop	2
dl	0
loc	8
rs	9.4285
c	0
b	0
f	0

from datetime import datetime


from peewee import (CharField, IntegerField, DateTimeField, SmallIntegerField, ForeignKeyField,
                    FloatField, DoesNotExist, SqliteDatabase)
from playhouse.flask_utils import FlaskDB
from werkzeug.security import generate_password_hash, check_password_hash

db_wrapper = FlaskDB()



def init_database(app):
    db_wrapper._db = SqliteDatabase(app.config['DATABASE'], pragmas=(('journal_mode', 'wal'),))
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
    db_wrapper.init_app(app)
    return db_wrapper.database


def create_tables():
    db_wrapper.database.connect()
    db_wrapper.database.create_tables([User, Task, History], safe=True)
    db_wrapper.database.close()


class ModelBase(db_wrapper.Model):
    @classmethod
    def get_all(cls):
        return list(cls.select().dicts())


class User(ModelBase):
    username = CharField(unique=True)
    password = CharField()
    points = IntegerField(default=0)
    last_update = DateTimeField(default=datetime.utcnow)

    @classmethod
    def get_all(cls):
        """
        without password
        :return:
        """
        return list(cls.select(cls.username, cls.points, cls.last_update).dicts())

    @classmethod
    def get_by_name(cls, username):
        try:
            return cls.get(cls.username == username)
        except DoesNotExist:
            return False

    @classmethod
    def get_usernames(cls):
        query = cls.select(cls.username)
        result = set()
        for user in query:
            result.add(user.username)
        return result

    def set_password(self, password):
        self.password = generate_password_hash(password)

    def check_password(self, password):
        return check_password_hash(self.password, password)

    # Flask-Login required functions
    @property
    def is_active(self):
        return True

    @property
    def is_authenticated(self):
        return True

    @property
    def is_anonymous(self):
        return False

    def get_id(self):
        return self.id

    # Required for administrative interface
    def __unicode__(self):
        return self.username


class Task(ModelBase):
    task = CharField(unique=True)
    base_points = SmallIntegerField()
    time_factor = FloatField(default=0.0)
    state = SmallIntegerField(index=True, default=0)
    BACKLOG = 0
    TODO = 1
    DONE = 2
    todo_time = DateTimeField(null=True)
    last_done = DateTimeField(null=True)
    schedule_days = SmallIntegerField(null=True)

    @property
    def points(self):
        """
        Calculate real point value based on time_factor.
        :return:
        """
        now = datetime.utcnow()
        # set last_time to now if todo_time is not set.
        last_time = now if not self.todo_time else self.todo_time
        real_points = self.base_points + (self.time_factor * (now - last_time).days)
        return int(real_points)

    @classmethod
    def get_all(cls):
        """
        Overwrite get_all() method because we want to have active tasks only and
        need the property but dicts() doesn't work with it
        :return:
        """
        return list(cls.select().order_by(cls.base_points.desc()))

    @classmethod
    def set_state(cls, task_id, state, user_id):
        now = datetime.utcnow()
        points_obtained = 0
        with db_wrapper.database.atomic():
            # update task state and time
            task = cls.get(cls.id == task_id)
            task.state = state
            if state == cls.DONE:
                points_obtained = task.points
                task.todo_time = None
                task.last_done = now
            elif state == cls.TODO:
                task.todo_time = now
            elif state == cls.BACKLOG:
                task.todo_time = None
            task.save()

            # update user points if new state is DONE (user got points)
            if points_obtained > 0:
                User.update(points=User.points + points_obtained, last_update=now).where(
                    User.id == user_id).execute()

                # add to history
                History.create(task=task.task, user=user_id, points=points_obtained, time=now)

    @classmethod
    def set_todo(cls, task_id):
        now = datetime.utcnow()
        with db_wrapper.database.atomic():
            task = cls.get(cls.id == task_id)
            task.state = cls.TODO
            task.todo_time = now
            task.save()

    @classmethod
    def get_schedule_tasks(cls):
        return list(
            cls.select(cls.id, cls.last_done, cls.schedule_days)
               .where((cls.schedule_days.is_null(False)) & (cls.state != cls.TODO)).dicts())


    @staticmethod
    def do_custom_task(task, points, user_id):
        now = datetime.utcnow()
        with db_wrapper.database.atomic():
            # update user points
            User.update(points=User.points + points, last_update=now).where(User.id == user_id).execute()
            # add to history
            History.create(task=task, user=user_id, points=points, time=now)


class History(ModelBase):
    task = CharField()
    user = ForeignKeyField(User)
    points = SmallIntegerField()
    time = DateTimeField(default=datetime.utcnow())

    @classmethod
    def get_user_history(cls, user):
        return list(
            cls.select(cls.time, cls.task, cls.points)
               .join(User, on=(cls.user == User.id))

               .where(User.username == user)

               .order_by(cls.time.desc()).dicts())


    @classmethod
    def get_full_history(cls):
        return list(
            cls.select(cls.time, cls.points, User.username)
               .join(User, on=(cls.user == User.id))

               .order_by(cls.time.asc()).dicts())



1			from datetime import datetime
			0 ignored issues – show Bug introduced 2018-02-08 11:46 UTC by Report Bug Copy Issue Report There seems to be a cyclic import (chaoswg -> chaoswg.scheduler). Cyclic imports may cause partly loaded modules to be returned. This might lead to unexpected runtime behavior which is hard to debug. Loading history...
2
3			from peewee import (CharField, IntegerField, DateTimeField, SmallIntegerField, ForeignKeyField,
4			FloatField, DoesNotExist, SqliteDatabase)
5			from playhouse.flask_utils import FlaskDB
6			from werkzeug.security import generate_password_hash, check_password_hash
7
8			db_wrapper = FlaskDB()
			0 ignored issues – show Coding Style Naming introduced 2018-02-01 18:15 UTC by Report Bug Copy Issue Report The name `db_wrapper` does not conform to the constant naming conventions (`(([A-Z_][A-Z0-9_])\|(__.__))$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
9
10
11			def init_database(app):
12			db_wrapper._db = SqliteDatabase(app.config['DATABASE'], pragmas=(('journal_mode', 'wal'),))
			0 ignored issues – show Coding Style Best Practice introduced 2018-02-05 18:44 UTC by Report Bug Copy Issue Report It seems like `_db` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
13			db_wrapper.init_app(app)
14			return db_wrapper.database
15
16
17			def create_tables():
18			db_wrapper.database.connect()
19			db_wrapper.database.create_tables([User, Task, History], safe=True)
20			db_wrapper.database.close()
21
22
23			class ModelBase(db_wrapper.Model):
24			@classmethod
25			def get_all(cls):
26			return list(cls.select().dicts())
27
28
29			class User(ModelBase):
30			username = CharField(unique=True)
31			password = CharField()
32			points = IntegerField(default=0)
33			last_update = DateTimeField(default=datetime.utcnow)
34
35			@classmethod
36			def get_all(cls):
37			"""
38			without password
39			:return:
40			"""
41			return list(cls.select(cls.username, cls.points, cls.last_update).dicts())
42
43			@classmethod
44			def get_by_name(cls, username):
45			try:
46			return cls.get(cls.username == username)
47			except DoesNotExist:
48			return False
49
50			@classmethod
51			def get_usernames(cls):
52			query = cls.select(cls.username)
53			result = set()
54			for user in query:
55			result.add(user.username)
56			return result
57
58			def set_password(self, password):
59			self.password = generate_password_hash(password)
60
61			def check_password(self, password):
62			return check_password_hash(self.password, password)
63
64			# Flask-Login required functions
65			@property
66			def is_active(self):
67			return True
68
69			@property
70			def is_authenticated(self):
71			return True
72
73			@property
74			def is_anonymous(self):
75			return False
76
77			def get_id(self):
78			return self.id
79
80			# Required for administrative interface
81			def __unicode__(self):
82			return self.username
83
84
85			class Task(ModelBase):
86			task = CharField(unique=True)
87			base_points = SmallIntegerField()
88			time_factor = FloatField(default=0.0)
89			state = SmallIntegerField(index=True, default=0)
90			BACKLOG = 0
91			TODO = 1
92			DONE = 2
93			todo_time = DateTimeField(null=True)
94			last_done = DateTimeField(null=True)
95			schedule_days = SmallIntegerField(null=True)
96
97			@property
98			def points(self):
99			"""
100			Calculate real point value based on time_factor.
101			:return:
102			"""
103			now = datetime.utcnow()
104			# set last_time to now if todo_time is not set.
105			last_time = now if not self.todo_time else self.todo_time
106			real_points = self.base_points + (self.time_factor * (now - last_time).days)
107			return int(real_points)
108
109			@classmethod
110			def get_all(cls):
111			"""
112			Overwrite get_all() method because we want to have active tasks only and
113			need the property but dicts() doesn't work with it
114			:return:
115			"""
116			return list(cls.select().order_by(cls.base_points.desc()))
117
118			@classmethod
119			def set_state(cls, task_id, state, user_id):
120			now = datetime.utcnow()
121			points_obtained = 0
122			with db_wrapper.database.atomic():
123			# update task state and time
124			task = cls.get(cls.id == task_id)
125			task.state = state
126			if state == cls.DONE:
127			points_obtained = task.points
128			task.todo_time = None
129			task.last_done = now
130			elif state == cls.TODO:
131			task.todo_time = now
132			elif state == cls.BACKLOG:
133			task.todo_time = None
134			task.save()
135
136			# update user points if new state is DONE (user got points)
137			if points_obtained > 0:
138			User.update(points=User.points + points_obtained, last_update=now).where(
139			User.id == user_id).execute()
140
141			# add to history
142			History.create(task=task.task, user=user_id, points=points_obtained, time=now)
143
144			@classmethod
145			def set_todo(cls, task_id):
146			now = datetime.utcnow()
147			with db_wrapper.database.atomic():
148			task = cls.get(cls.id == task_id)
149			task.state = cls.TODO
150			task.todo_time = now
151			task.save()
152
153			@classmethod
154			def get_schedule_tasks(cls):
155			return list(
156			cls.select(cls.id, cls.last_done, cls.schedule_days)
157			.where((cls.schedule_days.is_null(False)) & (cls.state != cls.TODO)).dicts())
			0 ignored issues – show Coding Style introduced 2018-02-07 23:13 UTC by Report Bug Copy Issue Report Wrong hanging indentation (remove 3 spaces). Loading history...
158
159			@staticmethod
160			def do_custom_task(task, points, user_id):
161			now = datetime.utcnow()
162			with db_wrapper.database.atomic():
163			# update user points
164			User.update(points=User.points + points, last_update=now).where(User.id == user_id).execute()
165			# add to history
166			History.create(task=task, user=user_id, points=points, time=now)
167
168
169			class History(ModelBase):
170			task = CharField()
171			user = ForeignKeyField(User)
172			points = SmallIntegerField()
173			time = DateTimeField(default=datetime.utcnow())
174
175			@classmethod
176			def get_user_history(cls, user):
177			return list(
178			cls.select(cls.time, cls.task, cls.points)
179			.join(User, on=(cls.user == User.id))
			0 ignored issues – show Coding Style introduced 2018-02-07 23:13 UTC by Report Bug Copy Issue Report Wrong hanging indentation (remove 3 spaces). Loading history...
180			.where(User.username == user)
			0 ignored issues – show Coding Style introduced 2018-02-07 23:13 UTC by Report Bug Copy Issue Report Wrong hanging indentation (remove 3 spaces). Loading history...
181			.order_by(cls.time.desc()).dicts())
			0 ignored issues – show Coding Style introduced 2018-02-07 23:13 UTC by Report Bug Copy Issue Report Wrong hanging indentation (remove 3 spaces). Loading history...
182
183			@classmethod
184			def get_full_history(cls):
185			return list(
186			cls.select(cls.time, cls.points, User.username)
187			.join(User, on=(cls.user == User.id))
			0 ignored issues – show Coding Style introduced 2018-02-07 23:13 UTC by Report Bug Copy Issue Report Wrong hanging indentation (remove 3 spaces). Loading history...
188			.order_by(cls.time.asc()).dicts())
			0 ignored issues – show Coding Style introduced 2018-02-07 23:13 UTC by Report Bug Copy Issue Report Wrong hanging indentation (remove 3 spaces). Loading history...
189

Obihoernchen / ChaosWG-Manager

Pull Request — master (#6)

chaoswg.models.Task.set_todo() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like