107 lines
3.2 KiB
Python
107 lines
3.2 KiB
Python
import time
|
|
|
|
import requests
|
|
from django.conf import settings
|
|
from django.core.cache import cache
|
|
from httpx import Request
|
|
|
|
from utils.clients.gooyal_advertising_client import AuthenticatedClient
|
|
from utils.clients.gooyal_advertising_client.api.crm.crm_tickets_list import sync as crm_tickets_list_sync
|
|
from utils.clients.gooyal_advertising_client.api.crm.crm_application_tickets_list import sync as crm_application_tickets_list_sync
|
|
from utils.clients.gooyal_advertising_client.api.suggestions.suggestions_suggest_list import sync as suggestions_suggest_list_sync
|
|
from utils.clients.gooyal_advertising_client.models.paginated_ticket_list import PaginatedTicketList
|
|
from utils.clients.gooyal_advertising_client.models.paginated_suggestion_list import PaginatedSuggestionList
|
|
from utils.http_fallback import request_with_ssl_fallback, resolve_verify_ssl
|
|
|
|
|
|
def login_as_client_credentials():
|
|
access_token = cache.get('advertising_access_token')
|
|
if access_token:
|
|
print('already in cache')
|
|
return access_token
|
|
|
|
print('not in cache')
|
|
|
|
data = {
|
|
'grant_type': 'client_credentials',
|
|
'scope': settings.OAUTH2_PROVIDER_SCOPES,
|
|
}
|
|
auth = (settings.OAUTH2_PROVIDER_CLIENT_ID, settings.OAUTH2_PROVIDER_CLIENT_SECRET)
|
|
|
|
response = request_with_ssl_fallback(
|
|
'POST',
|
|
f'{settings.OAUTH2_PROVIDER_BASE_PUBLIC_URL}/token/',
|
|
data=data,
|
|
auth=auth,
|
|
)
|
|
print(response.content)
|
|
|
|
auth_data = response.json()
|
|
if 'access_token' in auth_data:
|
|
auth_data['created_at'] = time.time()
|
|
cache.set('advertising_access_token', auth_data, timeout=auth_data.get('expires_in') - 5)
|
|
|
|
return auth_data
|
|
|
|
|
|
def log_request(request: Request):
|
|
print(request.headers.get('authorization'))
|
|
print(
|
|
f'Request event hook: {request.method} {request.url} {request.headers} - Waiting for response'
|
|
)
|
|
|
|
|
|
def log_response(response):
|
|
request = response.request
|
|
print(f'Response event hook: {request.method} {request.url} - Status {response.status_code}')
|
|
|
|
|
|
def get_client():
|
|
access_token = login_as_client_credentials()['access_token']
|
|
verify_ssl = resolve_verify_ssl(settings.ADVERTISING_BASE_PUBLIC_URL)
|
|
client = AuthenticatedClient(
|
|
base_url=settings.ADVERTISING_BASE_PUBLIC_URL,
|
|
token=access_token,
|
|
verify_ssl=verify_ssl,
|
|
httpx_args={
|
|
'event_hooks': {
|
|
'request': [log_request],
|
|
'response': [log_response],
|
|
}
|
|
},
|
|
)
|
|
return client
|
|
|
|
|
|
def get_suggestions(limit=None, offset=None):
|
|
client = get_client()
|
|
response: PaginatedSuggestionList = suggestions_suggest_list_sync(
|
|
client=client,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
print(response)
|
|
return response
|
|
|
|
|
|
def get_crm_tickets(limit=None, offset=None):
|
|
client = get_client()
|
|
response: PaginatedTicketList = crm_tickets_list_sync(
|
|
client=client,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
print(response)
|
|
return response
|
|
|
|
|
|
def get_crm_application_tickets(limit=None, offset=None):
|
|
client = get_client()
|
|
response: PaginatedTicketList = crm_application_tickets_list_sync(
|
|
client=client,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
print(response)
|
|
return response
|
|
|