things3.things3.Things3.set_config() - Code Metrics - Inspection of "added support for preferences + enhancements for i..." - AlexanderWillner/KanbanView - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 914db1...556204 )

by Alexander

created 2020-05-01 13:17 UTC

things3.things3.Things3.set_config() A

↳ Parent: things3.things3

Complexity

Conditions

Size

Total Lines	8
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
eloc	7
nop	4
dl	0
loc	8
rs	9.3333
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_CONFIG = str(Path.home()) + '/.kanbanviewrc'
    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(FILE_CONFIG)

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

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

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

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

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

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

    def set_config(self, key, value, domain='DATABASE'):
        """Write variable to config."""
        if domain not in self.config:
            self.config.add_section(domain)
        if value is not None and key is not None:
            self.config.set(domain, str(key), str(value))
            with open(self.FILE_CONFIG, "w+") as configfile:
                self.config.write(configfile)

    def get_config(self, key, domain='DATABASE'):
        """Get variable from config."""
        result = None
        if domain in self.config and key in self.config[domain]:
            result = path.expanduser(self.config[domain][key])
        return result

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

AlexanderWillner / KanbanView

Push — master ( 914db1...556204 )

things3.things3.Things3.set_config() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like