RELEASING/changelog.py - superset - Git at Google

 #
 # Licensed to the Apache Software Foundation (ASF) under one or more
 # contributor license agreements.  See the NOTICE file distributed with
 # this work for additional information regarding copyright ownership.
 # The ASF licenses this file to You under the Apache License, Version 2.0
 # (the "License"); you may not use this file except in compliance with
 # the License.  You may obtain a copy of the License at
 #
 #    http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
 # pylint: disable=no-value-for-parameter

 import csv as lib_csv
 import os
 import re
 import sys
 from dataclasses import dataclass
 from typing import Any, Dict, Iterator, List, Optional, Union

 import click

 try:
     from github import BadCredentialsException, Github, PullRequest
 except ModuleNotFoundError:
     print("PyGithub is a required package for this script")
     exit(1)

 SUPERSET_REPO = "apache/superset"


 @dataclass
 class GitLog:
     """
     Represents a git log entry
     """

     sha: str
     author: str
     time: str
     message: str
     pr_number: Union[int, None] = None
     author_email: str = ""

     def __eq__(self, other: object) -> bool:
         """ A log entry is considered equal if it has the same PR number """
         if isinstance(other, self.__class__):
             return other.pr_number == self.pr_number
         return False

     def __repr__(self) -> str:
         return f"[{self.pr_number}]: {self.message} {self.time} {self.author}"


 class GitChangeLog:
     """
     Helper class to output a list of logs entries on a superset changelog format

     We want to map a git author to a github login, for that we call github's API
     """

     def __init__(
         self, version: str, logs: List[GitLog], access_token: Optional[str] = None
     ) -> None:
         self._version = version
         self._logs = logs
         self._github_login_cache: Dict[str, Optional[str]] = {}
         self._wait = 10
         github_token = access_token or os.environ.get("GITHUB_TOKEN")
         self._github = Github(github_token)
         self._superset_repo = ""

     def _fetch_github_pr(self, pr_number: int) -> PullRequest:
         """
         Fetches a github PR info
         """
         try:
             github_repo = self._github.get_repo(SUPERSET_REPO)
         except BadCredentialsException as ex:
             print(
                 f"Bad credentials to github provided"
                 f" use access_token parameter or set GITHUB_TOKEN"
             )
             sys.exit(1)

         return github_repo.get_pull(pr_number)

     def _get_github_login(self, git_log: GitLog) -> Optional[str]:
         """
         Tries to fetch a github login (username) from a git author
         """
         author_name = git_log.author
         github_login = self._github_login_cache.get(author_name)
         if github_login:
             return github_login
         if git_log.pr_number:
             pr_info = self._fetch_github_pr(git_log.pr_number)
             if pr_info:
                 github_login = pr_info.user.login
             else:
                 github_login = author_name
         # set cache
         self._github_login_cache[author_name] = github_login
         return github_login

     def _get_changelog_version_head(self) -> str:
         return f"### {self._version} ({self._logs[0].time})"

     def __repr__(self) -> str:
         result = f"\n{self._get_changelog_version_head()}\n"
         for i, log in enumerate(self._logs):
             github_login = self._get_github_login(log)
             if not github_login:
                 github_login = log.author
             result = result + (
                 f"- [#{log.pr_number}]"
                 f"(https://github.com/{SUPERSET_REPO}/pull/{log.pr_number}) "
                 f"{log.message} (@{github_login})\n"
             )
             print(f"\r {i}/{len(self._logs)}", end="", flush=True)
         return result

     def __iter__(self) -> Iterator[Dict[str, Any]]:
         for log in self._logs:
             yield {
                 "pr_number": log.pr_number,
                 "pr_link": f"https://github.com/{SUPERSET_REPO}/pull/"
                 f"{log.pr_number}",
                 "message": log.message,
                 "time": log.time,
                 "author": log.author,
                 "email": log.author_email,
                 "sha": log.sha,
             }


 class GitLogs:
     """
     Manages git log entries from a specific branch/tag

     Can compare git log entries by PR number
     """

     def __init__(self, git_ref: str) -> None:
         self._git_ref = git_ref
         self._logs: List[GitLog] = []

     @property
     def git_ref(self) -> str:
         return self._git_ref

     @property
     def logs(self) -> List[GitLog]:
         return self._logs

     def fetch(self) -> None:
         self._logs = list(map(self._parse_log, self._git_logs()))[::-1]

     def diff(self, git_logs: "GitLogs") -> List[GitLog]:
         return [log for log in git_logs.logs if log not in self._logs]

     def __repr__(self) -> str:
         return f"{self._git_ref}, Log count:{len(self._logs)}"

     @staticmethod
     def _git_get_current_head() -> str:
         output = os.popen("git status | head -1").read()
         match = re.match("(?:HEAD detached at|On branch) (.*)", output)
         if not match:
             return ""
         return match.group(1)

     def _git_checkout(self, git_ref: str) -> None:
         os.popen(f"git checkout {git_ref}").read()
         current_head = self._git_get_current_head()
         if current_head != git_ref:
             print(f"Could not checkout {git_ref}")
             sys.exit(1)

     def _git_logs(self) -> List[str]:
         # let's get current git ref so we can revert it back
         current_git_ref = self._git_get_current_head()
         self._git_checkout(self._git_ref)
         output = (
             os.popen('git --no-pager log --pretty=format:"%h|%an|%ae|%ad|%s|"')
             .read()
             .split("\n")
         )
         # revert to git ref, let's be nice
         self._git_checkout(current_git_ref)
         return output

     @staticmethod
     def _parse_log(log_item: str) -> GitLog:
         pr_number = None
         split_log_item = log_item.split("|")
         # parse the PR number from the log message
         match = re.match(r".*\(\#(\d*)\)", split_log_item[4])
         if match:
             pr_number = int(match.group(1))
         return GitLog(
             sha=split_log_item[0],
             author=split_log_item[1],
             author_email=split_log_item[2],
             time=split_log_item[3],
             message=split_log_item[4],
             pr_number=pr_number,
         )


 @dataclass
 class BaseParameters:
     previous_logs: GitLogs
     current_logs: GitLogs


 def print_title(message: str) -> None:
     print(f"{50*'-'}")
     print(message)
     print(f"{50*'-'}")


 @click.group()
 @click.pass_context
 @click.option("--previous_version", help="The previous release version", required=True)
 @click.option("--current_version", help="The current release version", required=True)
 def cli(ctx, previous_version: str, current_version: str) -> None:
     """ Welcome to change log generator  """
     previous_logs = GitLogs(previous_version)
     current_logs = GitLogs(current_version)
     previous_logs.fetch()
     current_logs.fetch()
     base_parameters = BaseParameters(previous_logs, current_logs)
     ctx.obj = base_parameters


 @cli.command("compare")
 @click.pass_obj
 def compare(base_parameters: BaseParameters) -> None:
     """ Compares both versions (by PR) """
     previous_logs = base_parameters.previous_logs
     current_logs = base_parameters.current_logs
     print_title(
         f"Pull requests from " f"{current_logs.git_ref} not in {previous_logs.git_ref}"
     )
     previous_diff_logs = previous_logs.diff(current_logs)
     for diff_log in previous_diff_logs:
         print(f"{diff_log}")

     print_title(
         f"Pull requests from " f"{previous_logs.git_ref} not in {current_logs.git_ref}"
     )
     current_diff_logs = current_logs.diff(previous_logs)
     for diff_log in current_diff_logs:
         print(f"{diff_log}")


 @cli.command("changelog")
 @click.option(
     "--csv", help="The csv filename to export the changelog to",
 )
 @click.option(
     "--access_token",
     help="The github access token,"
     " if not provided will try to fetch from GITHUB_TOKEN env var",
 )
 @click.pass_obj
 def change_log(base_parameters: BaseParameters, csv: str, access_token: str) -> None:
     """ Outputs a changelog (by PR) """
     previous_logs = base_parameters.previous_logs
     current_logs = base_parameters.current_logs
     previous_diff_logs = previous_logs.diff(current_logs)
     logs = GitChangeLog(
         current_logs.git_ref, previous_diff_logs[::-1], access_token=access_token
     )
     if csv:
         with open(csv, "w") as csv_file:
             log_items = list(logs)
             field_names = log_items[0].keys()
             writer = lib_csv.DictWriter(
                 csv_file,
                 delimiter=",",
                 quotechar='"',
                 quoting=lib_csv.QUOTE_ALL,
                 fieldnames=field_names,
             )
             writer.writeheader()
             for log in logs:
                 writer.writerow(log)
     else:
         print("Fetching github usernames, this may take a while:")
         print(logs)


 cli()
	#
	# Licensed to the Apache Software Foundation (ASF) under one or more
	# contributor license agreements. See the NOTICE file distributed with
	# this work for additional information regarding copyright ownership.
	# The ASF licenses this file to You under the Apache License, Version 2.0
	# (the "License"); you may not use this file except in compliance with
	# the License. You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing, software
	# distributed under the License is distributed on an "AS IS" BASIS,
	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	# See the License for the specific language governing permissions and
	# limitations under the License.
	#
	# pylint: disable=no-value-for-parameter

	import csv as lib_csv
	import os
	import re
	import sys
	from dataclasses import dataclass
	from typing import Any, Dict, Iterator, List, Optional, Union

	import click

	try:
	from github import BadCredentialsException, Github, PullRequest
	except ModuleNotFoundError:
	print("PyGithub is a required package for this script")
	exit(1)

	SUPERSET_REPO = "apache/superset"


	@dataclass
	class GitLog:
	"""
	Represents a git log entry
	"""

	sha: str
	author: str
	time: str
	message: str
	pr_number: Union[int, None] = None
	author_email: str = ""

	def __eq__(self, other: object) -> bool:
	""" A log entry is considered equal if it has the same PR number """
	if isinstance(other, self.__class__):
	return other.pr_number == self.pr_number
	return False

	def __repr__(self) -> str:
	return f"[{self.pr_number}]: {self.message} {self.time} {self.author}"


	class GitChangeLog:
	"""
	Helper class to output a list of logs entries on a superset changelog format

	We want to map a git author to a github login, for that we call github's API
	"""

	def __init__(
	self, version: str, logs: List[GitLog], access_token: Optional[str] = None
	) -> None:
	self._version = version
	self._logs = logs
	self._github_login_cache: Dict[str, Optional[str]] = {}
	self._wait = 10
	github_token = access_token or os.environ.get("GITHUB_TOKEN")
	self._github = Github(github_token)
	self._superset_repo = ""

	def _fetch_github_pr(self, pr_number: int) -> PullRequest:
	"""
	Fetches a github PR info
	"""
	try:
	github_repo = self._github.get_repo(SUPERSET_REPO)
	except BadCredentialsException as ex:
	print(
	f"Bad credentials to github provided"
	f" use access_token parameter or set GITHUB_TOKEN"
	)
	sys.exit(1)

	return github_repo.get_pull(pr_number)

	def _get_github_login(self, git_log: GitLog) -> Optional[str]:
	"""
	Tries to fetch a github login (username) from a git author
	"""
	author_name = git_log.author
	github_login = self._github_login_cache.get(author_name)
	if github_login:
	return github_login
	if git_log.pr_number:
	pr_info = self._fetch_github_pr(git_log.pr_number)
	if pr_info:
	github_login = pr_info.user.login
	else:
	github_login = author_name
	# set cache
	self._github_login_cache[author_name] = github_login
	return github_login

	def _get_changelog_version_head(self) -> str:
	return f"### {self._version} ({self._logs[0].time})"

	def __repr__(self) -> str:
	result = f"\n{self._get_changelog_version_head()}\n"
	for i, log in enumerate(self._logs):
	github_login = self._get_github_login(log)
	if not github_login:
	github_login = log.author
	result = result + (
	f"- [#{log.pr_number}]"
	f"(https://github.com/{SUPERSET_REPO}/pull/{log.pr_number}) "
	f"{log.message} (@{github_login})\n"
	)
	print(f"\r {i}/{len(self._logs)}", end="", flush=True)
	return result

	def __iter__(self) -> Iterator[Dict[str, Any]]:
	for log in self._logs:
	yield {
	"pr_number": log.pr_number,
	"pr_link": f"https://github.com/{SUPERSET_REPO}/pull/"
	f"{log.pr_number}",
	"message": log.message,
	"time": log.time,
	"author": log.author,
	"email": log.author_email,
	"sha": log.sha,
	}


	class GitLogs:
	"""
	Manages git log entries from a specific branch/tag

	Can compare git log entries by PR number
	"""

	def __init__(self, git_ref: str) -> None:
	self._git_ref = git_ref
	self._logs: List[GitLog] = []

	@property
	def git_ref(self) -> str:
	return self._git_ref

	@property
	def logs(self) -> List[GitLog]:
	return self._logs

	def fetch(self) -> None:
	self._logs = list(map(self._parse_log, self._git_logs()))[::-1]

	def diff(self, git_logs: "GitLogs") -> List[GitLog]:
	return [log for log in git_logs.logs if log not in self._logs]

	def __repr__(self) -> str:
	return f"{self._git_ref}, Log count:{len(self._logs)}"

	@staticmethod
	def _git_get_current_head() -> str:
	output = os.popen("git status \| head -1").read()
	match = re.match("(?:HEAD detached at\|On branch) (.*)", output)
	if not match:
	return ""
	return match.group(1)

	def _git_checkout(self, git_ref: str) -> None:
	os.popen(f"git checkout {git_ref}").read()
	current_head = self._git_get_current_head()
	if current_head != git_ref:
	print(f"Could not checkout {git_ref}")
	sys.exit(1)

	def _git_logs(self) -> List[str]:
	# let's get current git ref so we can revert it back
	current_git_ref = self._git_get_current_head()
	self._git_checkout(self._git_ref)
	output = (
	os.popen('git --no-pager log --pretty=format:"%h\|%an\|%ae\|%ad\|%s\|"')
	.read()
	.split("\n")
	)
	# revert to git ref, let's be nice
	self._git_checkout(current_git_ref)
	return output

	@staticmethod
	def _parse_log(log_item: str) -> GitLog:
	pr_number = None
	split_log_item = log_item.split("\|")
	# parse the PR number from the log message
	match = re.match(r".\(\#(\d)\)", split_log_item[4])
	if match:
	pr_number = int(match.group(1))
	return GitLog(
	sha=split_log_item[0],
	author=split_log_item[1],
	author_email=split_log_item[2],
	time=split_log_item[3],
	message=split_log_item[4],
	pr_number=pr_number,
	)


	@dataclass
	class BaseParameters:
	previous_logs: GitLogs
	current_logs: GitLogs


	def print_title(message: str) -> None:
	print(f"{50*'-'}")
	print(message)
	print(f"{50*'-'}")


	@click.group()
	@click.pass_context
	@click.option("--previous_version", help="The previous release version", required=True)
	@click.option("--current_version", help="The current release version", required=True)
	def cli(ctx, previous_version: str, current_version: str) -> None:
	""" Welcome to change log generator """
	previous_logs = GitLogs(previous_version)
	current_logs = GitLogs(current_version)
	previous_logs.fetch()
	current_logs.fetch()
	base_parameters = BaseParameters(previous_logs, current_logs)
	ctx.obj = base_parameters


	@cli.command("compare")
	@click.pass_obj
	def compare(base_parameters: BaseParameters) -> None:
	""" Compares both versions (by PR) """
	previous_logs = base_parameters.previous_logs
	current_logs = base_parameters.current_logs
	print_title(
	f"Pull requests from " f"{current_logs.git_ref} not in {previous_logs.git_ref}"
	)
	previous_diff_logs = previous_logs.diff(current_logs)
	for diff_log in previous_diff_logs:
	print(f"{diff_log}")

	print_title(
	f"Pull requests from " f"{previous_logs.git_ref} not in {current_logs.git_ref}"
	)
	current_diff_logs = current_logs.diff(previous_logs)
	for diff_log in current_diff_logs:
	print(f"{diff_log}")


	@cli.command("changelog")
	@click.option(
	"--csv", help="The csv filename to export the changelog to",
	)
	@click.option(
	"--access_token",
	help="The github access token,"
	" if not provided will try to fetch from GITHUB_TOKEN env var",
	)
	@click.pass_obj
	def change_log(base_parameters: BaseParameters, csv: str, access_token: str) -> None:
	""" Outputs a changelog (by PR) """
	previous_logs = base_parameters.previous_logs
	current_logs = base_parameters.current_logs
	previous_diff_logs = previous_logs.diff(current_logs)
	logs = GitChangeLog(
	current_logs.git_ref, previous_diff_logs[::-1], access_token=access_token
	)
	if csv:
	with open(csv, "w") as csv_file:
	log_items = list(logs)
	field_names = log_items[0].keys()
	writer = lib_csv.DictWriter(
	csv_file,
	delimiter=",",
	quotechar='"',
	quoting=lib_csv.QUOTE_ALL,
	fieldnames=field_names,
	)
	writer.writeheader()
	for log in logs:
	writer.writerow(log)
	else:
	print("Fetching github usernames, this may take a while:")
	print(logs)


	cli()