github-marketing-tools.py

import csv
import math

import gidgethub.httpx
import httpx
import trio
import asyncclick as click

STARGAZAERS = """query StargazersCount($owner: String!, $name: String!, $after: String = null)
{
repository(owner: $owner, name: $name) {
    stargazers(first: 100, after: $after) {
      edges {
        node {
          email
        }
      }
      pageInfo {
        endCursor
        hasNextPage
      }
    }
  }
}"""


async def write_csv(receiver):
    async with receiver:
        seen_emails = {}

        with open('/home/thedrow/Documents/Projects/github-marketing-tools/emails.csv', 'w+', newline='') as csvfile:
            writer = csv.writer(csvfile)
            async for response in receiver:
                for node in response["repository"]["stargazers"]["edges"]:
                    email = node["node"]["email"].strip()

                    if len(email) == 0 or email in seen_emails:
                        continue

                    seen_emails.add(email
                    writer.writerow([email])
            csvfile.flush()


async def fetch_stargazers(nursery, sender, gh, owner, name, after=None):
    with sender:
        async with httpx.AsyncClient() as client:
            gh = gidgethub.httpx.GitHubAPI(client, "github-marketing-tools",
                                           oauth_token=token)

            while True:
                response = await gh.graphql(STARGAZAERS, owner=owner, name=name, after=after)

                sender.send(response)

                page_info = response["repository"]["stargazers"]["pageInfo"]

                if not page_info["hasNextPage"]:
                    break
                
                after = page_info["endCursor"]


@click.command()
@click.option("--token", default=None, help="Github API token.")
@click.argument("repository")
async def stargazers(repository, token=None):
    owner, name = repository.split('/')
    sender, receiver = trio.open_memory_channel(math.inf)

    async with trio.open_nursery() as nursery:
        nursery.start_soon(fetch_stargazers, sender, gh, owner, name)
        nursery.start_soon(write_csv, receiver)


if __name__ == '__main__':
    stargazers()