examples automation examples-dir for tsp

tools/azure-rest-api-specs-examples-automation/automation/csv_database.py

@@ -11,9 +11,9 @@
 from models import Release
 
 
-example_repo: str = 'https://github.com/Azure/azure-rest-api-specs-examples'
-csvdb_folder: str = 'csvdb'
-metadata_branch: str = 'metadata'
+example_repo: str = "https://github.com/Azure/azure-rest-api-specs-examples"
+csvdb_folder: str = "csvdb"
+metadata_branch: str = "metadata"
 
 
 @dataclasses.dataclass(eq=True)
@@ -55,101 +55,105 @@ def __init__(self, work_dir: str):
         self.work_dir = work_dir
         self.example_metadata_path = path.join(self.work_dir, csvdb_folder)
 
-        self.index_file_path = path.join(self.example_metadata_path, 'java-library-example-index.csv')
-        self.list_file_path = path.join(self.example_metadata_path, 'java-library-example-list.csv')
+        self.index_file_path = path.join(self.example_metadata_path, "java-library-example-index.csv")
+        self.list_file_path = path.join(self.example_metadata_path, "java-library-example-list.csv")
 
     def checkout(self):
         # checkout metadata branch from azure-rest-api-specs-examples repo
-        cmd = ['git', 'clone',
-               '--quiet',
-               '--depth', '1',
-               '--branch', metadata_branch,
-               example_repo, self.example_metadata_path]
-        logging.info(f'Checking out repository: {example_repo}, branch {metadata_branch}')
-        logging.info('Command line: ' + ' '.join(cmd))
+        cmd = [
+            "git",
+            "clone",
+            "--quiet",
+            "--depth",
+            "1",
+            "--branch",
+            metadata_branch,
+            example_repo,
+            self.example_metadata_path,
+        ]
+        logging.info(f"Checking out repository: {example_repo}, branch {metadata_branch}")
+        logging.info("Command line: " + " ".join(cmd))
         subprocess.check_call(cmd, cwd=self.work_dir)
 
     def load(self):
-        with open(self.index_file_path, 'r', newline='') as csv_file:
-            csv_reader = csv.reader(csv_file, delimiter=',', quotechar='"', quoting=csv.QUOTE_MINIMAL)
+        with open(self.index_file_path, "r", newline="") as csv_file:
+            csv_reader = csv.reader(csv_file, delimiter=",", quotechar='"', quoting=csv.QUOTE_MINIMAL)
             self.release_db = DatabaseInternal(csv_reader)
 
-        with open(self.list_file_path, 'r', newline='') as csv_file:
-            csv_reader = csv.reader(csv_file, delimiter=',', quotechar='"', quoting=csv.QUOTE_MINIMAL)
+        with open(self.list_file_path, "r", newline="") as csv_file:
+            csv_reader = csv.reader(csv_file, delimiter=",", quotechar='"', quoting=csv.QUOTE_MINIMAL)
             self.file_db = DatabaseInternal(csv_reader)
 
     def dump(self):
-        with open(self.index_file_path, 'w', newline='') as csv_file:
-            csv_writer = csv.writer(csv_file, delimiter=',', quotechar='"', quoting=csv.QUOTE_MINIMAL)
-            csv_writer.writerow(['id', 'name', 'language', 'tag', 'package', 'version', 'date_epoch', 'date'])
+        with open(self.index_file_path, "w", newline="") as csv_file:
+            csv_writer = csv.writer(csv_file, delimiter=",", quotechar='"', quoting=csv.QUOTE_MINIMAL)
+            csv_writer.writerow(["id", "name", "language", "tag", "package", "version", "date_epoch", "date"])
             for row in self.release_db.rows:
                 csv_writer.writerow(row)
 
-        with open(self.list_file_path, 'w', newline='') as csv_file:
-            csv_writer = csv.writer(csv_file, delimiter=',', quotechar='"', quoting=csv.QUOTE_MINIMAL)
-            csv_writer.writerow(['id', 'file', 'release_id'])
+        with open(self.list_file_path, "w", newline="") as csv_file:
+            csv_writer = csv.writer(csv_file, delimiter=",", quotechar='"', quoting=csv.QUOTE_MINIMAL)
+            csv_writer.writerow(["id", "file", "release_id"])
             for row in self.file_db.rows:
                 csv_writer.writerow(row)
 
     def commit(self, tag):
         if not self.branch:
             # git checkout new branch
-            self.date_str = datetime.now().strftime('%Y-%m-%d')
-            self.branch = f'automation-metadata-{self.date_str}'
-            cmd = ['git', 'checkout', '-b', self.branch]
-            logging.info('Command line: ' + ' '.join(cmd))
+            self.date_str = datetime.now().strftime("%Y-%m-%d")
+            self.branch = f"automation-metadata-{self.date_str}"
+            cmd = ["git", "checkout", "-b", self.branch]
+            logging.info("Command line: " + " ".join(cmd))
             subprocess.check_call(cmd, cwd=self.example_metadata_path)
 
         # git add
