from django.conf import settings from django.core.cache import cache from gql import Client, gql from gql.transport.requests import RequestsHTTPTransport from social_django.models import UserSocialAuth from .models import User GQL_ENDPOINT = 'https://api.github.com/graphql' CACHE_KEY = 'cached-sponsors:%s' def check_is_sponsor(user: User) -> bool: try: token = user.social_auth.get(provider='github').extra_data.get('access_token') except UserSocialAuth.DoesNotExist: return False if not token: return False client = Client( transport=RequestsHTTPTransport( url=GQL_ENDPOINT, use_json=True, headers={ 'Authorization': f'Bearer {token}', } ) ) after = None while True: params = 'first: 1' if after: params += f', after:"{after}"' query = ''' query { viewer { sponsorshipsAsSponsor(%s) { pageInfo { startCursor hasNextPage endCursor } totalRecurringMonthlyPriceInDollars nodes { sponsorable { ... on Organization { login } ... on User { login } } } } } } ''' % (params,) response = client.execute(gql(query)) info = response['viewer']['sponsorshipsAsSponsor'] after = info['pageInfo']['endCursor'] nodes = info['nodes'] if not len(nodes): break for node in nodes: if node['sponsorable']['login'].lower() not in settings.GITHUB_ELIGIBLE_SPONSORSHIPS: continue if info['totalRecurringMonthlyPriceInDollars'] >= settings.GITHUB_SPONSORS_MIN_PAYMENT: return True return False def check_is_sponsor_cached(user: User) -> bool: cache_key = CACHE_KEY % user.id if not cache.get(cache_key): cache.set(cache_key, check_is_sponsor(user), timeout=30) return cache.get(cache_key)