chaoswg.models.Task.get_schedule_tasks() - Code Metrics - Obihoernchen/ChaosWG-Manager - Measure and Improve Code Quality continuously with Scrutinizer

chaoswg.models.Task.get_schedule_tasks() A
last analyzed 2018-10-11 18:46 UTC

↳ Parent: chaoswg.models

Complexity

Conditions

Size

Total Lines	5
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	5
nop	1
dl	0
loc	5
rs	10
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

    @classmethod
    def register(cls, username, password):
        if cls.get_by_name(username):
            return False
        if User.create(username=username, password=generate_password_hash(password)):
            return True

    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(index=True)
    time_factor = FloatField(default=0.0)
    state = SmallIntegerField(default=0)
    BACKLOG = 0
    TODO = 1
    DONE = 2
    todo_time = DateTimeField(null=True)
    last_done = DateTimeField(null=True)
    schedule_days = SmallIntegerField(null=True)

    class Meta:
        indexes = ((('schedule_days', 'state'), False),)

    @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(index=True, 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
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			@classmethod
59			def register(cls, username, password):
60			if cls.get_by_name(username):
61			return False
62			if User.create(username=username, password=generate_password_hash(password)):
63			return True
64
65			def set_password(self, password):
66			self.password = generate_password_hash(password)
67
68			def check_password(self, password):
69			return check_password_hash(self.password, password)
70
71			# Flask-Login required functions
72			@property
73			def is_active(self):
74			return True
75
76			@property
77			def is_authenticated(self):
78			return True
79
80			@property
81			def is_anonymous(self):
82			return False
83
84			def get_id(self):
85			return self.id
86
87			# Required for administrative interface
88			def __unicode__(self):
89			return self.username
90
91
92			class Task(ModelBase):
93			task = CharField(unique=True)
94			base_points = SmallIntegerField(index=True)
95			time_factor = FloatField(default=0.0)
96			state = SmallIntegerField(default=0)
97			BACKLOG = 0
98			TODO = 1
99			DONE = 2
100			todo_time = DateTimeField(null=True)
101			last_done = DateTimeField(null=True)
102			schedule_days = SmallIntegerField(null=True)
103
104			class Meta:
105			indexes = ((('schedule_days', 'state'), False),)
106
107			@property
108			def points(self):
109			"""
110			Calculate real point value based on time_factor.
111			:return:
112			"""
113			now = datetime.utcnow()
114			# set last_time to now if todo_time is not set.
115			last_time = now if not self.todo_time else self.todo_time
116			real_points = self.base_points + (self.time_factor * (now - last_time).days)
117			return int(real_points)
118
119			@classmethod
120			def get_all(cls):
121			"""
122			Overwrite get_all() method because we want to have active tasks only and
123			need the property but dicts() doesn't work with it
124			:return:
125			"""
126			return list(cls.select().order_by(cls.base_points.desc()))
127
128			@classmethod
129			def set_state(cls, task_id, state, user_id):
130			now = datetime.utcnow()
131			points_obtained = 0
132			with db_wrapper.database.atomic():
133			# update task state and time
134			task = cls.get(cls.id == task_id)
135			task.state = state
136			if state == cls.DONE:
137			points_obtained = task.points
138			task.todo_time = None
139			task.last_done = now
140			elif state == cls.TODO:
141			task.todo_time = now
142			elif state == cls.BACKLOG:
143			task.todo_time = None
144			task.save()
145
146			# update user points if new state is DONE (user got points)
147			if points_obtained > 0:
148			User.update(points=User.points + points_obtained, last_update=now).where(
149			User.id == user_id).execute()
150
151			# add to history
152			History.create(task=task.task, user=user_id, points=points_obtained, time=now)
153
154			@classmethod
155			def set_todo(cls, task_id):
156			now = datetime.utcnow()
157			with db_wrapper.database.atomic():
158			task = cls.get(cls.id == task_id)
159			task.state = cls.TODO
160			task.todo_time = now
161			task.save()
162
163			@classmethod
164			def get_schedule_tasks(cls):
165			return list(
166			cls.select(cls.id, cls.last_done, cls.schedule_days)
167			.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...
168
169			@staticmethod
170			def do_custom_task(task, points, user_id):
171			now = datetime.utcnow()
172			with db_wrapper.database.atomic():
173			# update user points
174			User.update(points=User.points + points, last_update=now).where(User.id == user_id).execute()
175			# add to history
176			History.create(task=task, user=user_id, points=points, time=now)
177
178
179			class History(ModelBase):
180			task = CharField()
181			user = ForeignKeyField(User)
182			points = SmallIntegerField()
183			time = DateTimeField(index=True, default=datetime.utcnow())
184
185			@classmethod
186			def get_user_history(cls, user):
187			return list(
188			cls.select(cls.time, cls.task, cls.points)
189			.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...
190			.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...
191			.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...
192
193			@classmethod
194			def get_full_history(cls):
195			return list(
196			cls.select(cls.time, cls.points, User.username)
197			.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...
198			.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...
199

Obihoernchen / ChaosWG-Manager

chaoswg.models.Task.get_schedule_tasks() A last analyzed 2018-10-11 18:46 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

chaoswg.models.Task.get_schedule_tasks() A
last analyzed 2018-10-11 18:46 UTC