mirrors/todoist-taskwarrior
1
0
Fork 0
mirror of https://git.webmeisterei.com/webmeisterei/todoist-taskwarrior.git synced 2023-12-21 10:23:00 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add twoway sync

Browse Source
This commit is contained in:
René Jochum 2019-08-05 04:16:17 +02:00
parent 81ee6d16dc
commit 74da3bee81
No known key found for this signature in database
GPG Key ID: 9E8B1C32F5F318A9
10 changed files with 468 additions and 252 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -38,3 +38,5 @@ pip-delete-this-directory.txt
# Environments
venv/
# vscode
.vscode/

1
LICENSE
View File

@ -1,6 +1,7 @@
MIT License
Copyright (c) 2018 Matt Snider
Copyright (c) 2019 Webmeisterei Informationstechnologie GmbH
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal

87
README.md
View File

@ -1,53 +1,54 @@
# todoist-taskwarrior
A tool for migrating Todoist tasks to Taskwarrior.
A tool for syncing Todoist tasks with Taskwarrior.
## Installation
```bash
git clone https://git.webmeisterei.com/webmeisterei/todoist-taskwarrior.git
cd todoist-taskwarrior/
```
- To install in Virtualenv:
```bash
virtualenv -p /usr/bin/python3 venv
venv/bin/pip install -r requirements.txt
venv/bin/python setup.py install
```
- To install global:
```bash
sudo pip3 install -r requirements.txt
sudo python3 setup.py install
```
## Configure
First optain a Todoist API key from the [Todoist Integrations Settings](https://todoist.com/prefs/integrations).
Now you can configure `titwsync` with (replace `./venv/bin/titwsync` with `titwsync` if you use todoist_taskwarrior without a virtualenv):
```sh
./venv/bin/titwsync configure --map-project Inbox= --map-project Company=work --map-project Company.SubProject=work.subproject --map-tag books=reading <TODOIST_API_KEY>
```
`titwsync configure` writes the configuration to `~/.titwsyncrc.yaml`, with the key: `taskwarrior.project_sync.PROJECT_NAME` you can enable or disable the sync of a whole project!
## Usage
Running the tool requires that your Todoist API key is available from the
environment under the name `TODOIST_API_KEY`. The key can be found or created in
the [Todoist Integrations Settings](https://todoist.com/prefs/integrations).
the ).
The main task is `migrate` which will import all tasks. Since Todoist's internal
The main task is `sync` which will sync all tasks. Since Todoist's internal
ID is saved with the task, subsequent runs will detect and skip duplicates:
```sh
$ python -m todoist_taskwarrior.cli migrate --help
Usage: cli.py migrate [OPTIONS]
Options:
-i, --interactive
--no-sync
--help Show this message and exit.
```
Using the `--interactive` flag will prompt the user for input for each task,
allowing the task to be modified before import:
Replace `./venv/bin/titwsync` with `titwsync` if you use todoist_taskwarrior without a virtualenv.
```sh
$ python -m todoist_taskwarrior.cli migrate --interactive
Task 1 of 315: Work on an open source project
tid: 142424242
name: Work on an open source project
project: Open Source
priority:
tags:
entry: 2019-01-18T12:00:00+00:00
due: 2019-01-21T17:00:00+00:00
recur: 3 days
```
By default, `migrate` will refetch all tasks from Todoist on each run. To skip
this step and use the cached data without refetching, use the --no-sync flag.
The flags `--map-project` and `--map-tag` can be specified multiple times to translate or completely remove specific flags
```sh
$ python -m todoist_taskwarrior.cli migrate \
--map-project Errands=chores \
--map-project 'XYZ Corp'=work \
--map-tag books=reading
./venv/bin/titwsync sync
```
## Development
@ -55,6 +56,14 @@ $ python -m todoist_taskwarrior.cli migrate \
### Testing
```sh
$ python -m pytest tests
python -m pytest tests
```
## License
Licensed under the MIT license.
## Authors
- 2018-2019 [matt-snider](https://github.com/matt-snider/todoist-taskwarrior)
- 2019- [webmeisterei](https://git.webmeisterei.com/webmeisterei/todoist-taskwarrior)

6
TODO.md
View File

@ -1,6 +0,0 @@
# TODO:
* `revert` command that deletes all tasks with `todoist_id`
* Save skipped list so we don't prompt user on next run
* Allow input of scheduled, wait

1
requirements.txt
View File

@ -1,5 +1,6 @@
Click==7.0
todoist-python==8.0.0
pyyaml
# Temporarily use this until upstream PR #121 is merged
# https://github.com/ralphbean/taskw/pull/121

37
setup.py Normal file
View File

@ -0,0 +1,37 @@
# -*- coding: utf-8 -*-
from setuptools import find_packages
from setuptools import setup
setup(
name='todoist_taskwarrior',
version='0.1.0.dev0',
description="Todoist <-> Taskwarrior two-way sync",
classifiers=[
"Programming Language :: Python",
"Programming Language :: Python :: 3.5",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
],
keywords='Todoist Taskwarrior',
author='René Jochum',
author_email='rene@webmeisterei.com',
url='https://git.webmeisterei.com/webmeisterei/todoist-taskwarrior',
license='MIT',
packages=find_packages('.', exclude=['ez_setup']),
package_dir={'': '.'},
include_package_data=True,
zip_safe=True,
install_requires=[
'setuptools',
'Click',
'todoist-python',
'taskw',
],
entry_points={
'console_scripts': [
'titwsync=todoist_taskwarrior.cli:cli'
],
}
)

553
todoist_taskwarrior/cli.py
View File

@ -1,27 +1,88 @@
import click
import os
import sys
import datetime
import dateutil.parser
import io
import yaml
from taskw import TaskWarrior
from todoist.api import TodoistAPI
from . import errors, io, utils, validation
from . import errors, log, utils, validation
# This is the location where the todoist
# data will be cached.
TODOIST_CACHE = '~/.todoist-sync/'
TITWSYNCRC = '~/.titwsyncrc.yaml'
config = None
todoist = None
taskwarrior = None
""" CLI Commands """
@click.group()
def cli():
"""Manage the migration of data from Todoist into Taskwarrior. """
pass
"""Two-way sync of Todoist and Taskwarrior. """
global config, todoist, taskwarrior
is_help_cmd = '-h' in sys.argv or '--help' in sys.argv
rcfile = os.path.expanduser(TITWSYNCRC)
with open(rcfile, 'r') as stream:
config = yaml.safe_load(stream)
if 'todoist' not in config or 'api_key' not in config['todoist'] \
and not is_help_cmd:
log.error('Run configure first. Exiting.')
exit(1)
todoist = TodoistAPI(config['todoist']['api_key'], cache=TODOIST_CACHE)
# Create the TaskWarrior client, overriding config to
# create a `todoist_id` field which we'll use to
# prevent duplicates
taskwarrior = TaskWarrior(config_overrides={
'uda.todoist_id.type': 'string',
'uda.todoist_sync.type': 'date',
})
@cli.command()
@click.option('-p', '--map-project', metavar='SRC=DST', multiple=True,
callback=validation.validate_map,
help='Project names specified will be translated '
'from SRC to DST. '
'If DST is omitted, the project will be unset when SRC matches.')
@click.option('-t', '--map-tag', metavar='SRC=DST', multiple=True,
callback=validation.validate_map,
help='Tags specified will be translated from SRC to DST. '
'If DST is omitted, the tag will be removed when SRC matches.')
@click.argument('todoist_api_key')
def configure(map_project, map_tag, todoist_api_key):
"""Configure sync.
Use --map-project to change or remove the project. Project hierarchies will
be period-delimited during conversion. For example in the following,
'Work Errands' and 'House Errands' will be both be changed to 'errands',
'Programming.Open Source' will be changed to 'oss', and the project will be
removed when it is 'Taxes':
\r
--map-project 'Work Errands'=errands
--map-project 'House Errands'=errands
--map-project 'Programming.Open Source'=oss
--map-project Taxes=
"""
data = {'todoist': {'api_key': todoist_api_key}, 'taskwarrior': {}}
data['todoist']['project_map'] = map_project
data['todoist']['tag_map'] = map_tag
data['taskwarrior']['project_sync'] = {
k: True for k in map_project.values()}
rcfile = os.path.expanduser(TITWSYNCRC)
with io.open(rcfile, 'w', encoding='utf8') as outfile:
yaml.dump(data, outfile, default_flow_style=False, allow_unicode=True)
@cli.command()
@ -36,12 +97,13 @@ def synchronize():
~/.todoist-sync
"""
with io.with_feedback('Syncing tasks with todoist'):
with log.with_feedback('Syncing tasks with todoist'):
todoist.sync()
@cli.command()
@click.confirmation_option(prompt=f'Are you sure you want to delete {TODOIST_CACHE}?')
@click.confirmation_option(
prompt=f'Are you sure you want to delete {TODOIST_CACHE}?')
def clean():
"""Remove the data stored in the Todoist task cache.
@ -53,238 +115,340 @@ def clean():
# Delete all files in directory
for file_entry in os.scandir(cache_dir):
with io.with_feedback(f'Removing file {file_entry.path}'):
with log.with_feedback(f'Removing file {file_entry.path}'):
os.remove(file_entry)
# Delete directory
with io.with_feedback(f'Removing directory {cache_dir}'):
with log.with_feedback(f'Removing directory {cache_dir}'):
os.rmdir(cache_dir)
@cli.command()
@click.option('-i', '--interactive', is_flag=True, default=False,
help='Interactively choose which tasks to import and modify them '
'during the import.')
@click.option('--sync/--no-sync', default=True,
help='Enable/disable Todoist synchronization of the local task cache.')
@click.option('-p', '--map-project', metavar='SRC=DST', multiple=True,
callback=validation.validate_map,
help='Project names specified will be translated from SRC to DST. '
'If DST is omitted, the project will be unset when SRC matches.')
@click.option('-t', '--map-tag', metavar='SRC=DST', multiple=True,
callback=validation.validate_map,
help='Tags specified will be translated from SRC to DST. '
'If DST is omitted, the tag will be removed when SRC matches.')
@click.pass_context
def migrate(ctx, interactive, sync, map_project, map_tag):
"""Migrate tasks from Todoist to Taskwarrior.
By default this command will synchronize with the Todoist servers
and then migrate all tasks to Taskwarrior.
Pass --no-sync to skip synchronization.
Passing -i or --interactive allows more control over the import, putting
the user into an interactive command loop. Per task, the user can decide
whether to skip, rename, change the priority, or change the tags, before
moving on to the next task.
Use --map-project to change or remove the project. Project hierarchies will
be period-delimited during conversion. For example in the following,
'Work Errands' and 'House Errands' will be both be changed to 'errands',
'Programming.Open Source' will be changed to 'oss', and the project will be
removed when it is 'Taxes':
\r
--map-project 'Work Errands'=errands
--map-project 'House Errands'=errands
--map-project 'Programming.Open Source'=oss
--map-project Taxes=
def sync(ctx):
"""Sync tasks between Todoist and Taskwarrior.
This command can be run multiple times and will not duplicate tasks.
This is tracked in Taskwarrior by setting and detecting the
`todoist_id` property on the task.
"""
if sync:
ctx.invoke(synchronize)
# Sync todoist to cache
ctx.invoke(synchronize)
tasks = todoist.items.all()
io.important(f'Starting migration of {len(tasks)} tasks...')
for idx, task in enumerate(tasks):
data = {}
tid = data['tid'] = task['id']
name = data['name'] = task['content']
ti_project_list = _ti_project_list()
default_project = utils.try_map(config['todoist']['project_map'], 'Inbox')
# Log message and check if exists
io.important(f'Task {idx + 1} of {len(tasks)}: {name}')
if check_task_exists(tid):
io.info(f'Already exists (todoist_id={tid})')
config_ps = config['taskwarrior']['project_sync']
# Sync Taskwarrior->Todoist
tw_tasks = taskwarrior.load_tasks()
log.important(f'Starting to sync tasks from Taskwarrior...')
for tw_task in tw_tasks['pending']:
if 'project' not in tw_task:
tw_task['project'] = default_project
desc = tw_task['description']
project = tw_task['project']
if (tw_task['project'] not in config_ps or
not config_ps[tw_task['project']]):
log.warn(f'Ignoring Task {desc} ({project})')
continue
# Project
p = todoist.projects.get_by_id(task['project_id'])
project_hierarchy = [p]
while p['parent_id']:
p = todoist.projects.get_by_id(p['parent_id'])
project_hierarchy.insert(0, p)
# Log message
log.important(f'Sync Task {desc} ({project})')
project_name = '.'.join(p['name'] for p in project_hierarchy)
project_name = utils.try_map(
map_project,
project_name
)
data['project'] = utils.maybe_quote_ws(project_name)
if 'todoist_id' in tw_task:
ti_task = todoist.items.get_by_id(tw_task['todoist_id'])
if ti_task is None:
# Ti task has been deleted
taskwarrior.task_delete(uuid=tw_task['uuid'])
continue
# Priority
data['priority'] = utils.parse_priority(task['priority'])
ti_task = ti_task['item']
c_ti_task = _convert_ti_task(ti_task, ti_project_list)
# Tags
data['tags'] = [
utils.try_map(map_tag, todoist.labels.get_by_id(l_id)['name'])
for l_id in task['labels']
]
# Sync Todoist with Taskwarrior task
_sync_task(tw_task, c_ti_task, ti_project_list)
continue
# Dates
data['entry'] = utils.parse_date(task['date_added'])
data['due'] = utils.parse_due(utils.try_get_model_prop(task, 'due'))
data['recur'] = parse_recur_or_prompt(utils.try_get_model_prop(task, 'due'))
# Add Todoist task
_ti_add_task(tw_task, ti_project_list)
if not interactive:
add_task(**data)
with log.with_feedback('Syncing tasks with todoist'):
todoist.commit()
todoist.sync()
# Sync Todoist->Taskwarrior
tasks = todoist.items.all()
log.important(f'Starting sync of {len(tasks)} tasks from Todoist...')
for idx, ti_task in enumerate(tasks):
c_ti_task = _convert_ti_task(ti_task, ti_project_list)
desc = c_ti_task['description']
project = c_ti_task['project']
if (c_ti_task['project'] not in config_ps or
not config_ps[c_ti_task['project']]):
log.warn(f'Ignoring Task {desc} ({project})')
continue
# Log message
log.important(f'Sync Task {desc} ({project})')
# Sync Todoist with Taskwarrior task
_, tw_task = taskwarrior.get_task(todoist_id=ti_task['id'])
if bool(tw_task):
if 'project' not in tw_task:
tw_task['project'] = default_project
_sync_task(tw_task, c_ti_task, ti_project_list)
continue
# Add Taskwarrior task
_tw_add_task(c_ti_task)
with log.with_feedback('Syncing tasks with todoist'):
todoist.commit()
todoist.sync()
def _convert_ti_task(ti_task, ti_project_list):
data = {}
data['tid'] = ti_task['id']
data['description'] = ti_task['content']
# Project
project_name = ''
for project_name, p in ti_project_list.items():
if p['id'] == ti_task['project_id']:
break
data['project'] = project_name
# Priority
data['priority'] = utils.ti_priority_to_tw(ti_task['priority'])
# Tags
data['tags'] = [
utils.try_map(config['todoist']['tag_map'],
todoist.labels.get_by_id(l_id)['name'])
for l_id in ti_task['labels']
]
# Dates
data['entry'] = utils.parse_date(ti_task['date_added'])
data['due'] = utils.parse_due(utils.try_get_model_prop(ti_task, 'due'))
data['recur'] = parse_recur_or_prompt(
utils.try_get_model_prop(ti_task, 'due'))
data['status'] = 'completed' if ti_task['checked'] == 1 else 'pending'
return data
def _sync_task(tw_task, ti_task, ti_project_list):
if 'todoist_sync' in tw_task:
ti_stamp = dateutil.parser.parse(tw_task['todoist_sync']).timestamp()
tw_stamp = dateutil.parser.parse(tw_task['modified']).timestamp()
if tw_stamp > ti_stamp:
_ti_update_task(tw_task, ti_project_list)
else:
add_task_interactive(**data)
_tw_update_task(tw_task, ti_task)
else:
_tw_update_task(tw_task, ti_task)
def check_task_exists(tid):
""" Given a Taskwarrior ID, check if the task exists """
_, task = taskwarrior.get_task(todoist_id=tid)
return bool(task)
def add_task(tid, name, project, tags, priority, entry, due, recur):
def _tw_add_task(ti_task):
"""Add a taskwarrior task from todoist task
Returns the taskwarrior task.
"""
with io.with_feedback(f"Importing '{name}' ({project})"):
description = ti_task['description']
project = ti_task['project']
with log.with_feedback(f"Taskwarrior add '{description}' ({project})"):
return taskwarrior.task_add(
name,
project=project,
tags=tags,
priority=priority,
entry=entry,
due=due,
recur=recur,
todoist_id=tid,
ti_task['description'],
project=ti_task['project'],
tags=ti_task['tags'],
priority=ti_task['priority'],
entry=ti_task['entry'],
due=ti_task['due'],
recur=ti_task['recur'],
status=ti_task['status'],
todoist_id=ti_task['tid'],
todoist_sync=datetime.datetime.now(),
)
def add_task_interactive(**task_data):
"""Interactively add tasks
def _tw_update_task(tw_task, ti_task):
y - add task
n - skip task
d - change description
P - change project
p - change priority
t - change tags
r - change recur
q - quit immediately
? - print help
"""
callbacks = {
'y': lambda: task_data,
'n': lambda: task_data,
def _compare_value(item):
return ((ti_task[item] and item not in tw_task) or
(item in tw_task and tw_task[item] != ti_task[item]))
# Rename
'd': lambda: {
**task_data,
'name': io.prompt(
'Set name',
default=task_data['name'],
value_proc=lambda x: x.strip(),
),
},
description = ti_task['description']
project = ti_task['project']
with log.on_error(f"TW updating '{description}' ({project})"):
changed = False
# Edit tags
't': lambda: {
**task_data,
'tags': io.prompt(
'Set tags (space delimited)',
default=' '.join(task_data['tags']),
show_default=False,
value_proc=lambda x: x.split(' '),
),
},
if tw_task['description'] != ti_task['description']:
tw_task['description'] = ti_task['description']
changed = True
# Edit project
'P': lambda: {
**task_data,
'project': io.prompt(
'Set project',
default=task_data['project'],
),
},
if tw_task['project'] != ti_task['project']:
tw_task['project'] = ti_task['project']
changed = True
# Edit priority
'p': lambda: {
**task_data,
'priority': io.prompt(
'Set priority',
default='',
show_default=False,
type=click.Choice(['L', 'M', 'H', '']),
),
},
if _compare_value('tags'):
tw_task['tags'] = ti_task['tags']
changed = True
# Edit recur
'r': lambda: {
**task_data,
'recur': io.prompt(
'Set recurrence (todoist style)',
default='',
value_proc=validation.validate_recur,
),
},
if _compare_value('priority'):
tw_task['priority'] = ti_task['priority']
changed = True
if _compare_value('entry'):
tw_task['entry'] = ti_task['entry']
changed = True
if _compare_value('due'):
tw_task['due'] = ti_task['due']
changed = True
if _compare_value('recur'):
tw_task['recur'] = ti_task['recur']
changed = True
if tw_task['status'] != ti_task['status']:
tw_task['status'] = ti_task['status']
changed = True
if changed:
tid = ti_task['tid']
log.info(f'TW updating (todoist_id={tid})...', nl=False)
log.success('OK')
tw_task['todoist_sync'] = datetime.datetime.now()
taskwarrior.task_update(tw_task)
# Quit
'q': lambda: exit(1),
def _ti_update_task(tw_task, ti_project_list):
description = tw_task['description']
project = tw_task['project']
with log.on_error(f"Todoist update '{description}' ({project})"):
changed = False
# Help message
# Note: this echoes prompt help and then returns the
# task_data unchanged.
'?': lambda: io.warn('\n'.join([
x.strip() for x in
add_task_interactive.__doc__.split('\n')
])) or task_data,
}
ti_task = todoist.items.get_by_id(tw_task['todoist_id'])
response = None
while response not in ('y', 'n'):
io.task(task_data)
response = io.prompt(
"Import this task?",
type=click.Choice(callbacks.keys()),
show_choices=True,
if tw_task['description'] != ti_task['item']['content']:
ti_task['item']['content'] = tw_task['description']
changed = True
project = ti_project_list[tw_task['project']]
if ti_task['item']['project_id'] != project['id']:
changed = True
priority = 0
if 'priority' in tw_task:
priority = utils.tw_priority_to_ti(tw_task['priority'])
if ti_task['item']['priority'] != priority:
ti_task['item']['priority'] = priority
changed = True
if ((ti_task['item']['checked'] == 0 and
tw_task['status'] == 'completed') or
(ti_task['item']['checked'] == 1 and
tw_task['status'] == 'pending')):
changed = True
if changed:
tid = ti_task['item']['id']
log.info(f'Updating (todoist_id={tid})', nl=False)
log.success('OK')
todoist.items.update(tid, **ti_task)
# Move to another project
if ti_task['item']['project_id'] != project['id']:
todoist.items.move(tid, project_id=project['id'])
# Open/close ti task
if ti_task['item']['checked'] == 1 and \
tw_task['status'] == 'pending':
todoist.items.uncomplete(tid)
elif ti_task['item']['checked'] == 0 and \
tw_task['status'] == 'completed':
todoist.items.complete(tid)
tw_task['todoist_sync'] = datetime.datetime.now()
taskwarrior.task_update(tw_task)
else:
# Always set latest sync time so no more sync accures
tid = ti_task['item']['id']
log.info(f'TI updating (todoist_id={tid})...', nl=False)
log.success('OK')
tw_task['todoist_sync'] = datetime.datetime.now()
taskwarrior.task_update(tw_task)
def _ti_add_task(tw_task, ti_project_list):
description = tw_task['description']
project = tw_task['project']
with log.on_error(f"Todoist add '{description}' ({project})"):
# Add the item and commit the change
data = {}
if tw_task['project'] not in ti_project_list:
project = tw_task['project']
log.error(f'Project "{project}" not found on Todoist.')
return
data['project_id'] = ti_project_list[tw_task['project']]['id']
if 'priority' in tw_task:
data['priority'] = utils.tw_priority_to_ti(tw_task['priority'])
ti_task = todoist.items.add(tw_task['description'], **data)
todoist.commit()
tid = ti_task['id']
log.info(f'TI add (todoist_id={tid})')
tw_task['todoist_id'] = tid
tw_task['todoist_sync'] = datetime.datetime.now()
taskwarrior.task_update(tw_task)
def _ti_project_list():
result = {}
for p in todoist.projects.all():
project_hierarchy = [p]
pp = p
while pp['parent_id']:
pp = todoist.projects.get_by_id(p['parent_id'])
project_hierarchy.insert(0, pp)
project_name = '.'.join(p['name'] for p in project_hierarchy)
project_name = utils.try_map(
config['todoist']['project_map'],
project_name
)
result[utils.maybe_quote_ws(project_name)] = p
# Execute operation
task_data = callbacks[response]()
if response == 'n':
io.warn('Skipping task')
return
return add_task(**task_data)
return result
def parse_recur_or_prompt(due):
try:
return utils.parse_recur(due)
except errors.UnsupportedRecurrence:
io.error("Unsupported recurrence: '%s'. Please enter a valid value" % due['string'])
return io.prompt(
log.error(
"Unsupported recurrence: '%s'. "
"Please enter a valid value" % due['string'])
return log.prompt(
'Set recurrence (todoist style)',
default='',
value_proc=validation.validate_recur,
@ -294,19 +458,4 @@ def parse_recur_or_prompt(due):
""" Entrypoint """
if __name__ == '__main__':
is_help_cmd = '-h' in sys.argv or '--help' in sys.argv
todoist_api_key = os.getenv('TODOIST_API_KEY')
if todoist_api_key is None and not is_help_cmd:
io.error('TODOIST_API_KEY environment variable not specified. Exiting.')
exit(1)
todoist = TodoistAPI(todoist_api_key, cache=TODOIST_CACHE)
# Create the TaskWarrior client, overriding config to
# create a `todoist_id` field which we'll use to
# prevent duplicates
taskwarrior = TaskWarrior(config_overrides={
'uda.todoist_id.type': 'string',
})
cli()

7
todoist_taskwarrior/errors.py
View File

@ -7,3 +7,10 @@ class UnsupportedRecurrence(Exception):
super().__init__('Unsupported recurrence: %s' % date_string)
self.date_string = date_string
class TIItemNotFoundAfterCommit(Exception):
pass
class TIProjectNotFound(Exception):
pass

9
todoist_taskwarrior/io.py → todoist_taskwarrior/log.py
View File

@ -61,3 +61,12 @@ def with_feedback(description, success_status='OK', error_status='FAILED'):
else:
success(success_status)
@contextlib.contextmanager
def on_error(description, error_status='FAILED'):
try:
yield
except Exception as e:
info(f'{description}... ', nl=False)
error(f'{error_status} ({e})')
raise

17
todoist_taskwarrior/utils.py
View File

@ -24,16 +24,21 @@ def try_get_model_prop(m, key, default=None):
""" Priorities """
PRIORITY_MAP = {1: None, 2: 'L', 3: 'M', 4: 'H'}
TI_PRIORITY_MAP = {1: None, 2: 'L', 3: 'M', 4: 'H'}
def parse_priority(priority):
def ti_priority_to_tw(priority):
""" Converts a priority from Todoist to Taskwarrior.
Todoist saves priorities as 1, 2, 3, 4, whereas Taskwarrior uses L, M, H.
These values map very easily to eachother, as Todoist priority 1 indicates that
no priority has been set.
"""
return PRIORITY_MAP[int(priority)]
return TI_PRIORITY_MAP[int(priority)]
def tw_priority_to_ti(priority):
tw_priority_map = dict(map(reversed, TI_PRIORITY_MAP.items()))
return tw_priority_map[priority]
""" Strings """
@ -71,8 +76,10 @@ def parse_date(date):
"""
if not date:
return None
return dateutil.parser.parse(date).isoformat()
naive = dateutil.parser.parse(date)
dt = naive.replace(tzinfo=None)
return dt.strftime('%Y%m%dT%H%M%SZ')
def parse_recur(due):