src.things3.Things3.get_rows() - Code Metrics - Inspection of "final fixes for release 2.0" - AlexanderWillner/KanbanView - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 3e6a93...b6cc82 )

by Alexander

created 2020-04-06 11:15 UTC

src.things3.Things3.get_rows() A

↳ Parent: src.things3

Complexity

Conditions

Size

Total Lines	41
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	5
nop	2
dl	0
loc	41
rs	10
c	0
b	0
f	0

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""Simple read-only API for Things 3."""

from __future__ import print_function

__author__ = "Alexander Willner"
__copyright__ = "2020 Alexander Willner"
__credits__ = ["Alexander Willner"]
__license__ = "Apache"
__version__ = "2.0.0"
__maintainer__ = "Alexander Willner"
__email__ = "[email protected]"
__status__ = "Development"

import sqlite3
from os.path import expanduser
from os import environ


class Things3():
    """Simple read-only API for Things 3."""
    # Variables
    database = None
    cursor = None
    json = False
    tag_waiting = "Waiting" if not environ.get('TAG_WAITING') \
        else environ.get('TAG_WAITING')
    tag_mit = "MIT" if not environ.get('TAG_MIT') \
        else environ.get('TAG_MIT')

    # Database info
    FILE_SQLITE = '~/Library/Containers/'\
                  'com.culturedcode.ThingsMac.beta/Data/Library/'\
                  'Application Support/Cultured Code/Things/Things.sqlite3'\
        if not environ.get('THINGSDB') else environ.get('THINGSDB')
    TASKTABLE = "TMTask"
    AREATABLE = "TMArea"
    TAGTABLE = "TMTag"
    TASKTAGTABLE = "TMTaskTag"
    ISNOTTRASHED = "TASK.trashed = 0"
    ISTRASHED = "TASK.trashed = 1"
    ISOPEN = "TASK.status = 0"
    ISNOTSTARTED = "TASK.start = 0"
    ISCANCELLED = "TASK.status = 2"
    ISCOMPLETED = "TASK.status = 3"
    ISSTARTED = "TASK.start = 1"
    ISPOSTPONED = "TASK.start = 2"
    ISTASK = "TASK.type = 0"
    ISPROJECT = "TASK.type = 1"
    ISHEADING = "TASK.type = 2"
    ISOPENTASK = ISTASK + " AND " + ISNOTTRASHED + " AND " + ISOPEN
    DATECREATE = "creationDate"
    DATEMOD = "userModificationDate"
    DATEDUE = "dueDate"
    DATESTART = "unixepoch"
    DATESTOP = "stopDate"

    # Query Index
    I_UUID = 0
    I_TITLE = 1
    I_CONTEXT = 2
    I_CONTEXT_UUID = 3
    I_DUE = 4
    I_CREATE = 5
    I_MOD = 6
    I_START = 7
    I_STOP = 8

    def __init__(self,
                 database=FILE_SQLITE,
                 tag_waiting='Waiting',
                 tag_mit='MIT',
                 json=False):
        self.database = expanduser(database)
        self.tag_mit = tag_mit
        self.tag_waiting = tag_waiting
        self.cursor = sqlite3.connect(self.database).cursor()
        self.json = json

    def get_inbox(self):
        """Get all tasks from the inbox."""
        query = self.ISOPENTASK + " AND " + self.ISNOTSTARTED + \
            " ORDER BY TASK.duedate DESC , TASK.todayIndex"
        return self.get_rows(query)

    def get_today(self):
        """Get all tasks from the today list."""
        query = self.ISOPENTASK + " AND " + self.ISSTARTED + \
            " AND TASK.startdate is NOT NULL" + \
            " ORDER BY TASK.duedate DESC , TASK.todayIndex"
        return self.get_rows(query)

    def get_someday(self):
        """Get someday tasks."""
        query = self.ISOPENTASK + " AND " + self.ISPOSTPONED + \
            " AND TASK.startdate IS NULL AND TASK.recurrenceRule IS NULL" + \
            " ORDER BY TASK.duedate DESC, TASK.creationdate DESC"
        return self.get_rows(query)

    def get_upcoming(self):
        """Get upcoming tasks."""
        query = self.ISOPENTASK + " AND " + self.ISPOSTPONED + \
            " AND (TASK.startDate NOT NULL " + \
            "      OR TASK.recurrenceRule NOT NULL)" + \
            " ORDER BY TASK.startdate, TASK.todayIndex"
        return self.get_rows(query)

    def get_waiting(self):
        """Get waiting tasks."""
        query = self.ISOPENTASK + \
            " AND TAGS.tags=(SELECT uuid FROM " + self.TAGTABLE + \
            " WHERE title='" + self.tag_waiting + "')" + \
            " ORDER BY TASK.duedate DESC , TASK.todayIndex"
        return self.get_rows(query)

    def get_mit(self):
        """Get most important tasks."""
        query = self.ISOPENTASK + " AND " + self.ISSTARTED + \
            " AND PROJECT.status = 0 " \
            " AND TAGS.tags=(SELECT uuid FROM " + self.TAGTABLE + \
            " WHERE title='" + self.tag_mit + "')" + \
            " ORDER BY TASK.duedate DESC , TASK.todayIndex"
        return self.get_rows(query)

    def get_anytime(self):
        """Get anytime tasks."""
        query = self.ISOPENTASK + " AND " + self.ISSTARTED + \
            " AND TASK.startdate is NULL" + \
            " AND (TASK.area NOT NULL OR TASK.project in " + \
            "(SELECT uuid FROM " + self.TASKTABLE + \
            " WHERE uuid=TASK.project AND start=1" + \
            " AND trashed=0))" + \
            " ORDER BY TASK.duedate DESC , TASK.todayIndex"
        return self.get_rows(query)

    def get_completed(self):
        """Get completed tasks."""
        query = self.ISNOTTRASHED + " AND " + self.ISTASK + \
            " AND " + self.ISCOMPLETED + \
            " ORDER BY TASK." + self.DATESTOP
        return self.get_rows(query)

    def get_cancelled(self):
        """Get cancelled tasks."""
        query = self.ISNOTTRASHED + " AND " + self.ISTASK + \
            " AND " + self.ISCANCELLED + \
            " ORDER BY TASK." + self.DATESTOP
        return self.get_rows(query)

    def get_trashed(self):
        """Get trashed tasks."""
        query = self.ISTRASHED + " AND " + self.ISTASK + \
            " ORDER BY TASK." + self.DATESTOP
        return self.get_rows(query)

    def get_all(self):
        """Get all tasks."""
        query = self.ISNOTTRASHED + " AND " + self.ISTASK
        return self.get_rows(query)

    def get_due(self):
        """Get due tasks."""
        query = self.ISOPENTASK + " AND TASK.dueDate NOT NULL" + \
            " ORDER BY TASK.dueDate"
        return self.get_rows(query)

    @staticmethod
    def get_not_implemented():
        """Not implemented warning."""
        return [["0", "not implemented", "no context", "0", "0", "0", "0",
                 "0", "0", "0", "0", "0", "0", "0", "0", "0", "0", "0", "0"]]

    def get_rows(self, sql):
        """Query Things database."""

        sql = """
            SELECT DISTINCT
                TASK.uuid,
                TASK.title,
                CASE
                    WHEN AREA.title IS NOT NULL THEN AREA.title
                    WHEN PROJECT.title IS NOT NULL THEN PROJECT.title
                    WHEN HEADING.title IS NOT NULL THEN HEADING.title
                END,
                CASE
                    WHEN AREA.uuid IS NOT NULL THEN AREA.uuid
                    WHEN PROJECT.uuid IS NOT NULL THEN PROJECT.uuid
                END,
                CASE
                    WHEN TASK.recurrenceRule IS NULL
                    THEN date(TASK.dueDate,"unixepoch")
                ELSE NULL
                END,
                date(TASK.creationDate,"unixepoch"),
                date(TASK.userModificationDate,"unixepoch"),
                date(TASK.startDate,"unixepoch"),
                date(TASK.stopDate,"unixepoch")
            FROM
                TMTask AS TASK
            LEFT OUTER JOIN
                TMTask PROJECT ON TASK.project = PROJECT.uuid
            LEFT OUTER JOIN
                TMArea AREA ON TASK.area = AREA.uuid
            LEFT OUTER JOIN
                TMTask HEADING ON TASK.actionGroup = HEADING.uuid
            LEFT OUTER JOIN
                TMTaskTag TAGS ON TASK.uuid = TAGS.tasks
            LEFT OUTER JOIN
                TMTag TAG ON TAGS.tags = TAG.uuid
            WHERE """ + sql

        self.cursor.execute(sql)
        return self.cursor.fetchall()

    def convert_task_to_model(self, task):
        """Convert task to model."""
        model = {'uuid': task[self.I_UUID],
                 'title': task[self.I_TITLE],
                 'context': task[self.I_CONTEXT],
                 'context_uuid': task[self.I_CONTEXT_UUID],
                 'due': task[self.I_DUE],
                 'created': task[self.I_CREATE],
                 'modified': task[self.I_MOD],
                 'started': task[self.I_START],
                 'stopped': task[self.I_STOP]
                 }
        return model

    def convert_tasks_to_model(self, tasks):
        """Convert tasks to model."""
        model = []
        for task in tasks:
            model.append(self.convert_task_to_model(task))
        return model

    functions = {
        "inbox": get_inbox,
        "today": get_today,
        "next": get_anytime,
        "backlog": get_someday,
        "upcoming": get_upcoming,
        "waiting": get_waiting,
        "mit": get_mit,
        "completed": get_completed,
        "cancelled": get_cancelled,
        "trashed": get_trashed,
        "all": get_all,
        "due": get_due,
    }


1			#!/usr/bin/env python3
2			# -- coding: utf-8 --
3
4			"""Simple read-only API for Things 3."""
5
6			from __future__ import print_function
7
8			__author__ = "Alexander Willner"
9			__copyright__ = "2020 Alexander Willner"
10			__credits__ = ["Alexander Willner"]
11			__license__ = "Apache"
12			__version__ = "2.0.0"
13			__maintainer__ = "Alexander Willner"
14			__email__ = "[email protected]"
15			__status__ = "Development"
16
17			import sqlite3
18			from os.path import expanduser
19			from os import environ
20
21
22			class Things3():
23			"""Simple read-only API for Things 3."""
24			# Variables
25			database = None
26			cursor = None
27			json = False
28			tag_waiting = "Waiting" if not environ.get('TAG_WAITING') \
29			else environ.get('TAG_WAITING')
30			tag_mit = "MIT" if not environ.get('TAG_MIT') \
31			else environ.get('TAG_MIT')
32
33			# Database info
34			FILE_SQLITE = '~/Library/Containers/'\
35			'com.culturedcode.ThingsMac.beta/Data/Library/'\
36			'Application Support/Cultured Code/Things/Things.sqlite3'\
37			if not environ.get('THINGSDB') else environ.get('THINGSDB')
38			TASKTABLE = "TMTask"
39			AREATABLE = "TMArea"
40			TAGTABLE = "TMTag"
41			TASKTAGTABLE = "TMTaskTag"
42			ISNOTTRASHED = "TASK.trashed = 0"
43			ISTRASHED = "TASK.trashed = 1"
44			ISOPEN = "TASK.status = 0"
45			ISNOTSTARTED = "TASK.start = 0"
46			ISCANCELLED = "TASK.status = 2"
47			ISCOMPLETED = "TASK.status = 3"
48			ISSTARTED = "TASK.start = 1"
49			ISPOSTPONED = "TASK.start = 2"
50			ISTASK = "TASK.type = 0"
51			ISPROJECT = "TASK.type = 1"
52			ISHEADING = "TASK.type = 2"
53			ISOPENTASK = ISTASK + " AND " + ISNOTTRASHED + " AND " + ISOPEN
54			DATECREATE = "creationDate"
55			DATEMOD = "userModificationDate"
56			DATEDUE = "dueDate"
57			DATESTART = "unixepoch"
58			DATESTOP = "stopDate"
59
60			# Query Index
61			I_UUID = 0
62			I_TITLE = 1
63			I_CONTEXT = 2
64			I_CONTEXT_UUID = 3
65			I_DUE = 4
66			I_CREATE = 5
67			I_MOD = 6
68			I_START = 7
69			I_STOP = 8
70
71			def __init__(self,
72			database=FILE_SQLITE,
73			tag_waiting='Waiting',
74			tag_mit='MIT',
75			json=False):
76			self.database = expanduser(database)
77			self.tag_mit = tag_mit
78			self.tag_waiting = tag_waiting
79			self.cursor = sqlite3.connect(self.database).cursor()
80			self.json = json
81
82			def get_inbox(self):
83			"""Get all tasks from the inbox."""
84			query = self.ISOPENTASK + " AND " + self.ISNOTSTARTED + \
85			" ORDER BY TASK.duedate DESC , TASK.todayIndex"
86			return self.get_rows(query)
87
88			def get_today(self):
89			"""Get all tasks from the today list."""
90			query = self.ISOPENTASK + " AND " + self.ISSTARTED + \
91			" AND TASK.startdate is NOT NULL" + \
92			" ORDER BY TASK.duedate DESC , TASK.todayIndex"
93			return self.get_rows(query)
94
95			def get_someday(self):
96			"""Get someday tasks."""
97			query = self.ISOPENTASK + " AND " + self.ISPOSTPONED + \
98			" AND TASK.startdate IS NULL AND TASK.recurrenceRule IS NULL" + \
99			" ORDER BY TASK.duedate DESC, TASK.creationdate DESC"
100			return self.get_rows(query)
101
102			def get_upcoming(self):
103			"""Get upcoming tasks."""
104			query = self.ISOPENTASK + " AND " + self.ISPOSTPONED + \
105			" AND (TASK.startDate NOT NULL " + \
106			" OR TASK.recurrenceRule NOT NULL)" + \
107			" ORDER BY TASK.startdate, TASK.todayIndex"
108			return self.get_rows(query)
109
110			def get_waiting(self):
111			"""Get waiting tasks."""
112			query = self.ISOPENTASK + \
113			" AND TAGS.tags=(SELECT uuid FROM " + self.TAGTABLE + \
114			" WHERE title='" + self.tag_waiting + "')" + \
115			" ORDER BY TASK.duedate DESC , TASK.todayIndex"
116			return self.get_rows(query)
117
118			def get_mit(self):
119			"""Get most important tasks."""
120			query = self.ISOPENTASK + " AND " + self.ISSTARTED + \
121			" AND PROJECT.status = 0 " \
122			" AND TAGS.tags=(SELECT uuid FROM " + self.TAGTABLE + \
123			" WHERE title='" + self.tag_mit + "')" + \
124			" ORDER BY TASK.duedate DESC , TASK.todayIndex"
125			return self.get_rows(query)
126
127			def get_anytime(self):
128			"""Get anytime tasks."""
129			query = self.ISOPENTASK + " AND " + self.ISSTARTED + \
130			" AND TASK.startdate is NULL" + \
131			" AND (TASK.area NOT NULL OR TASK.project in " + \
132			"(SELECT uuid FROM " + self.TASKTABLE + \
133			" WHERE uuid=TASK.project AND start=1" + \
134			" AND trashed=0))" + \
135			" ORDER BY TASK.duedate DESC , TASK.todayIndex"
136			return self.get_rows(query)
137
138			def get_completed(self):
139			"""Get completed tasks."""
140			query = self.ISNOTTRASHED + " AND " + self.ISTASK + \
141			" AND " + self.ISCOMPLETED + \
142			" ORDER BY TASK." + self.DATESTOP
143			return self.get_rows(query)
144
145			def get_cancelled(self):
146			"""Get cancelled tasks."""
147			query = self.ISNOTTRASHED + " AND " + self.ISTASK + \
148			" AND " + self.ISCANCELLED + \
149			" ORDER BY TASK." + self.DATESTOP
150			return self.get_rows(query)
151
152			def get_trashed(self):
153			"""Get trashed tasks."""
154			query = self.ISTRASHED + " AND " + self.ISTASK + \
155			" ORDER BY TASK." + self.DATESTOP
156			return self.get_rows(query)
157
158			def get_all(self):
159			"""Get all tasks."""
160			query = self.ISNOTTRASHED + " AND " + self.ISTASK
161			return self.get_rows(query)
162
163			def get_due(self):
164			"""Get due tasks."""
165			query = self.ISOPENTASK + " AND TASK.dueDate NOT NULL" + \
166			" ORDER BY TASK.dueDate"
167			return self.get_rows(query)
168
169			@staticmethod
170			def get_not_implemented():
171			"""Not implemented warning."""
172			return [["0", "not implemented", "no context", "0", "0", "0", "0",
173			"0", "0", "0", "0", "0", "0", "0", "0", "0", "0", "0", "0"]]
174
175			def get_rows(self, sql):
176			"""Query Things database."""
177
178			sql = """
179			SELECT DISTINCT
180			TASK.uuid,
181			TASK.title,
182			CASE
183			WHEN AREA.title IS NOT NULL THEN AREA.title
184			WHEN PROJECT.title IS NOT NULL THEN PROJECT.title
185			WHEN HEADING.title IS NOT NULL THEN HEADING.title
186			END,
187			CASE
188			WHEN AREA.uuid IS NOT NULL THEN AREA.uuid
189			WHEN PROJECT.uuid IS NOT NULL THEN PROJECT.uuid
190			END,
191			CASE
192			WHEN TASK.recurrenceRule IS NULL
193			THEN date(TASK.dueDate,"unixepoch")
194			ELSE NULL
195			END,
196			date(TASK.creationDate,"unixepoch"),
197			date(TASK.userModificationDate,"unixepoch"),
198			date(TASK.startDate,"unixepoch"),
199			date(TASK.stopDate,"unixepoch")
200			FROM
201			TMTask AS TASK
202			LEFT OUTER JOIN
203			TMTask PROJECT ON TASK.project = PROJECT.uuid
204			LEFT OUTER JOIN
205			TMArea AREA ON TASK.area = AREA.uuid
206			LEFT OUTER JOIN
207			TMTask HEADING ON TASK.actionGroup = HEADING.uuid
208			LEFT OUTER JOIN
209			TMTaskTag TAGS ON TASK.uuid = TAGS.tasks
210			LEFT OUTER JOIN
211			TMTag TAG ON TAGS.tags = TAG.uuid
212			WHERE """ + sql
213
214			self.cursor.execute(sql)
215			return self.cursor.fetchall()
216
217			def convert_task_to_model(self, task):
218			"""Convert task to model."""
219			model = {'uuid': task[self.I_UUID],
220			'title': task[self.I_TITLE],
221			'context': task[self.I_CONTEXT],
222			'context_uuid': task[self.I_CONTEXT_UUID],
223			'due': task[self.I_DUE],
224			'created': task[self.I_CREATE],
225			'modified': task[self.I_MOD],
226			'started': task[self.I_START],
227			'stopped': task[self.I_STOP]
228			}
229			return model
230
231			def convert_tasks_to_model(self, tasks):
232			"""Convert tasks to model."""
233			model = []
234			for task in tasks:
235			model.append(self.convert_task_to_model(task))
236			return model
237
238			functions = {
239			"inbox": get_inbox,
240			"today": get_today,
241			"next": get_anytime,
242			"backlog": get_someday,
243			"upcoming": get_upcoming,
244			"waiting": get_waiting,
245			"mit": get_mit,
246			"completed": get_completed,
247			"cancelled": get_cancelled,
248			"trashed": get_trashed,
249			"all": get_all,
250			"due": get_due,
251			}
252

AlexanderWillner / KanbanView

Push — master ( 3e6a93...b6cc82 )

src.things3.Things3.get_rows() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like