Coverage for src / gh_secrets_and_vars_async / push.py: 45%

1import asyncio 

2import os 

3from pathlib import Path 

4 

5import click 

6import github.GithubException 

7from dotenv import load_dotenv 

8from loguru import logger 

9from rich import print 

10 

11from .common import configure_logging, get_github_repo 

12 

13 

14@click.command("sync") 

15@click.option("--verbose", "-v", is_flag=True, help="Print all the output.") 

16@click.option( 

17 "--dry-run", "-d", is_flag=True, help="Run through the process, but make no changes to GitHub." 

18) 

19@click.argument("filename", type=click.Path(exists=True, readable=True), default=".env") 

20def push_command(verbose: bool, dry_run: bool, filename: click.Path): 

21 """Push .env secrets and variables to a GitHub repository. 

22 

23 Requires GH_REPO, GH_ACCOUNT, and GH_TOKEN in your .env file. 

24 Sensitive keys (containing TOKEN, SECRET, KEY, etc.) become GitHub secrets; 

25 all others become GitHub variables. 

26 """ 

27 configure_logging(verbose) 

28 if dry_run: 

29 logger.info("Dry run mode enabled. No changes will be made to GitHub.") 

30 results = asyncio.run(perform_update(filename, verbose, dry_run)) 

31 if verbose: 

32 print(results) 

33 total_secrets = len(results["SECRETS"]) 

34 total_vars = len(results["VARIABLES"]) 

35 print(f"Updated {total_secrets} secrets and {total_vars} variables.") 

36 

37 

38async def perform_update(filename, verbose: bool = False, dry_run: bool = False): 

39 """Perform the update of GitHub repository secrets and environment variables.""" 

40 if not filename: 

41 raise ValueError("No filename specified. Exiting to avoid an accident.") 

42 

43 load_dotenv(str(filename), override=True) 

44 github_repo = os.environ.get("GH_REPO", None) 

45 github_account = os.environ.get("GH_ACCOUNT", None) 

46 

47 file_path = Path(filename.__str__()) 

48 secrets = {} 

49 not_secrets = {} 

50 SECRETS_INDICATORS = [ 

51 "secret", 

52 "key", 

53 "token", 

54 "bearer", 

55 "password", 

56 "pass", 

57 "pwd", 

58 "pword", 

59 "hash", 

60 ] 

61 logger.debug(f"Reading file {file_path}") 

62 

63 with file_path.open("r") as file: 

64 for line in file: 

65 line = line.strip() 

66 if not line or line.startswith("#") or line.startswith(";") or "=" not in line: 

67 continue 

68 key, value = line.strip().split("=", 1) 

69 

70 match key: 

71 case key if key.startswith("AWS_PROFILE"): 

72 continue 

73 case key if any(indicator in key.casefold() for indicator in SECRETS_INDICATORS): 

74 secrets[key] = value 

75 case _: 

76 not_secrets[key] = value 

77 

78 try: 

79 repo = get_github_repo(github_account or "", github_repo or "") 

80 except github.GithubException: 

81 logger.critical( 

82 f"Repo {github_repo} not found. Ensure GH_REPO and GH_ACCOUNT are in your env file." 

83 ) 

84 exit(1) 

85 

86 secret_update_result = await create_or_update_github_secrets( 

87 repo=repo, env_data=secrets, dry_run=dry_run 

88 ) 

89 not_secret_update_result = await create_or_update_github_variables( 

90 repo=repo, env_data=not_secrets, dry_run=dry_run 

91 ) 

92 

93 results = {"SECRETS": secret_update_result, "VARIABLES": not_secret_update_result} 

94 

95 return results 

96 

97 

98async def create_or_update_github_secrets(repo, env_data, dry_run: bool = False): 

99 """Create or update GitHub repository secrets.""" 

100 secrets = await asyncio.to_thread(repo.get_secrets) 

101 secret_names = [secret.name for secret in secrets] 

102 dry_run_prefix = "[DRY RUN] " if dry_run else "" 

103 

104 tasks = [] 

105 for env_var_name, env_var_value in env_data.items(): 

106 if env_var_name in secret_names: 

107 logger.info(f"{dry_run_prefix}Updating secret {env_var_name}...") 

108 tasks.append(asyncio.to_thread(repo.create_secret, env_var_name, env_var_value)) 

109 else: 

110 logger.info(f"{dry_run_prefix}Creating secret {env_var_name}...") 

111 tasks.append(asyncio.to_thread(repo.create_secret, env_var_name, env_var_value)) 

112 

113 for secret_name in secret_names: 

114 if secret_name not in env_data.keys(): 

115 logger.info(f"{dry_run_prefix}Deleting secret {secret_name}...") 

116 tasks.append(asyncio.to_thread(repo.delete_secret, secret_name)) 

117 

118 if dry_run: 

119 results = list(secrets) 

120 else: 

121 results = await asyncio.gather(*tasks) 

122 return results 

123 

124 

125async def create_or_update_github_variables(repo, env_data, dry_run: bool = False): 

126 """Create or update GitHub repository environment variables.""" 

127 vars = await asyncio.to_thread(repo.get_variables) 

128 var_names = [var.name for var in vars] 

129 dry_run_prefix = "[DRY RUN] " if dry_run else "" 

130 

131 tasks = [] 

132 for env_var_name, env_var_value in env_data.items(): 

133 if env_var_name in var_names: 

134 logger.info(f"{dry_run_prefix}Updating variable {env_var_name}...") 

135 

136 def delete_then_create_variable(repo, env_var_name, env_var_value): 

137 repo.delete_variable(env_var_name) 

138 return repo.create_variable(env_var_name, env_var_value) 

139 

140 tasks.append( 

141 asyncio.to_thread(delete_then_create_variable, repo, env_var_name, env_var_value) 

142 ) 

143 else: 

144 logger.info(f"{dry_run_prefix}Creating variable {env_var_name}...") 

145 tasks.append(asyncio.to_thread(repo.create_variable, env_var_name, env_var_value)) 

146 

147 for var_name in var_names: 

148 if var_name not in env_data.keys(): 

149 logger.info(f"{dry_run_prefix}Deleting variable {var_name}...") 

150 tasks.append(asyncio.to_thread(repo.delete_variable, var_name)) 

151 

152 if dry_run: 

153 results = list(vars) 

154 else: 

155 results = await asyncio.gather(*tasks) 

156 

157 return results