Code development platform for open source projects from the European Union institutions

Skip to content
Snippets Groups Projects
Commit 943664b9 authored by fandrem's avatar fandrem
Browse files

David refactor and add comment method

parent 1d4f53e2
No related branches found
No related tags found
No related merge requests found
from logging import getLogger
from .connector import api
#from library.leankit.models.Card import Card
#from library.leankit.models.ListResponse import BoardListResponse
from .connector import api
# from .connector import api
__author__ = "DVE"
__email__ = "david.vernazobres@ext.ec.europa.eu"
......@@ -9,9 +12,9 @@ __license__ = "MIT"
__version__ = "1.0.0"
def get_users():
def get_users(limit=200):
log.debug('Getting users')
return api._get('/io/user?limit=200')
return api._get('/io/user?limit=' + str(200))
def get_user(user_info):
......@@ -24,34 +27,69 @@ def create_user(duser):
return api._post('/io/user', duser)
def get_invitations():
def get_invitations(limit=200):
log.debug('Getting intation user pending')
return api._get('/io/invitation?limit=200')
return api._get('/io/invitation?limit=' + str(limit))
def get_boards():
def get_boards(limit=100):
log.debug('Getting boards')
return api._get('/io/board?limit=100')
return api._get('/io/board?limit=' + str(limit))
def get_board(board_name):
log.debug('Getting board')
return api._get('/io/board?limit=100&search=' + board_name)
def get_board_by_name(board_name: str, limit=100) -> dict:
search_results = search_board(board_name)
for result in search_results.boards:
if result["title"] == board_name:
return result
return None
def get_board(board_name: str, limit=100):
return get_board_by_name(board_name, limit)
#necessary check the import
#def search_board(board_name, limit=100) -> BoardListResponse:
# log.debug('Getting board')
# results = api._get('/io/board', url_params={'limit': str(limit), 'search': board_name})
# # print(results)
# list_result = BoardListResponse.decode(results)
# return list_result
def get_board_detail(board_id):
log.debug('Getting board')
return api._get('/io/board/' + board_id)
return api._get('/io/board/' + str(board_id))
def get_cards(board_id=None, limit=100, offset=0):
log.debug('Getting cards')
params={}
if board_id is not None:
params['board'] = str(board_id)
if limit is not None:
params['limit'] = str(limit)
if offset is not None:
params['offset'] = str(offset)
return api._get('/io/card/', params)
def get_card(card_id):
log.debug('Getting card')
return api._get('/io/card/' + card_id)
return api._get('/io/card/' + str(card_id))
def create_card(dcard):
log.debug('Assigning card')
return api._post('/io/card/', dcard)
#necessary check the import
#def create_card(card: Card, returnFullRecord=False):
# """
# see https://success.planview.com/Planview_LeanKit/LeanKit_API/01_v2/card/create
#
# :param card: Card-class instance
# :param returnFullRecord: Returns the full card record
# :return: full card object
# """
# log.debug('Assigning card')
# return api._post(f'/io/card/?returnFullRecord={str(bool(returnFullRecord)).lower}', dict(card))
def card_assign(assign_card):
......@@ -61,12 +99,50 @@ def card_assign(assign_card):
def get_cardtype_from_board(board_id):
log.debug('Getting card')
return api._get('/io/board/' + board_id + '/cardType')
return api._get('/io/board/' + str(board_id) + '/cardType')
def update_card(card_id, dcard):
log.debug('Updating card')
return api._patch('/io/card/' + card_id, dcard)
return api._patch('/io/card/' + str(card_id), dcard)
def add_comment(card_id, dcard):
log.debug('add comment to card')
return api._post('/io/card/' + str(card_id) + '/comment', dcard)
# HELPER METHODS
# get boards ids
def get_boards_ids(boards: dict = None) -> object:
"""
Helper method to enumerate boardname and id only as a list of tuples
Parameters
----------
boards : dict
use data from leankit.get_boards()
Returns
-------
eg. [('31512086836591', '[SEC]'), ('31512089953496', 'Compliance'), ('31512089554405', 'Coordination [NET]'),...]
"""
if not boards:
boards = get_boards()
return [(board['id'], board['title']) for board in boards['boards']]
def get_cards_by_board(board_id):
request_params = {
'board': board_id, # sets the board ID to be queried - filled in later by the program
'limit': '100000', # sets the limit on the number of cards that can be queried at a time
}
leankitData = api._get(f'/io/card', url_params=request_params)
return leankitData
'''
......
import logging
import requests
from requests import Response, Request
from pprint import pformat
from json import dumps as json_dumps
from . import env_user
class Leankit(object):
# see https://success.planview.com/Planview_LeanKit/LeanKit_API/01_v2/01-overview/rate-limiting
RESPONSE_ERROR_TOO_MANY_REQUESTS = 429
def __init__(self):
self.session = requests.Session()
......@@ -15,15 +19,15 @@ class Leankit(object):
self.session_token = env_user.bearer
self.headers = {'Content-Type': 'application/json', 'Authorization': 'bearer ' + self.session_token}
def _get(self, url):
def _get(self, url, url_params={}, **json_kwargs):
log.debug('GET {}'.format(url))
try:
request = self.session.get(self.base + url, verify=True, headers=self.headers, proxies=self.proxies)
request = self.session.get(self.base + url, params=url_params, verify=True, headers=self.headers, proxies=self.proxies)
except Exception as error:
raise ConnectionError("Unable to make request: {}".format(error))
if request.ok:
try:
response = request.json()
response = request.json(**json_kwargs)
return response
'''
else:
......@@ -35,9 +39,9 @@ class Leankit(object):
raise IOError("Invalid response")
else:
msg = 'Server responded with code {0.status_code}'.format(request)
msg += ' ' + str(request.text)
raise ConnectionError(msg)
def _post(self, url, params):
log.debug('POST {}'.format(url))
print('POST {}'.format(url))
......@@ -74,7 +78,6 @@ class Leankit(object):
msg = 'Server responded with code {0.status_code}'.format(request)
raise ConnectionError(msg)
def _patch(self, url, params):
log.debug('PATCH {}'.format(url))
print('PATCH {}'.format(url))
......@@ -111,6 +114,13 @@ class Leankit(object):
msg = 'Server responded with code {0.status_code}'.format(request)
raise ConnectionError(msg)
def check_api_limits(self, response: Response):
# see https://success.planview.com/Planview_LeanKit/LeanKit_API/01_v2/01-overview/rate-limiting
x_ratelimit_limit = response.headers["X-RateLimit-Limit"]
x_ratelimit_remaining = response.headers["X-RateLimit-Remaining"]
x_ratelimit_reset = response.headers["X-RateLimit-Reset"]
retry_after = response.headers["Retry-After"]
log = logging.getLogger(__name__)
api = Leankit()
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment