add get_projects and get_tasks codes
This commit is contained in:
parent
08d0e70e26
commit
1605920d65
@ -3,6 +3,9 @@ Inkycal Todoist Module
|
|||||||
Copyright by aceinnolab
|
Copyright by aceinnolab
|
||||||
"""
|
"""
|
||||||
import arrow
|
import arrow
|
||||||
|
import json
|
||||||
|
import logging
|
||||||
|
import requests
|
||||||
|
|
||||||
from inkycal.modules.template import inkycal_module
|
from inkycal.modules.template import inkycal_module
|
||||||
from inkycal.custom import *
|
from inkycal.custom import *
|
||||||
@ -11,6 +14,95 @@ from todoist_api_python.api import TodoistAPI
|
|||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
class LoginVikunja():
|
||||||
|
def __init__(self, username, password, totp_passcode=None, token=None, api_url='http://192.168.50.10:3456/api/v1/'):
|
||||||
|
self.username = username
|
||||||
|
self.password = password
|
||||||
|
self.totp_passcode = totp_passcode
|
||||||
|
self.token = None
|
||||||
|
self.api_url = api_url
|
||||||
|
self._access_token = token
|
||||||
|
if self._access_token is None:
|
||||||
|
self._access_token = self.get_token()
|
||||||
|
|
||||||
|
def _create_url(self, path):
|
||||||
|
return self.api_url + path
|
||||||
|
|
||||||
|
"""returns the token from the login request"""
|
||||||
|
def _post_login_request(self, username, password, totp_passcode):
|
||||||
|
login_url = self._create_url('login')
|
||||||
|
payload = {
|
||||||
|
'long_token': True,
|
||||||
|
'username': username,
|
||||||
|
'password': password,
|
||||||
|
'totp_passcode': totp_passcode
|
||||||
|
}
|
||||||
|
return requests.post(login_url, json=payload, timeout=5)
|
||||||
|
|
||||||
|
def _get_access_token(self):
|
||||||
|
if not self._access_token:
|
||||||
|
token_json = self._post_login_request(self.username, self.password, self.totp_passcode)
|
||||||
|
if token_json.status_code == 200:
|
||||||
|
token = json.loads(token_json.text)
|
||||||
|
self._access_token = token['token']
|
||||||
|
else:
|
||||||
|
raise Exception('Login failed')
|
||||||
|
return self._access_token
|
||||||
|
|
||||||
|
def get_token(self):
|
||||||
|
return self._get_access_token()
|
||||||
|
|
||||||
|
def get_headers(self):
|
||||||
|
return {'Authorization': 'Bearer ' + self._get_access_token()}
|
||||||
|
|
||||||
|
class ApiVikunja():
|
||||||
|
def __init__(self, username, password, totp_passcode=None, token=None, api_url='http://192.168.50.10:3456/api/v1/'):
|
||||||
|
self.username = username
|
||||||
|
self.password = password
|
||||||
|
self.totp_passcode = totp_passcode
|
||||||
|
self.token = None
|
||||||
|
self.api_url = api_url
|
||||||
|
self._cache = {'projects': None, 'tasks': None, 'labels': None}
|
||||||
|
self._login = LoginVikunja(username, password, totp_passcode, token, api_url)
|
||||||
|
def _create_url(self, path):
|
||||||
|
return self.api_url + path
|
||||||
|
|
||||||
|
def _to_json(self, response):
|
||||||
|
try:
|
||||||
|
return response.json()
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f'Error parsing json: {e}')
|
||||||
|
raise e
|
||||||
|
|
||||||
|
def _get_json(self, url, params=None, headers=None):
|
||||||
|
if params is None:
|
||||||
|
params = {}
|
||||||
|
response = requests.get(url, params=params, headers=headers, timeout=5)
|
||||||
|
response.raise_for_status()
|
||||||
|
json_result = self._to_json(response)
|
||||||
|
total_pages = int(response.headers.get('x-pagination-total-pages', 1))
|
||||||
|
if total_pages > 1:
|
||||||
|
logger.debug('Trying to get all pages')
|
||||||
|
for page in range(2, total_pages + 1):
|
||||||
|
logger.debug(f'Getting page {page}')
|
||||||
|
params.update({'page': page})
|
||||||
|
response = requests.get(url, params=params, headers=headers, timeout=5)
|
||||||
|
response.raise_for_status()
|
||||||
|
json_result = json_result + self._to_json(response)
|
||||||
|
return json_result
|
||||||
|
|
||||||
|
def get_projects(self):
|
||||||
|
if self._cache['projects'] is None:
|
||||||
|
self._cache['projects'] = self._get_json(self._create_url('projects'), headers=self._login.get_headers())
|
||||||
|
return self._cache['projects']
|
||||||
|
def get_tasks(self, exclude_completed=True):
|
||||||
|
if self._cache['tasks'] is None:
|
||||||
|
url = self._create_url('tasks/all')
|
||||||
|
params = {'filter': 'done = false'} if exclude_completed else {}
|
||||||
|
self._cache['tasks'] = self._get_json(url, params, headers=self._login.get_headers()) or []
|
||||||
|
return self._cache['tasks']
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
class Todoist(inkycal_module):
|
class Todoist(inkycal_module):
|
||||||
"""Todoist api class
|
"""Todoist api class
|
||||||
@ -70,6 +162,10 @@ class Todoist(inkycal_module):
|
|||||||
"""Validate module-specific parameters"""
|
"""Validate module-specific parameters"""
|
||||||
if not isinstance(self.api_key, str):
|
if not isinstance(self.api_key, str):
|
||||||
print('api_key has to be a string: "Yourtopsecretkey123" ')
|
print('api_key has to be a string: "Yourtopsecretkey123" ')
|
||||||
|
|
||||||
|
def get_projects():
|
||||||
|
|
||||||
|
pass
|
||||||
|
|
||||||
def generate_image(self):
|
def generate_image(self):
|
||||||
"""Generate image for this module"""
|
"""Generate image for this module"""
|
||||||
|
@ -1,23 +1,35 @@
|
|||||||
import requests
|
import requests
|
||||||
import json
|
import json
|
||||||
|
from inkycal.modules.inkycal_vikunja import LoginVikunja
|
||||||
|
from inkycal.modules.inkycal_vikunja import ApiVikunja
|
||||||
|
import unittest
|
||||||
|
|
||||||
api_url = 'http://192.168.50.10:3456/api/v1/'
|
class TestLoginVikunja(unittest.TestCase):
|
||||||
|
def setUp(self) -> None:
|
||||||
|
self.api_url = 'http://192.168.50.10:3456/api/v1/'
|
||||||
|
self.username = 'iicd'
|
||||||
|
self.password = '9297519Mhz.'
|
||||||
|
self.totp_passcode = None
|
||||||
|
def test_post_login_request(self):
|
||||||
|
login = LoginVikunja(self.username, self.password, self.totp_passcode, self.api_url)
|
||||||
|
token_json = login._post_login_request(self.username, self.password, self.totp_passcode)
|
||||||
|
self.assertTrue(token_json.status_code == 200)
|
||||||
|
|
||||||
def _create_url(path):
|
class TestApiVikunja(unittest.TestCase):
|
||||||
return api_url + path
|
def setUp(self) -> None:
|
||||||
def _post_login_request(username, password, totp_passcode):
|
self.api_url = 'http://192.168.50.10:3456/api/v1/'
|
||||||
login_url = _create_url('login')
|
self.username = 'iicd'
|
||||||
payload = {
|
self.password = '9297519Mhz.'
|
||||||
'long_token': True,
|
self.totp_passcode = None
|
||||||
'username': username,
|
self.api = ApiVikunja(self.username, self.password, self.totp_passcode, None, self.api_url)
|
||||||
'password': password,
|
|
||||||
'totp_passcode': totp_passcode
|
def test_get_projects(self):
|
||||||
}
|
json_projects = self.api.get_projects()
|
||||||
return requests.post(login_url, json=payload, timeout=5)
|
print(json_projects)
|
||||||
|
self.assertTrue(json_projects)
|
||||||
if __name__ == '__main__':
|
|
||||||
username = 'iicd'
|
def test_get_tasks(self):
|
||||||
password = '9297519Mhz.'
|
json_tasks = self.api.get_tasks(exclude_completed=True)
|
||||||
totp_passcode = None
|
print(json_tasks)
|
||||||
result = _post_login_request(username, password, totp_passcode)
|
self.assertTrue(json_tasks)
|
||||||
print(result.json())
|
|
Loading…
Reference in New Issue
Block a user