-        cmd = ['git', 'add', 'java-library-example-index.csv']
-        logging.info('Command line: ' + ' '.join(cmd))
+        cmd = ["git", "add", "java-library-example-index.csv"]
+        logging.info("Command line: " + " ".join(cmd))
         subprocess.check_call(cmd, cwd=self.example_metadata_path)
 
-        cmd = ['git', 'add', 'java-library-example-list.csv']
-        logging.info('Command line: ' + ' '.join(cmd))
+        cmd = ["git", "add", "java-library-example-list.csv"]
+        logging.info("Command line: " + " ".join(cmd))
         subprocess.check_call(cmd, cwd=self.example_metadata_path)
 
         # git commit
-        title = f'[Automation] Update metadata on {tag}'
-        logging.info(f'git commit: {title}')
-        cmd = ['git',
-               '-c', 'user.name=azure-sdk',
-               '-c', 'user.email=azuresdk@microsoft.com',
-               'commit', '-m', title]
-        logging.info('Command line: ' + ' '.join(cmd))
+        title = f"[Automation] Update metadata on {tag}"
+        logging.info(f"git commit: {title}")
+        cmd = ["git", "-c", "user.name=azure-sdk", "-c", "user.email=azuresdk@microsoft.com", "commit", "-m", title]
+        logging.info("Command line: " + " ".join(cmd))
         subprocess.check_call(cmd, cwd=self.example_metadata_path)
 
     def push(self, github_token: str):
         if self.branch:
-            title = f'[Automation] Update metadata on {self.date_str}'
+            title = f"[Automation] Update metadata on {self.date_str}"
             # git push
-            remote_uri = 'https://' + github_token + '@' + example_repo[len('https://'):]
-            cmd = ['git', 'push', remote_uri, self.branch]
+            remote_uri = "https://" + github_token + "@" + example_repo[len("https://") :]
+            cmd = ["git", "push", remote_uri, self.branch]
             # do not print this as it contains token
             # logging.info('Command line: ' + ' '.join(cmd))
             subprocess.check_call(cmd, cwd=self.example_metadata_path)
 
             # create github pull request
             owner = _repository_owner(example_repo)
             name = _repository_name(example_repo)
-            head = f'{owner}:{self.branch}'
+            head = f"{owner}:{self.branch}"
             repo = GitHubRepository(owner, name, github_token)
             pull_number = repo.create_pull_request(title, head, metadata_branch)
-            repo.add_label(pull_number, ['auto-merge'])
-            logging.info(f'succeeded, pull number {pull_number}')
+            repo.add_label(pull_number, ["auto-merge"])
+            logging.info(f"succeeded, pull number {pull_number}")
 
-    def new_release(self, name: str, language: str, tag: str, package: str, version: str, date: datetime,
-                    files: List[str]) -> bool:
+    def new_release(
+        self, name: str, language: str, tag: str, package: str, version: str, date: datetime, files: List[str]
+    ) -> bool:
         # add a new release and all the example files
         # return false, if release already exists in DB
 
         release_id = self._query_release(name, language)
         if release_id:
-            logging.warning(f'Release already exists for {language}#{name}')
+            logging.warning(f"Release already exists for {language}#{name}")
             return False
 
         date_epoch = int(date.timestamp())
-        date_str = datetime.fromtimestamp(date_epoch).strftime('%m/%d/%Y')
+        date_str = datetime.fromtimestamp(date_epoch).strftime("%m/%d/%Y")
 
         release_id = self.release_db.append([name, language, tag, package, version, date_epoch, date_str])
 
@@ -178,8 +182,8 @@ def _query_release(self, name: str, language: str) -> Union[str, None]:
 
 
 def _repository_owner(repository: str) -> str:
-    return re.match(r'https://github.com/([^/:]+)/.*', repository).group(1)
+    return re.match(r"https://github.com/([^/:]+)/.*", repository).group(1)
 
 
 def _repository_name(repository: str) -> str:
-    return re.match(r'https://github.com/[^/:]+/(.*)', repository).group(1)
+    return re.match(r"https://github.com/[^/:]+/(.*)", repository).group(1)

tools/azure-rest-api-specs-examples-automation/automation/github.py

@@ -4,7 +4,7 @@
 
 
 class GitHubRepository:
-    api_host: str = 'https://api.github.com'
+    api_host: str = "https://api.github.com"
     owner: str
     name: str
     token: str
@@ -15,84 +15,65 @@ def __init__(self, owner: str, name: str, token):
         self.token = token
 
     def create_pull_request(self, title: str, head: str, base: str) -> int:
-        logging.info(f'Create pull request: {head}')
+        logging.info(f"Create pull request: {head}")
 
