things3.things3.Things3.get_completed() - Code Metrics - Inspection of "Started Eisenhower view" - AlexanderWillner/KanbanView - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 9d0b37...891a54 )

by Alexander

created 2020-04-27 19:57 UTC

things3.things3.Things3.get_completed() A

↳ Parent: things3.things3

Complexity

Conditions

Size

Total Lines	9
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	1
dl	0
loc	9
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 License 2.0"
__version__ = "2.5.3"
__maintainer__ = "Alexander Willner"
__email__ = "[email protected]"
__status__ = "Development"

import sqlite3
import sys
from random import shuffle
from os import environ, path
import getpass
import configparser
from pathlib import Path


# pylint: disable=R0904
class Things3():
    """Simple read-only API for Things 3."""

    # Database info
    FILE_DB = '/Library/Containers/'\
              'com.culturedcode.ThingsMac/Data/Library/'\
              'Application Support/Cultured Code/Things/Things.sqlite3'

    TABLE_TASK = "TMTask"
    TABLE_AREA = "TMArea"
    TABLE_TAG = "TMTag"
    TABLE_TASKTAG = "TMTaskTag"
    DATE_CREATE = "creationDate"
    DATE_MOD = "userModificationDate"
    DATE_DUE = "dueDate"
    DATE_START = "startDate"
    DATE_STOP = "stopDate"
    IS_INBOX = "start = 0"
    IS_ANYTIME = "start = 1"
    IS_SOMEDAY = "start = 2"
    IS_SCHEDULED = f"{DATE_START} IS NOT NULL"
    IS_NOT_SCHEDULED = f"{DATE_START} IS NULL"
    IS_DUE = f"{DATE_DUE} IS NOT NULL"
    IS_NOT_DUE = f"{DATE_DUE} IS NULL"
    IS_RECURRING = "recurrenceRule IS NOT NULL"
    IS_NOT_RECURRING = "recurrenceRule IS NULL"
    IS_TASK = "type = 0"
    IS_PROJECT = "type = 1"
    IS_HEADING = "type = 2"
    IS_TRASHED = "trashed = 1"
    IS_NOT_TRASHED = "trashed = 0"
    IS_OPEN = "status = 0"
    IS_CANCELLED = "status = 2"
    IS_DONE = "status = 3"
    RECURRING_IS_NOT_PAUSED = "instanceCreationPaused = 0"
    RECURRING_HAS_NEXT_STARTDATE = "nextInstanceStartDate IS NOT NULL"
    MODE_TASK = "type = 0"
    MODE_PROJECT = "type = 1"

    # Variables
    debug = False
    user = getpass.getuser()
    database = f"/Users/{user}/{FILE_DB}"
    filter = ""
    tag_waiting = "Waiting"
    tag_mit = "MIT"
    tag_cleanup = "Cleanup"
    anonymize = False
    config = configparser.ConfigParser()
    config.read(str(Path.home()) + '/.kanbanviewrc')

    # pylint: disable=R0913
    def __init__(self,
                 database=None,
                 tag_waiting=None,
                 tag_mit=None,
                 tag_cleanup=None,
                 anonymize=None):

        cfg = self.get_from_config(self.config, tag_waiting, 'TAG_WAITING')
        self.tag_waiting = cfg if cfg else self.tag_waiting

        cfg = self.get_from_config(self.config, anonymize, 'ANONYMIZE')
        self.anonymize = cfg if cfg else self.anonymize

        cfg = self.get_from_config(self.config, tag_mit, 'TAG_MIT')
        self.tag_mit = cfg if cfg else self.tag_mit

        cfg = self.get_from_config(self.config, tag_cleanup, 'TAG_CLEANUP')
        self.tag_cleanup = cfg if cfg else self.tag_cleanup

        cfg = self.get_from_config(self.config, database, 'THINGSDB')
        self.database = cfg if cfg else self.database

    @staticmethod
    def get_from_config(config, variable, preference):
        """Set variable. Priority: input, environment, config"""
        result = None
        if variable is not None:
            result = variable
        elif environ.get(preference):
            result = environ.get(preference)
        elif 'DATABASE' in config and preference in config['DATABASE']:
            result = path.expanduser(config['DATABASE'][preference])
        return result

    @staticmethod
    def anonymize_string(string):
        """Scramble text."""
        if string is None:
            return None
        string = list(string)
        shuffle(string)
        string = ''.join(string)
        return string

    @staticmethod
    def dict_factory(cursor, row):
        """Convert SQL result into a dictionary"""
        dictionary = {}
        for idx, col in enumerate(cursor.description):
            dictionary[col[0]] = row[idx]
        return dictionary

    def anonymize_tasks(self, tasks):
        """Scramble output for screenshots."""
        if self.anonymize:
            for task in tasks:
                task['title'] = self.anonymize_string(task['title'])
                task['context'] = self.anonymize_string(task['context'])
        return tasks

    def get_inbox(self):
        """Get all tasks from the inbox."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                TASK.{self.IS_INBOX}
                ORDER BY TASK.duedate DESC , TASK.todayIndex
                """
        return self.get_rows(query)

    def get_today(self):
        """Get all tasks from the today list."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                (TASK.{self.IS_ANYTIME} OR (
                     TASK.{self.IS_SOMEDAY} AND
                     TASK.{self.DATE_START} <= strftime('%s', 'now')
                     )
                ) AND
                TASK.{self.IS_SCHEDULED} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.duedate DESC , TASK.todayIndex
                """
        return self.get_rows(query)

    def get_someday(self):
        """Get someday tasks."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                TASK.{self.IS_SOMEDAY} AND
                TASK.{self.IS_NOT_SCHEDULED} AND
                TASK.{self.IS_NOT_RECURRING} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.duedate DESC, TASK.creationdate DESC
                """
        return self.get_rows(query)

    def get_upcoming(self):
        """Get upcoming tasks."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                TASK.{self.IS_SOMEDAY} AND
                TASK.{self.IS_SCHEDULED} AND
                TASK.{self.IS_NOT_RECURRING} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.startdate, TASK.todayIndex
                """
        return self.get_rows(query)

    def get_waiting(self):
        """Get waiting tasks."""
        return self.get_tag(self.tag_waiting)

    def get_mit(self):
        """Get most important tasks."""
        return self.get_tag(self.tag_mit)

    def get_tag(self, tag):
        """Get task with specific tag"""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                TASK.{self.IS_NOT_RECURRING} AND
                TAGS.tags=(SELECT uuid FROM {self.TABLE_TAG}
                             WHERE title='{tag}'
                          )
                AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.duedate DESC , TASK.todayIndex
                """
        return self.get_rows(query)

    def get_tag_today(self, tag):
        """Get today tasks with specific tag"""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                (TASK.{self.IS_ANYTIME} OR (
                     TASK.{self.IS_SOMEDAY} AND
                     TASK.{self.DATE_START} <= strftime('%s', 'now')
                     )
                ) AND
                TAGS.tags=(SELECT uuid FROM {self.TABLE_TAG}
                             WHERE title='{tag}') AND
                TASK.{self.IS_SCHEDULED} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.duedate DESC , TASK.todayIndex
            """
        return self.get_rows(query)

    def get_anytime(self):
        """Get anytime tasks."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                TASK.{self.IS_ANYTIME} AND
                TASK.{self.IS_NOT_SCHEDULED} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_ANYTIME} AND
                            PROJECT.{self.IS_NOT_SCHEDULED} AND
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_ANYTIME} AND
                            HEADPROJ.{self.IS_NOT_SCHEDULED} AND
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.duedate DESC , TASK.todayIndex
                """
        if self.filter:
            # ugly hack for Kanban task view on project
            query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                TASK.{self.IS_ANYTIME} AND
                TASK.{self.IS_NOT_SCHEDULED} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.duedate DESC , TASK.todayIndex
                """
        return self.get_rows(query)

    def get_completed(self):
        """Get completed tasks."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_DONE}
                ORDER BY TASK.{self.DATE_STOP}
                """
        return self.get_rows(query)

    def get_cancelled(self):
        """Get cancelled tasks."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_CANCELLED}
                ORDER BY TASK.{self.DATE_STOP}
                """
        return self.get_rows(query)

    def get_trashed(self):
        """Get trashed tasks."""
        query = f"""
                TASK.{self.IS_TRASHED} AND
                TASK.{self.IS_TASK}
                ORDER BY TASK.{self.DATE_STOP}
                """
        return self.get_rows(query)

    def get_projects(self):
        """Get projects."""
        query = f"""
                SELECT
                    TASK.uuid,
                    TASK.title,
                    NULL as context,
                    (SELECT COUNT(uuid)
                     FROM TMTask AS PROJECT_TASK
                     WHERE
                       PROJECT_TASK.project = TASK.uuid AND
                       PROJECT_TASK.{self.IS_NOT_TRASHED} AND
                       PROJECT_TASK.{self.IS_OPEN}
                    ) AS size
                FROM
                    {self.TABLE_TASK} AS TASK
                WHERE
                    TASK.{self.IS_NOT_TRASHED} AND
                    TASK.{self.IS_PROJECT} AND
                    TASK.{self.IS_OPEN}
                ORDER BY TASK.title COLLATE NOCASE
                """
        return self.execute_query(query)

    def get_areas(self):
        """Get areas."""
        query = f"""
                SELECT
                    AREA.uuid AS uuid,
                    AREA.title AS title,
                    (SELECT COUNT(uuid)
                        FROM TMTask AS PROJECT
                        WHERE
                        PROJECT.area = AREA.uuid AND
                        PROJECT.{self.IS_NOT_TRASHED} AND
                        PROJECT.{self.IS_OPEN}
                    ) AS size
                FROM
                    {self.TABLE_AREA} AS AREA
                ORDER BY AREA.title COLLATE NOCASE
                """
        return self.execute_query(query)

    def get_all(self):
        """Get all tasks."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                """
        return self.get_rows(query)

    def get_due(self):
        """Get due tasks."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                TASK.{self.IS_DUE} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.{self.DATE_DUE}
                """
        return self.get_rows(query)

    def get_lint(self):
        """Get tasks that float around"""
        query = f"""
            TASK.{self.IS_NOT_TRASHED} AND
            TASK.{self.IS_OPEN} AND
            TASK.{self.IS_TASK} AND
            (TASK.{self.IS_SOMEDAY} OR TASK.{self.IS_ANYTIME}) AND
            TASK.project IS NULL AND
            TASK.area IS NULL AND
            TASK.actionGroup IS NULL
            """
        return self.get_rows(query)

    def get_empty_projects(self):
        """Get projects that are empty"""
        query = f"""
            TASK.{self.IS_NOT_TRASHED} AND
            TASK.{self.IS_OPEN} AND
            TASK.{self.IS_PROJECT} AND
            TASK.{self.IS_ANYTIME}
            GROUP BY TASK.uuid
            HAVING
                (SELECT COUNT(uuid)
                 FROM TMTask AS PROJECT_TASK
                 WHERE
                   PROJECT_TASK.project = TASK.uuid AND
                   PROJECT_TASK.{self.IS_NOT_TRASHED} AND
                   PROJECT_TASK.{self.IS_OPEN} AND
                   (PROJECT_TASK.{self.IS_ANYTIME} OR
                    PROJECT_TASK.{self.IS_SCHEDULED} OR
                      (PROJECT_TASK.{self.IS_RECURRING} AND
                       PROJECT_TASK.{self.RECURRING_IS_NOT_PAUSED} AND
                       PROJECT_TASK.{self.RECURRING_HAS_NEXT_STARTDATE}
                      )
                   )
                ) = 0
            """
        return self.get_rows(query)

    def get_largest_projects(self):
        """Get projects that are empty"""
        query = f"""
            SELECT
                TASK.uuid,
                TASK.title AS title,
                creationDate AS created,
                userModificationDate AS modified,
                (SELECT COUNT(uuid)
                 FROM TMTask AS PROJECT_TASK
                 WHERE
                   PROJECT_TASK.project = TASK.uuid AND
                   PROJECT_TASK.{self.IS_NOT_TRASHED} AND
                   PROJECT_TASK.{self.IS_OPEN}
                ) AS tasks
            FROM
                {self.TABLE_TASK} AS TASK
            WHERE
               TASK.{self.IS_NOT_TRASHED} AND
               TASK.{self.IS_OPEN} AND
               TASK.{self.IS_PROJECT}
            GROUP BY TASK.uuid
            ORDER BY tasks COLLATE NOCASE DESC
            """
        return self.execute_query(query)

    def get_daystats(self):
        """Get a history of task activities"""
        days = 365
        query = f"""
                WITH RECURSIVE timeseries(x) AS (
                    SELECT 0
                    UNION ALL
                    SELECT x+1 FROM timeseries
                    LIMIT {days}
                )
                SELECT
                    date(julianday("now", "-{days} days"),
                         "+" || x || " days") as date,
                    CREATED.TasksCreated as created,
                    CLOSED.TasksClosed as completed,
                    CANCELLED.TasksCancelled as cancelled,
                    TRASHED.TasksTrashed as trashed
                FROM timeseries
                LEFT JOIN
                    (SELECT COUNT(uuid) AS TasksCreated,
                        date(creationDate,"unixepoch") AS DAY
                        FROM {self.TABLE_TASK} AS TASK
                        WHERE DAY NOT NULL
                          AND TASK.{self.IS_TASK}
                        GROUP BY DAY)
                    AS CREATED ON CREATED.DAY = date
                LEFT JOIN
                    (SELECT COUNT(uuid) AS TasksCancelled,
                        date(stopDate,"unixepoch") AS DAY
                        FROM {self.TABLE_TASK} AS TASK
                        WHERE DAY NOT NULL
                          AND TASK.{self.IS_CANCELLED} AND TASK.{self.IS_TASK}
                        GROUP BY DAY)
                        AS CANCELLED ON CANCELLED.DAY = date
                LEFT JOIN
                    (SELECT COUNT(uuid) AS TasksTrashed,
                        date(userModificationDate,"unixepoch") AS DAY
                        FROM {self.TABLE_TASK} AS TASK
                        WHERE DAY NOT NULL
                          AND TASK.{self.IS_TRASHED} AND TASK.{self.IS_TASK}
                        GROUP BY DAY)
                        AS TRASHED ON TRASHED.DAY = date
                LEFT JOIN
                    (SELECT COUNT(uuid) AS TasksClosed,
                        date(stopDate,"unixepoch") AS DAY
                        FROM {self.TABLE_TASK} AS TASK
                        WHERE DAY NOT NULL
                          AND TASK.{self.IS_DONE} AND TASK.{self.IS_TASK}
                        GROUP BY DAY)
                        AS CLOSED ON CLOSED.DAY = date
                """
        return self.execute_query(query)

    def get_minutes_today(self):
        """Count the planned minutes for today."""
        query = f"""
                SELECT
                    SUM(TAG.title) AS minutes
                FROM
                    {self.TABLE_TASK} 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
                    TMTask HEADPROJ ON HEADING.project = HEADPROJ.uuid
                LEFT OUTER JOIN
                    TMTaskTag TAGS ON TASK.uuid = TAGS.tasks
                LEFT OUTER JOIN
                    TMTag TAG ON TAGS.tags = TAG.uuid
                WHERE
                    printf("%d", TAG.title) = TAG.title AND
                    TASK.{self.IS_NOT_TRASHED} AND
                    TASK.{self.IS_TASK} AND
                    TASK.{self.IS_OPEN} AND
                    TASK.{self.IS_ANYTIME} AND
                    TASK.{self.IS_SCHEDULED} AND (
                        (
                            PROJECT.title IS NULL OR (
                                PROJECT.{self.IS_NOT_TRASHED}
                            )
                        ) AND (
                            HEADPROJ.title IS NULL OR (
                                HEADPROJ.{self.IS_NOT_TRASHED}
                            )
                        )
                    )
                """
        return self.execute_query(query)

    def get_cleanup(self):
        """Tasks and projects that need work."""
        result = []
        result.extend(self.get_lint())
        result.extend(self.get_empty_projects())
        result.extend(self.get_tag(self.tag_cleanup))
        result = [i for n, i in enumerate(result) if i not in result[n + 1:]]
        return result

    @staticmethod
    def get_not_implemented():
        """Not implemented warning."""
        return [{"title": "not implemented"}]

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

        sql = f"""
            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 AS context,
                CASE
                    WHEN AREA.uuid IS NOT NULL THEN AREA.uuid
                    WHEN PROJECT.uuid IS NOT NULL THEN PROJECT.uuid
                END AS context_uuid,
                CASE
                    WHEN TASK.recurrenceRule IS NULL
                    THEN date(TASK.dueDate,"unixepoch")
                ELSE NULL
                END AS due,
                date(TASK.creationDate,"unixepoch") as created,
                date(TASK.userModificationDate,"unixepoch") as modified,
                date(TASK.startDate,"unixepoch") as started,
                date(TASK.stopDate,"unixepoch") as stopped,
                (SELECT COUNT(uuid)
                 FROM TMTask AS PROJECT_TASK
                 WHERE
                   PROJECT_TASK.project = TASK.uuid AND
                   PROJECT_TASK.{self.IS_NOT_TRASHED} AND
                   PROJECT_TASK.{self.IS_OPEN}
                ) AS size,
                CASE
                    WHEN TASK.type = 0 THEN 'task'
                    WHEN TASK.type = 1 THEN 'project'
                    WHEN TASK.type = 2 THEN 'heading'
                END AS type
            FROM
                {self.TABLE_TASK} AS TASK
            LEFT OUTER JOIN
                {self.TABLE_TASK} PROJECT ON TASK.project = PROJECT.uuid
            LEFT OUTER JOIN
                {self.TABLE_AREA} AREA ON TASK.area = AREA.uuid
            LEFT OUTER JOIN
                {self.TABLE_TASK} HEADING ON TASK.actionGroup = HEADING.uuid
            LEFT OUTER JOIN
                {self.TABLE_TASK} HEADPROJ ON HEADING.project = HEADPROJ.uuid
            LEFT OUTER JOIN
                {self.TABLE_TASKTAG} TAGS ON TASK.uuid = TAGS.tasks
            LEFT OUTER JOIN
                {self.TABLE_TAG} TAG ON TAGS.tags = TAG.uuid
            WHERE
                {self.filter}
                {sql}
                """

        return self.execute_query(sql)

    def execute_query(self, sql):
        """Run the actual query"""
        if self.debug is True:
            print(sql)
        try:
            connection = sqlite3.connect(
                'file:' + self.database + '?mode=ro', uri=True)
            connection.row_factory = Things3.dict_factory
            cursor = connection.cursor()
            cursor.execute(sql)
            tasks = cursor.fetchall()
            tasks = self.anonymize_tasks(tasks)
            if self.debug:
                for task in tasks:
                    print(task)
            return tasks
        except sqlite3.OperationalError as error:
            print(f"Could not query the database at: {self.database}.")
            print(f"Details: {error}.")
            sys.exit(2)

    # pylint: disable=C0103
    def mode_project(self):
        """Hack to switch to project view"""
        self.IS_TASK = self.MODE_PROJECT

    # pylint: disable=C0103
    def mode_task(self):
        """Hack to switch to project view"""
        self.IS_TASK = self.MODE_TASK

    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,
        "projects": get_projects,
        "areas": get_areas,
        "all": get_all,
        "due": get_due,
        "lint": get_lint,
        "empty": get_empty_projects,
        "cleanup": get_cleanup,
        "top-proj": get_largest_projects,
        "stats-day": get_daystats,
        "stats-min-today": get_minutes_today
    }


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 License 2.0"
12			__version__ = "2.5.3"
13			__maintainer__ = "Alexander Willner"
14			__email__ = "[email protected]"
15			__status__ = "Development"
16
17			import sqlite3
18			import sys
19			from random import shuffle
20			from os import environ, path
21			import getpass
22			import configparser
23			from pathlib import Path
24
25
26			# pylint: disable=R0904
27			class Things3():
28			"""Simple read-only API for Things 3."""
29
30			# Database info
31			FILE_DB = '/Library/Containers/'\
32			'com.culturedcode.ThingsMac/Data/Library/'\
33			'Application Support/Cultured Code/Things/Things.sqlite3'
34
35			TABLE_TASK = "TMTask"
36			TABLE_AREA = "TMArea"
37			TABLE_TAG = "TMTag"
38			TABLE_TASKTAG = "TMTaskTag"
39			DATE_CREATE = "creationDate"
40			DATE_MOD = "userModificationDate"
41			DATE_DUE = "dueDate"
42			DATE_START = "startDate"
43			DATE_STOP = "stopDate"
44			IS_INBOX = "start = 0"
45			IS_ANYTIME = "start = 1"
46			IS_SOMEDAY = "start = 2"
47			IS_SCHEDULED = f"{DATE_START} IS NOT NULL"
48			IS_NOT_SCHEDULED = f"{DATE_START} IS NULL"
49			IS_DUE = f"{DATE_DUE} IS NOT NULL"
50			IS_NOT_DUE = f"{DATE_DUE} IS NULL"
51			IS_RECURRING = "recurrenceRule IS NOT NULL"
52			IS_NOT_RECURRING = "recurrenceRule IS NULL"
53			IS_TASK = "type = 0"
54			IS_PROJECT = "type = 1"
55			IS_HEADING = "type = 2"
56			IS_TRASHED = "trashed = 1"
57			IS_NOT_TRASHED = "trashed = 0"
58			IS_OPEN = "status = 0"
59			IS_CANCELLED = "status = 2"
60			IS_DONE = "status = 3"
61			RECURRING_IS_NOT_PAUSED = "instanceCreationPaused = 0"
62			RECURRING_HAS_NEXT_STARTDATE = "nextInstanceStartDate IS NOT NULL"
63			MODE_TASK = "type = 0"
64			MODE_PROJECT = "type = 1"
65
66			# Variables
67			debug = False
68			user = getpass.getuser()
69			database = f"/Users/{user}/{FILE_DB}"
70			filter = ""
71			tag_waiting = "Waiting"
72			tag_mit = "MIT"
73			tag_cleanup = "Cleanup"
74			anonymize = False
75			config = configparser.ConfigParser()
76			config.read(str(Path.home()) + '/.kanbanviewrc')
77
78			# pylint: disable=R0913
79			def __init__(self,
80			database=None,
81			tag_waiting=None,
82			tag_mit=None,
83			tag_cleanup=None,
84			anonymize=None):
85
86			cfg = self.get_from_config(self.config, tag_waiting, 'TAG_WAITING')
87			self.tag_waiting = cfg if cfg else self.tag_waiting
88
89			cfg = self.get_from_config(self.config, anonymize, 'ANONYMIZE')
90			self.anonymize = cfg if cfg else self.anonymize
91
92			cfg = self.get_from_config(self.config, tag_mit, 'TAG_MIT')
93			self.tag_mit = cfg if cfg else self.tag_mit
94
95			cfg = self.get_from_config(self.config, tag_cleanup, 'TAG_CLEANUP')
96			self.tag_cleanup = cfg if cfg else self.tag_cleanup
97
98			cfg = self.get_from_config(self.config, database, 'THINGSDB')
99			self.database = cfg if cfg else self.database
100
101			@staticmethod
102			def get_from_config(config, variable, preference):
103			"""Set variable. Priority: input, environment, config"""
104			result = None
105			if variable is not None:
106			result = variable
107			elif environ.get(preference):
108			result = environ.get(preference)
109			elif 'DATABASE' in config and preference in config['DATABASE']:
110			result = path.expanduser(config['DATABASE'][preference])
111			return result
112
113			@staticmethod
114			def anonymize_string(string):
115			"""Scramble text."""
116			if string is None:
117			return None
118			string = list(string)
119			shuffle(string)
120			string = ''.join(string)
121			return string
122
123			@staticmethod
124			def dict_factory(cursor, row):
125			"""Convert SQL result into a dictionary"""
126			dictionary = {}
127			for idx, col in enumerate(cursor.description):
128			dictionary[col[0]] = row[idx]
129			return dictionary
130
131			def anonymize_tasks(self, tasks):
132			"""Scramble output for screenshots."""
133			if self.anonymize:
134			for task in tasks:
135			task['title'] = self.anonymize_string(task['title'])
136			task['context'] = self.anonymize_string(task['context'])
137			return tasks
138
139			def get_inbox(self):
140			"""Get all tasks from the inbox."""
141			query = f"""
142			TASK.{self.IS_NOT_TRASHED} AND
143			TASK.{self.IS_TASK} AND
144			TASK.{self.IS_OPEN} AND
145			TASK.{self.IS_INBOX}
146			ORDER BY TASK.duedate DESC , TASK.todayIndex
147			"""
148			return self.get_rows(query)
149
150			def get_today(self):
151			"""Get all tasks from the today list."""
152			query = f"""
153			TASK.{self.IS_NOT_TRASHED} AND
154			TASK.{self.IS_TASK} AND
155			TASK.{self.IS_OPEN} AND
156			(TASK.{self.IS_ANYTIME} OR (
157			TASK.{self.IS_SOMEDAY} AND
158			TASK.{self.DATE_START} <= strftime('%s', 'now')
159			)
160			) AND
161			TASK.{self.IS_SCHEDULED} AND (
162			(
163			PROJECT.title IS NULL OR (
164			PROJECT.{self.IS_NOT_TRASHED}
165			)
166			) AND (
167			HEADPROJ.title IS NULL OR (
168			HEADPROJ.{self.IS_NOT_TRASHED}
169			)
170			)
171			)
172			ORDER BY TASK.duedate DESC , TASK.todayIndex
173			"""
174			return self.get_rows(query)
175
176			def get_someday(self):
177			"""Get someday tasks."""
178			query = f"""
179			TASK.{self.IS_NOT_TRASHED} AND
180			TASK.{self.IS_TASK} AND
181			TASK.{self.IS_OPEN} AND
182			TASK.{self.IS_SOMEDAY} AND
183			TASK.{self.IS_NOT_SCHEDULED} AND
184			TASK.{self.IS_NOT_RECURRING} AND (
185			(
186			PROJECT.title IS NULL OR (
187			PROJECT.{self.IS_NOT_TRASHED}
188			)
189			) AND (
190			HEADPROJ.title IS NULL OR (
191			HEADPROJ.{self.IS_NOT_TRASHED}
192			)
193			)
194			)
195			ORDER BY TASK.duedate DESC, TASK.creationdate DESC
196			"""
197			return self.get_rows(query)
198
199			def get_upcoming(self):
200			"""Get upcoming tasks."""
201			query = f"""
202			TASK.{self.IS_NOT_TRASHED} AND
203			TASK.{self.IS_TASK} AND
204			TASK.{self.IS_OPEN} AND
205			TASK.{self.IS_SOMEDAY} AND
206			TASK.{self.IS_SCHEDULED} AND
207			TASK.{self.IS_NOT_RECURRING} AND (
208			(
209			PROJECT.title IS NULL OR (
210			PROJECT.{self.IS_NOT_TRASHED}
211			)
212			) AND (
213			HEADPROJ.title IS NULL OR (
214			HEADPROJ.{self.IS_NOT_TRASHED}
215			)
216			)
217			)
218			ORDER BY TASK.startdate, TASK.todayIndex
219			"""
220			return self.get_rows(query)
221
222			def get_waiting(self):
223			"""Get waiting tasks."""
224			return self.get_tag(self.tag_waiting)
225
226			def get_mit(self):
227			"""Get most important tasks."""
228			return self.get_tag(self.tag_mit)
229
230			def get_tag(self, tag):
231			"""Get task with specific tag"""
232			query = f"""
233			TASK.{self.IS_NOT_TRASHED} AND
234			TASK.{self.IS_TASK} AND
235			TASK.{self.IS_OPEN} AND
236			TASK.{self.IS_NOT_RECURRING} AND
237			TAGS.tags=(SELECT uuid FROM {self.TABLE_TAG}
238			WHERE title='{tag}'
239			)
240			AND (
241			(
242			PROJECT.title IS NULL OR (
243			PROJECT.{self.IS_NOT_TRASHED}
244			)
245			) AND (
246			HEADPROJ.title IS NULL OR (
247			HEADPROJ.{self.IS_NOT_TRASHED}
248			)
249			)
250			)
251			ORDER BY TASK.duedate DESC , TASK.todayIndex
252			"""
253			return self.get_rows(query)
254
255			def get_tag_today(self, tag):
256			"""Get today tasks with specific tag"""
257			query = f"""
258			TASK.{self.IS_NOT_TRASHED} AND
259			TASK.{self.IS_TASK} AND
260			TASK.{self.IS_OPEN} AND
261			(TASK.{self.IS_ANYTIME} OR (
262			TASK.{self.IS_SOMEDAY} AND
263			TASK.{self.DATE_START} <= strftime('%s', 'now')
264			)
265			) AND
266			TAGS.tags=(SELECT uuid FROM {self.TABLE_TAG}
267			WHERE title='{tag}') AND
268			TASK.{self.IS_SCHEDULED} AND (
269			(
270			PROJECT.title IS NULL OR (
271			PROJECT.{self.IS_NOT_TRASHED}
272			)
273			) AND (
274			HEADPROJ.title IS NULL OR (
275			HEADPROJ.{self.IS_NOT_TRASHED}
276			)
277			)
278			)
279			ORDER BY TASK.duedate DESC , TASK.todayIndex
280			"""
281			return self.get_rows(query)
282
283			def get_anytime(self):
284			"""Get anytime tasks."""
285			query = f"""
286			TASK.{self.IS_NOT_TRASHED} AND
287			TASK.{self.IS_TASK} AND
288			TASK.{self.IS_OPEN} AND
289			TASK.{self.IS_ANYTIME} AND
290			TASK.{self.IS_NOT_SCHEDULED} AND (
291			(
292			PROJECT.title IS NULL OR (
293			PROJECT.{self.IS_ANYTIME} AND
294			PROJECT.{self.IS_NOT_SCHEDULED} AND
295			PROJECT.{self.IS_NOT_TRASHED}
296			)
297			) AND (
298			HEADPROJ.title IS NULL OR (
299			HEADPROJ.{self.IS_ANYTIME} AND
300			HEADPROJ.{self.IS_NOT_SCHEDULED} AND
301			HEADPROJ.{self.IS_NOT_TRASHED}
302			)
303			)
304			)
305			ORDER BY TASK.duedate DESC , TASK.todayIndex
306			"""
307			if self.filter:
308			# ugly hack for Kanban task view on project
309			query = f"""
310			TASK.{self.IS_NOT_TRASHED} AND
311			TASK.{self.IS_TASK} AND
312			TASK.{self.IS_OPEN} AND
313			TASK.{self.IS_ANYTIME} AND
314			TASK.{self.IS_NOT_SCHEDULED} AND (
315			(
316			PROJECT.title IS NULL OR (
317			PROJECT.{self.IS_NOT_TRASHED}
318			)
319			) AND (
320			HEADPROJ.title IS NULL OR (
321			HEADPROJ.{self.IS_NOT_TRASHED}
322			)
323			)
324			)
325			ORDER BY TASK.duedate DESC , TASK.todayIndex
326			"""
327			return self.get_rows(query)
328
329			def get_completed(self):
330			"""Get completed tasks."""
331			query = f"""
332			TASK.{self.IS_NOT_TRASHED} AND
333			TASK.{self.IS_TASK} AND
334			TASK.{self.IS_DONE}
335			ORDER BY TASK.{self.DATE_STOP}
336			"""
337			return self.get_rows(query)
338
339			def get_cancelled(self):
340			"""Get cancelled tasks."""
341			query = f"""
342			TASK.{self.IS_NOT_TRASHED} AND
343			TASK.{self.IS_TASK} AND
344			TASK.{self.IS_CANCELLED}
345			ORDER BY TASK.{self.DATE_STOP}
346			"""
347			return self.get_rows(query)
348
349			def get_trashed(self):
350			"""Get trashed tasks."""
351			query = f"""
352			TASK.{self.IS_TRASHED} AND
353			TASK.{self.IS_TASK}
354			ORDER BY TASK.{self.DATE_STOP}
355			"""
356			return self.get_rows(query)
357
358			def get_projects(self):
359			"""Get projects."""
360			query = f"""
361			SELECT
362			TASK.uuid,
363			TASK.title,
364			NULL as context,
365			(SELECT COUNT(uuid)
366			FROM TMTask AS PROJECT_TASK
367			WHERE
368			PROJECT_TASK.project = TASK.uuid AND
369			PROJECT_TASK.{self.IS_NOT_TRASHED} AND
370			PROJECT_TASK.{self.IS_OPEN}
371			) AS size
372			FROM
373			{self.TABLE_TASK} AS TASK
374			WHERE
375			TASK.{self.IS_NOT_TRASHED} AND
376			TASK.{self.IS_PROJECT} AND
377			TASK.{self.IS_OPEN}
378			ORDER BY TASK.title COLLATE NOCASE
379			"""
380			return self.execute_query(query)
381
382			def get_areas(self):
383			"""Get areas."""
384			query = f"""
385			SELECT
386			AREA.uuid AS uuid,
387			AREA.title AS title,
388			(SELECT COUNT(uuid)
389			FROM TMTask AS PROJECT
390			WHERE
391			PROJECT.area = AREA.uuid AND
392			PROJECT.{self.IS_NOT_TRASHED} AND
393			PROJECT.{self.IS_OPEN}
394			) AS size
395			FROM
396			{self.TABLE_AREA} AS AREA
397			ORDER BY AREA.title COLLATE NOCASE
398			"""
399			return self.execute_query(query)
400
401			def get_all(self):
402			"""Get all tasks."""
403			query = f"""
404			TASK.{self.IS_NOT_TRASHED} AND
405			TASK.{self.IS_TASK} AND (
406			(
407			PROJECT.title IS NULL OR (
408			PROJECT.{self.IS_NOT_TRASHED}
409			)
410			) AND (
411			HEADPROJ.title IS NULL OR (
412			HEADPROJ.{self.IS_NOT_TRASHED}
413			)
414			)
415			)
416			"""
417			return self.get_rows(query)
418
419			def get_due(self):
420			"""Get due tasks."""
421			query = f"""
422			TASK.{self.IS_NOT_TRASHED} AND
423			TASK.{self.IS_TASK} AND
424			TASK.{self.IS_OPEN} AND
425			TASK.{self.IS_DUE} AND (
426			(
427			PROJECT.title IS NULL OR (
428			PROJECT.{self.IS_NOT_TRASHED}
429			)
430			) AND (
431			HEADPROJ.title IS NULL OR (
432			HEADPROJ.{self.IS_NOT_TRASHED}
433			)
434			)
435			)
436			ORDER BY TASK.{self.DATE_DUE}
437			"""
438			return self.get_rows(query)
439
440			def get_lint(self):
441			"""Get tasks that float around"""
442			query = f"""
443			TASK.{self.IS_NOT_TRASHED} AND
444			TASK.{self.IS_OPEN} AND
445			TASK.{self.IS_TASK} AND
446			(TASK.{self.IS_SOMEDAY} OR TASK.{self.IS_ANYTIME}) AND
447			TASK.project IS NULL AND
448			TASK.area IS NULL AND
449			TASK.actionGroup IS NULL
450			"""
451			return self.get_rows(query)
452
453			def get_empty_projects(self):
454			"""Get projects that are empty"""
455			query = f"""
456			TASK.{self.IS_NOT_TRASHED} AND
457			TASK.{self.IS_OPEN} AND
458			TASK.{self.IS_PROJECT} AND
459			TASK.{self.IS_ANYTIME}
460			GROUP BY TASK.uuid
461			HAVING
462			(SELECT COUNT(uuid)
463			FROM TMTask AS PROJECT_TASK
464			WHERE
465			PROJECT_TASK.project = TASK.uuid AND
466			PROJECT_TASK.{self.IS_NOT_TRASHED} AND
467			PROJECT_TASK.{self.IS_OPEN} AND
468			(PROJECT_TASK.{self.IS_ANYTIME} OR
469			PROJECT_TASK.{self.IS_SCHEDULED} OR
470			(PROJECT_TASK.{self.IS_RECURRING} AND
471			PROJECT_TASK.{self.RECURRING_IS_NOT_PAUSED} AND
472			PROJECT_TASK.{self.RECURRING_HAS_NEXT_STARTDATE}
473			)
474			)
475			) = 0
476			"""
477			return self.get_rows(query)
478
479			def get_largest_projects(self):
480			"""Get projects that are empty"""
481			query = f"""
482			SELECT
483			TASK.uuid,
484			TASK.title AS title,
485			creationDate AS created,
486			userModificationDate AS modified,
487			(SELECT COUNT(uuid)
488			FROM TMTask AS PROJECT_TASK
489			WHERE
490			PROJECT_TASK.project = TASK.uuid AND
491			PROJECT_TASK.{self.IS_NOT_TRASHED} AND
492			PROJECT_TASK.{self.IS_OPEN}
493			) AS tasks
494			FROM
495			{self.TABLE_TASK} AS TASK
496			WHERE
497			TASK.{self.IS_NOT_TRASHED} AND
498			TASK.{self.IS_OPEN} AND
499			TASK.{self.IS_PROJECT}
500			GROUP BY TASK.uuid
501			ORDER BY tasks COLLATE NOCASE DESC
502			"""
503			return self.execute_query(query)
504
505			def get_daystats(self):
506			"""Get a history of task activities"""
507			days = 365
508			query = f"""
509			WITH RECURSIVE timeseries(x) AS (
510			SELECT 0
511			UNION ALL
512			SELECT x+1 FROM timeseries
513			LIMIT {days}
514			)
515			SELECT
516			date(julianday("now", "-{days} days"),
517			"+" \|\| x \|\| " days") as date,
518			CREATED.TasksCreated as created,
519			CLOSED.TasksClosed as completed,
520			CANCELLED.TasksCancelled as cancelled,
521			TRASHED.TasksTrashed as trashed
522			FROM timeseries
523			LEFT JOIN
524			(SELECT COUNT(uuid) AS TasksCreated,
525			date(creationDate,"unixepoch") AS DAY
526			FROM {self.TABLE_TASK} AS TASK
527			WHERE DAY NOT NULL
528			AND TASK.{self.IS_TASK}
529			GROUP BY DAY)
530			AS CREATED ON CREATED.DAY = date
531			LEFT JOIN
532			(SELECT COUNT(uuid) AS TasksCancelled,
533			date(stopDate,"unixepoch") AS DAY
534			FROM {self.TABLE_TASK} AS TASK
535			WHERE DAY NOT NULL
536			AND TASK.{self.IS_CANCELLED} AND TASK.{self.IS_TASK}
537			GROUP BY DAY)
538			AS CANCELLED ON CANCELLED.DAY = date
539			LEFT JOIN
540			(SELECT COUNT(uuid) AS TasksTrashed,
541			date(userModificationDate,"unixepoch") AS DAY
542			FROM {self.TABLE_TASK} AS TASK
543			WHERE DAY NOT NULL
544			AND TASK.{self.IS_TRASHED} AND TASK.{self.IS_TASK}
545			GROUP BY DAY)
546			AS TRASHED ON TRASHED.DAY = date
547			LEFT JOIN
548			(SELECT COUNT(uuid) AS TasksClosed,
549			date(stopDate,"unixepoch") AS DAY
550			FROM {self.TABLE_TASK} AS TASK
551			WHERE DAY NOT NULL
552			AND TASK.{self.IS_DONE} AND TASK.{self.IS_TASK}
553			GROUP BY DAY)
554			AS CLOSED ON CLOSED.DAY = date
555			"""
556			return self.execute_query(query)
557
558			def get_minutes_today(self):
559			"""Count the planned minutes for today."""
560			query = f"""
561			SELECT
562			SUM(TAG.title) AS minutes
563			FROM
564			{self.TABLE_TASK} AS TASK
565			LEFT OUTER JOIN
566			TMTask PROJECT ON TASK.project = PROJECT.uuid
567			LEFT OUTER JOIN
568			TMArea AREA ON TASK.area = AREA.uuid
569			LEFT OUTER JOIN
570			TMTask HEADING ON TASK.actionGroup = HEADING.uuid
571			LEFT OUTER JOIN
572			TMTask HEADPROJ ON HEADING.project = HEADPROJ.uuid
573			LEFT OUTER JOIN
574			TMTaskTag TAGS ON TASK.uuid = TAGS.tasks
575			LEFT OUTER JOIN
576			TMTag TAG ON TAGS.tags = TAG.uuid
577			WHERE
578			printf("%d", TAG.title) = TAG.title AND
579			TASK.{self.IS_NOT_TRASHED} AND
580			TASK.{self.IS_TASK} AND
581			TASK.{self.IS_OPEN} AND
582			TASK.{self.IS_ANYTIME} AND
583			TASK.{self.IS_SCHEDULED} AND (
584			(
585			PROJECT.title IS NULL OR (
586			PROJECT.{self.IS_NOT_TRASHED}
587			)
588			) AND (
589			HEADPROJ.title IS NULL OR (
590			HEADPROJ.{self.IS_NOT_TRASHED}
591			)
592			)
593			)
594			"""
595			return self.execute_query(query)
596
597			def get_cleanup(self):
598			"""Tasks and projects that need work."""
599			result = []
600			result.extend(self.get_lint())
601			result.extend(self.get_empty_projects())
602			result.extend(self.get_tag(self.tag_cleanup))
603			result = [i for n, i in enumerate(result) if i not in result[n + 1:]]
604			return result
605
606			@staticmethod
607			def get_not_implemented():
608			"""Not implemented warning."""
609			return [{"title": "not implemented"}]
610
611			def get_rows(self, sql):
612			"""Query Things database."""
613
614			sql = f"""
615			SELECT DISTINCT
616			TASK.uuid,
617			TASK.title,
618			CASE
619			WHEN AREA.title IS NOT NULL THEN AREA.title
620			WHEN PROJECT.title IS NOT NULL THEN PROJECT.title
621			WHEN HEADING.title IS NOT NULL THEN HEADING.title
622			END AS context,
623			CASE
624			WHEN AREA.uuid IS NOT NULL THEN AREA.uuid
625			WHEN PROJECT.uuid IS NOT NULL THEN PROJECT.uuid
626			END AS context_uuid,
627			CASE
628			WHEN TASK.recurrenceRule IS NULL
629			THEN date(TASK.dueDate,"unixepoch")
630			ELSE NULL
631			END AS due,
632			date(TASK.creationDate,"unixepoch") as created,
633			date(TASK.userModificationDate,"unixepoch") as modified,
634			date(TASK.startDate,"unixepoch") as started,
635			date(TASK.stopDate,"unixepoch") as stopped,
636			(SELECT COUNT(uuid)
637			FROM TMTask AS PROJECT_TASK
638			WHERE
639			PROJECT_TASK.project = TASK.uuid AND
640			PROJECT_TASK.{self.IS_NOT_TRASHED} AND
641			PROJECT_TASK.{self.IS_OPEN}
642			) AS size,
643			CASE
644			WHEN TASK.type = 0 THEN 'task'
645			WHEN TASK.type = 1 THEN 'project'
646			WHEN TASK.type = 2 THEN 'heading'
647			END AS type
648			FROM
649			{self.TABLE_TASK} AS TASK
650			LEFT OUTER JOIN
651			{self.TABLE_TASK} PROJECT ON TASK.project = PROJECT.uuid
652			LEFT OUTER JOIN
653			{self.TABLE_AREA} AREA ON TASK.area = AREA.uuid
654			LEFT OUTER JOIN
655			{self.TABLE_TASK} HEADING ON TASK.actionGroup = HEADING.uuid
656			LEFT OUTER JOIN
657			{self.TABLE_TASK} HEADPROJ ON HEADING.project = HEADPROJ.uuid
658			LEFT OUTER JOIN
659			{self.TABLE_TASKTAG} TAGS ON TASK.uuid = TAGS.tasks
660			LEFT OUTER JOIN
661			{self.TABLE_TAG} TAG ON TAGS.tags = TAG.uuid
662			WHERE
663			{self.filter}
664			{sql}
665			"""
666
667			return self.execute_query(sql)
668
669			def execute_query(self, sql):
670			"""Run the actual query"""
671			if self.debug is True:
672			print(sql)
673			try:
674			connection = sqlite3.connect(
675			'file:' + self.database + '?mode=ro', uri=True)
676			connection.row_factory = Things3.dict_factory
677			cursor = connection.cursor()
678			cursor.execute(sql)
679			tasks = cursor.fetchall()
680			tasks = self.anonymize_tasks(tasks)
681			if self.debug:
682			for task in tasks:
683			print(task)
684			return tasks
685			except sqlite3.OperationalError as error:
686			print(f"Could not query the database at: {self.database}.")
687			print(f"Details: {error}.")
688			sys.exit(2)
689
690			# pylint: disable=C0103
691			def mode_project(self):
692			"""Hack to switch to project view"""
693			self.IS_TASK = self.MODE_PROJECT
694
695			# pylint: disable=C0103
696			def mode_task(self):
697			"""Hack to switch to project view"""
698			self.IS_TASK = self.MODE_TASK
699
700			functions = {
701			"inbox": get_inbox,
702			"today": get_today,
703			"next": get_anytime,
704			"backlog": get_someday,
705			"upcoming": get_upcoming,
706			"waiting": get_waiting,
707			"mit": get_mit,
708			"completed": get_completed,
709			"cancelled": get_cancelled,
710			"trashed": get_trashed,
711			"projects": get_projects,
712			"areas": get_areas,
713			"all": get_all,
714			"due": get_due,
715			"lint": get_lint,
716			"empty": get_empty_projects,
717			"cleanup": get_cleanup,
718			"top-proj": get_largest_projects,
719			"stats-day": get_daystats,
720			"stats-min-today": get_minutes_today
721			}
722

AlexanderWillner / KanbanView

Push — master ( 9d0b37...891a54 )

things3.things3.Things3.get_completed() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like