import json import requests OAUTH_CLIENT_ID = '5yjU0g18LOIl1fzNgRQ9xGU0Ou4UTsimoMvvRZsG' OAUTH_CLIENT_SECRET = 'vyNixGW5OhOnvn9s1Z21RWdsEtsau1a45e15FAhFV3WfWy4lPdcTu6dmhR5MRKqpcKeScN7WvMGomZhzh7vJgcGWEbs7hptYriHI5ZIw3672hm4D3zUfcWYMjMUMuUOO' API_URI = 'http://127.0.0.1:8000' class ApiClient(): def __init__(self, phone_number, auth_data=None): self.auth_data = {} if auth_data: self.auth_data = auth_data self.phone_number = phone_number def login(self, password): phone_number = self.phone_number data = { "grant_type": "password", "username": phone_number, "password": password, "scope": 'read write groups introspection', "auth_gateway": 'phone_number:otp' } auth = (OAUTH_CLIENT_ID, OAUTH_CLIENT_SECRET) response = requests.post(f'{API_URI}/oauth2/token/', data=data, auth=auth) auth_data = response.json() if 'access_token' in auth_data: self.auth_data = auth_data return auth_data def _request(self, path, data=None, files=None, method='get', with_auth=True, encode=True): if files: header = {} else: header = { "Content-Type": "application/json", "charset": "utf-8" } if with_auth: access_token = self.auth_data.get('access_token', '') header.setdefault("Authorization", f"Bearer {access_token}", ) if data and encode: data = json.dumps(data) response = requests.request(method, f'{API_URI}/{path}', headers=header, data=data, files=files) try: return response.json() except: return response.content.decode() def request_otp(self): phone_number = self.phone_number path = 'request_otp/' method = 'post' data = { 'phone_number': phone_number } result = self._request(path=path, data=data, method=method, with_auth=False) return result def get_account(self): path = 'account' result = self._request(path=path) return result def update_account(self, **kwargs): path = 'account/' data = {} fields = ( 'username', "first_name", "last_name", 'name', 'iban', 'iban_card_number', 'iban_account_number', ) for key, value in kwargs.items(): if key in fields: data.setdefault(key, value) avatar = kwargs.get('avatar') if avatar: files = {'avatar': avatar} encode = False else: files = None encode = True result = self._request(path=path, data=data, method='put', files=files, encode=encode) return result def change_password(self, old_password, new_password, old_password_gateway='otp'): path = 'change_password/' data = {'old_password': old_password, 'new_password': new_password, 'old_password_gateway': old_password_gateway } result = self._request(path=path, data=data, method='put') return result def get_user_profile(self, code): path = f'users/{code}/' result = self._request(path=path) return result def get_transactions(self, page=None): path = 'transactions/' if page: page = int(page) path = f'{path}?page={page}' return self._request(path=path) def get_transaction(self, code): path = f'transactions/{code}' return self._request(path=path) def create_transaction(self, amount, delay, payer=None): data = { 'delay': delay, 'amount': amount } if payer: data.setdefault('payer', {'code': payer}) response = self._request('transactions/', method='post', data=data) return response and 'code' in response, response def pay_transaction(self, code): path = f'transactions/{code}/pay' return self._request(path=path) def receipt_transaction(self, code): path = f'transactions/{code}/receipt' return self._request(path=path) def get_introspection(self): path = 'oauth2/introspect/' access_token = self.auth_data.get('access_token', '') data = { 'token': access_token, } return self._request(path=path, data=data)