-        request_uri = f'{self.api_host}/repos/{self.owner}/{self.name}/pulls'
-        request_body = {
-            'title': title,
-            'head': head,
-            'base': base
-        }
-        pull_request_response = requests.post(request_uri,
-                                              json=request_body,
-                                              headers=self._headers())
+        request_uri = f"{self.api_host}/repos/{self.owner}/{self.name}/pulls"
+        request_body = {"title": title, "head": head, "base": base}
+        pull_request_response = requests.post(request_uri, json=request_body, headers=self._headers())
         if pull_request_response.status_code == 201:
-            logging.info('Pull request created')
-            return pull_request_response.json()['number']
+            logging.info("Pull request created")
+            return pull_request_response.json()["number"]
         else:
-            logging.error(f'Request failed: {pull_request_response.status_code}\n{pull_request_response.json()}')
+            logging.error(f"Request failed: {pull_request_response.status_code}\n{pull_request_response.json()}")
             pull_request_response.raise_for_status()
 
     def list_pull_requests(self) -> List[Dict[str, Any]]:
-        logging.info(f'List pull requests')
+        logging.info(f"List pull requests")
 
-        request_uri = f'{self.api_host}/repos/{self.owner}/{self.name}/pulls?per_page=100'
-        pull_request_response = requests.get(request_uri,
-                                             headers=self._headers())
+        request_uri = f"{self.api_host}/repos/{self.owner}/{self.name}/pulls?per_page=100"
+        pull_request_response = requests.get(request_uri, headers=self._headers())
         if pull_request_response.status_code == 200:
-            logging.info('Pull request created')
+            logging.info("Pull request created")
             return pull_request_response.json()
         else:
-            logging.error(f'Request failed: {pull_request_response.status_code}\n{pull_request_response.json()}')
+            logging.error(f"Request failed: {pull_request_response.status_code}\n{pull_request_response.json()}")
             return []
 
     def merge_pull_request(self, pull_request: Dict):
-        title = pull_request['title']
-        logging.info(f'Merge pull request: {title}')
+        title = pull_request["title"]
+        logging.info(f"Merge pull request: {title}")
 
-        pull_number = int(pull_request['number'])
+        pull_number = int(pull_request["number"])
 
-        request_uri = f'{self.api_host}/repos/{self.owner}/{self.name}/pulls/{pull_number}/merge'
-        request_body = {
-            'commit_title': title,
-            'merge_method': 'squash'
-        }
-        merge_response = requests.put(request_uri,
-                                      json=request_body,
-                                      headers=self._headers())
+        request_uri = f"{self.api_host}/repos/{self.owner}/{self.name}/pulls/{pull_number}/merge"
+        request_body = {"commit_title": title, "merge_method": "squash"}
+        merge_response = requests.put(request_uri, json=request_body, headers=self._headers())
         if merge_response.status_code == 200:
-            logging.info('Pull request merged')
+            logging.info("Pull request merged")
         else:
-            logging.error(f'Request failed: {merge_response.status_code}\n{merge_response.json()}')
+            logging.error(f"Request failed: {merge_response.status_code}\n{merge_response.json()}")
             merge_response.raise_for_status()
 
     def list_releases(self, per_page: int, page: int = 1) -> List[Dict[str, Any]]:
-        request_uri = f'{self.api_host}/repos/{self.owner}/{self.name}/releases'
-        releases_response = requests.get(request_uri,
-                                         params={'per_page': per_page, 'page': page},
-                                         headers=self._headers())
+        request_uri = f"{self.api_host}/repos/{self.owner}/{self.name}/releases"
+        releases_response = requests.get(
+            request_uri, params={"per_page": per_page, "page": page}, headers=self._headers()
+        )
         if releases_response.status_code == 200:
             return releases_response.json()
         else:
-            logging.error(f'Request failed: {releases_response.status_code}\n{releases_response.json()}')
+            logging.error(f"Request failed: {releases_response.status_code}\n{releases_response.json()}")
             releases_response.raise_for_status()
 
     def add_label(self, pull_number: int, labels: List[str]):
-        request_uri = f'{self.api_host}/repos/{self.owner}/{self.name}/issues/{pull_number}/labels'
-        request_body = {
-            'labels': labels
-        }
-        add_label_response = requests.post(request_uri,
-                                           json=request_body,
-                                           headers=self._headers())
+        request_uri = f"{self.api_host}/repos/{self.owner}/{self.name}/issues/{pull_number}/labels"
+        request_body = {"labels": labels}
+        add_label_response = requests.post(request_uri, json=request_body, headers=self._headers())
         if add_label_response.status_code == 200:
-            logging.info('Label added')
+            logging.info("Label added")
         else:
-            logging.error(f'Request failed: {add_label_response.status_code}\n{add_label_response.json()}')
+            logging.error(f"Request failed: {add_label_response.status_code}\n{add_label_response.json()}")
             add_label_response.raise_for_status()
 
     def _headers(self) -> Dict[str, str]:
-        return {
-            'X-GitHub-Api-Version': '2022-11-28',
-            'Authorization': f'token {self.token}'
-        }
+        return {"X-GitHub-Api-Version": "2022-11-28", "Authorization": f"token {self.token}"}