69 lines
2.6 KiB
Python
69 lines
2.6 KiB
Python
from dataclasses import dataclass
|
|
|
|
from ActiveCollabAPI import AC_API_VERSION
|
|
from ActiveCollabAPI.ACAuthenticator import ACAuthenticator
|
|
from ActiveCollabAPI.ACClient import ACClient
|
|
from ActiveCollabAPI.ACTokenAuthenticator import ACTokenAuthenticator
|
|
from ActiveCollabAPI.AcAccount import AcAccount, account_from_json
|
|
from ActiveCollabAPI.AcLoginResponse import AcLoginResponse
|
|
from ActiveCollabAPI.AcProject import AcProject, project_from_json
|
|
from ActiveCollabAPI.AcSession import AcSession
|
|
from ActiveCollabAPI.AcToken import AcToken
|
|
from ActiveCollabAPI.AcUser import AcUser
|
|
|
|
|
|
|
|
class ActiveCollab:
|
|
|
|
base_url: str = ""
|
|
|
|
session: AcSession = None
|
|
|
|
def __init__(self, base_url: str):
|
|
self.base_url = base_url
|
|
|
|
def login_to_first_account(self, email: str, password: str) -> AcSession:
|
|
login_res = self.user_login(email, password)
|
|
cur_account = self.select_first_account(login_res.accounts)
|
|
token = self.create_token(cur_account, login_res.user)
|
|
self.session = AcSession(login_res.user, login_res.accounts, cur_account, token)
|
|
return self.session
|
|
|
|
def user_login(self, email: str, password: str) -> AcLoginResponse:
|
|
auth = ACAuthenticator(self.base_url)
|
|
res = auth.login(email, password)
|
|
if res.status_code != 200:
|
|
raise Exception('Login failed!')
|
|
res_data = res.json()
|
|
if res_data['is_ok'] != 1:
|
|
raise Exception('Login failed! (2)')
|
|
accounts = list(map(lambda a: account_from_json(a), res_data['accounts']))
|
|
return AcLoginResponse(AcUser(**res_data['user']), accounts)
|
|
|
|
def select_first_account(self, accounts: list[AcAccount]) -> AcAccount:
|
|
return accounts[0]
|
|
|
|
def create_token(self, account: AcAccount, user: AcUser) -> AcToken:
|
|
authenticator = ACTokenAuthenticator(account.url + '/api/v%s' % AC_API_VERSION)
|
|
res = authenticator.issue_token_intent(user.intent)
|
|
if res.status_code != 200:
|
|
raise Exception('Request token failed!')
|
|
res_data = res.json()
|
|
if res_data['is_ok'] != 1:
|
|
raise Exception('Request token failed! (2)')
|
|
return AcToken(res_data["token"])
|
|
|
|
def get_info(self):
|
|
client = ACClient(self.session.cur_account, self.session.token)
|
|
res = client.get_info()
|
|
return res.json()
|
|
|
|
def get_projects(self):
|
|
client = ACClient(self.session.cur_account, self.session.token)
|
|
res = client.get_projects()
|
|
if res.status_code != 200:
|
|
raise Exception("Error %d" % res.status_code)
|
|
res_data = res.json()
|
|
projects = list(map(lambda p: project_from_json(p), res_data))
|
|
return projects
|