93 lines
3.4 KiB
Python
93 lines
3.4 KiB
Python
from time import time
|
|
from typing import Optional, Any, Dict
|
|
|
|
from oauthlib.oauth2 import BackendApplicationClient
|
|
from requests_oauthlib import OAuth2Session
|
|
|
|
from .models import PongModel, PostModel
|
|
|
|
oauth_url: str = 'https://sso.arbina.com/oauth'
|
|
vue_client_url: str = 'https://vue-client-api.arbina.com'
|
|
vue_interact_url: str = 'https://vue-app-interaction-api.arbina.com'
|
|
|
|
|
|
class ClientAPI:
|
|
client_id: str
|
|
client_secret: str
|
|
token: Optional[Any]
|
|
session: OAuth2Session
|
|
endpoints: Dict[str, str] = {'create': '/api/pages/{page_id}/posts', 'remove': '/api/posts/{post_id}',
|
|
'ping': f'{vue_interact_url}/api/app/interactions/ping', 'token': f'{oauth_url}/token'}
|
|
|
|
def build_url(self, endpoint: str) -> str:
|
|
if self.endpoints[endpoint].startswith('/'):
|
|
return f'{self.vue_apps_url}{self.endpoints[endpoint]}'
|
|
else:
|
|
return self.endpoints[endpoint]
|
|
|
|
def __init__(self, client_id: str, client_secret: str, vue_url: str = vue_client_url):
|
|
"""
|
|
Constructor of Vue Apps Client API
|
|
|
|
Use this object to make requests to our API with ease
|
|
:param client_id: provided to you in the process of App Creation at Vue Developer Portal
|
|
:param client_secret: non-shareable secret from Developer Portal that provides you with access to our API
|
|
:param vue_url: custom URL of Vue Apps API to use
|
|
"""
|
|
self.vue_apps_url = vue_url
|
|
|
|
self.client_id = client_id
|
|
self.client_secret = client_secret
|
|
|
|
oauth = OAuth2Session(client=BackendApplicationClient(client_id=self.client_id))
|
|
client_token = oauth.fetch_token(token_url=self.build_url('token'), client_id=self.client_id,
|
|
client_secret=self.client_secret)
|
|
|
|
self.token = client_token
|
|
self.setup_flow()
|
|
|
|
def setup_flow(self):
|
|
"""
|
|
Method sets up the client OAuth2 Session, which will be refreshing tokens for you automatically on requests
|
|
"""
|
|
|
|
def save_token(token):
|
|
if token:
|
|
self.token = token
|
|
|
|
auto_refresh_info = {'client_id': self.client_id, 'client_secret': self.client_secret}
|
|
self.session = OAuth2Session(token=self.token, client_id=self.client_id,
|
|
auto_refresh_url=self.build_url('token'),
|
|
auto_refresh_kwargs=auto_refresh_info, token_updater=save_token)
|
|
|
|
def acquire_token(self) -> Optional[Any]:
|
|
"""
|
|
Get available Access Token to send requests with OAuth2
|
|
:return: Access Token object
|
|
"""
|
|
return self.session.access_token
|
|
|
|
def ping_server(self) -> Optional[PongModel]:
|
|
query_url = self.build_url('ping')
|
|
response = self.session.get(query_url)
|
|
|
|
if response.status_code != 200:
|
|
return None
|
|
|
|
model = response.json()
|
|
return PongModel(ping_ttl_seconds=model['ping_ttl_seconds'])
|
|
|
|
def refresh_token(self) -> bool:
|
|
"""
|
|
Forces token to be refreshed and sends requests to /ping
|
|
:return: True when /ping sent HTTP 200 OK else False
|
|
"""
|
|
self.token['expires_at'] = time() - 10
|
|
self.session.token = self.token
|
|
pong = self.ping_server()
|
|
return pong is not None
|
|
|
|
def create_post(self, post: PostModel) -> bool:
|
|
url = self.build_url('/create').format(page_id=post.page_uin)
|
|
return True
|