import time from typing import Any import httpx import requests # from django.conf import settings # from django.core.cache import cache from django.db import models from httpx import Request from gotify_rest_api_client import AuthenticatedClient from gotify_rest_api_client.models import * from gotify_rest_api_client.api.message import create_message from gotify_rest_api_client.api.user import create_user from gotify_rest_api_client.api.client import create_client from gotify_rest_api_client.api.application import create_app ADMIN_CLIENT_TOKEN = 'CbWdsJrmRTDzbuD' import logging # logging.basicConfig( # format="%(levelname)s [%(asctime)s] %(name)s - %(message)s", # datefmt="%Y-%m-%d %H:%M:%S", # level=logging.DEBUG # ) logger = logging.getLogger('utils.gotify_clients') def log_request(request: Request): logger.debug(f'authorization: {request.headers["authorization"]}') logger.debug( f"Request event hook: {request.method} {request.url} {request.method} {request.headers} - Waiting for response") def log_response(response): request = response.request logger.debug(f"Response event hook: {request.method} {request.url} - Status {response.status_code}") # logger.debug(f"Response event hook: {response.read()}") def get_client(access_token): client = AuthenticatedClient(base_url="https://push.gooyal.com", token=access_token, httpx_args={"event_hooks": {"request": [log_request], "response": [log_response]}}, ) return client def get_basic_client(username, password): auth = httpx.BasicAuth(username=username, password=password) prefix, token = auth._auth_header.split() # httpx_client = httpx.Client(auth=auth, base_url="https://push.gooyal.com") client = AuthenticatedClient(base_url="https://push.gooyal.com", token=token, prefix=prefix, httpx_args={ "event_hooks": {"request": [log_request], "response": [log_response]}, # "auth": auth, # "base_url": "https://push.gooyal.com" } ) # client.set_httpx_client(httpx_client) return client def send_notif(data): class Tmp(): def to_dict(self): return data final_data = Tmp() client = get_client() response: Any = create_message.sync(client=client, body=final_data) return response def submit_user(user_uuid): data_dict = { "admin": False, "name": user_uuid, "pass": user_uuid } class Tmp(): def to_dict(self): return data_dict final_data = Tmp() client = get_client(ADMIN_CLIENT_TOKEN) response: UserExternalModel = create_user.sync(client=client, body=final_data) return response def submit_client(user_uuid): data_dict = { "name": user_uuid, } class Tmp(): def to_dict(self): return data_dict final_data = Tmp() client = get_basic_client(user_uuid, user_uuid) response: ClientModel = create_client.sync(client=client, body=final_data) return response def submit_application(token, user_uuid): final_data = ApplicationParamsModel( default_priority=5, name=user_uuid ) client = get_client(token) response: ApplicationModel = create_app.sync(client=client, body=final_data) return response sample_data = { "message": "**Backup** was successfully finished.", "priority": 2, "title": "Backup" } # ClientModel(id=20, name='820c1110-a3bd-47ce-a3f4-f8d2dd70fa9f', token='CbNecn03yWNIl93', last_used=None, additional_properties={}) # send_notif(sample_data) # print(submit_user('820c1110-a3bd-47ce-a3f4-f8d2dd70fa9f')) # print(submit_client('820c1110-a3bd-47ce-a3f4-f8d2dd70fa9f')) # print(submit_application('CbNecn03yWNIl93', '820c1110-a3bd-47ce-a3f4-f8d2dd70fa9f'))