Coverage for .nox/test-3-9/lib/python3.9/site-packages/nskit/vcs/codebase.py: 82%

1"""Manage a codebase.""" 

2from __future__ import annotations 

3 

4from pathlib import Path 

5import sys 

6from typing import List, Optional 

7 

8if sys.version_info.major <= 3 and sys.version_info.minor <= 8: 

9 from typing_extensions import Annotated 

10else: 

11 from typing import Annotated 

12 

13from pydantic import Field, field_validator, ValidationInfo 

14 

15from nskit.common.configuration import BaseConfiguration 

16from nskit.mixer import Recipe 

17from nskit.vcs.namespace_validator import NamespaceValidator 

18from nskit.vcs.repo import NamespaceOptionsType, NamespaceValidationRepo, Repo 

19from nskit.vcs.settings import CodebaseSettings 

20 

21 

22class Codebase(BaseConfiguration): 

23 """Object for managing a codebase.""" 

24 root_dir: Path = Field(default_factory=Path.cwd) 

25 namespaces_dir: Path = Path('.namespaces') 

26 settings: Annotated[CodebaseSettings, Field(validate_default=True)] = None 

27 namespace_validation_repo: Optional[NamespaceValidationRepo] = None 

28 

29 @field_validator('settings', mode='before') 

30 @classmethod 

31 def _validate_settings(cls, value, info: ValidationInfo): 

32 if value is None: 

33 namespace_validation_repo = None 

34 if (info.data.get('root_dir')/info.data.get('namespaces_dir')).exists(): 

35 # Namespaces repo exists 

36 namespace_validation_repo = NamespaceValidationRepo(local_dir=info.data.get('root_dir')/info.data.get('namespaces_dir')) 

37 value = CodebaseSettings(namespace_validation_repo=namespace_validation_repo) 

38 return value 

39 

40 @field_validator('namespace_validation_repo', mode='before') 

41 @classmethod 

42 def _validate_namespace_validation_repo_from_settings(cls, value, info: ValidationInfo): 

43 if value is None: 

44 try: 

45 value = info.data.get('settings').namespace_validation_repo 

46 except (AttributeError) as e: 

47 raise ValueError(e) from None 

48 return value 

49 

50 @field_validator('namespace_validation_repo', mode='after') 

51 @classmethod 

52 def _validate_namespace_validation_repo(cls, value, info: ValidationInfo): 

53 if value: 

54 value.local_dir = info.data.get('root_dir')/info.data.get('namespaces_dir') 

55 return value 

56 

57 def model_post_init(self, *args, **kwargs): 

58 """Set the settings namespace validation repo to the same as the codebase.""" 

59 super().model_post_init(*args, **kwargs) 

60 self.settings.namespace_validation_repo = self.namespace_validation_repo 

61 

62 @property 

63 def namespace_validator(self): 

64 """Get the namespace validator object.""" 

65 if self.namespace_validation_repo: 

66 return self.namespace_validation_repo.validator 

67 else: 

68 return NamespaceValidator(options=None) 

69 

70 def list_repos(self): 

71 """Get the repo names that are validated by the namespace_validator if provided.""" 

72 potential_repos = self.settings.provider_settings.repo_client.list() 

73 sdk_repos = [] 

74 for repo in potential_repos: 

75 result, _ = self.namespace_validator.validate_name(repo) 

76 if result: 

77 sdk_repos.append(repo) 

78 return sdk_repos 

79 

80 def clone(self): 

81 """Clone all repos that match the codebase to a local (nested) directory.""" 

82 # List repos 

83 root = self.root_dir 

84 root.mkdir(exist_ok=True, parents=True) 

85 repos = self.list_repos() 

86 # Create folder structure based on namespacing 

87 cloned = [] 

88 for repo in repos: 

89 repo_dir = self.root_dir/Path(*self.namespace_validator.to_parts(repo)) 

90 r = Repo( 

91 name=repo, 

92 local_dir=repo_dir, 

93 namespace_validation_repo=self.namespace_validation_repo, 

94 validation_level=self.settings.validation_level, 

95 provider_client=self.settings.provider_settings.repo_client) 

96 if not r.exists_locally: 

97 r.clone() 

98 r.install(codebase=self, deps=False) 

99 cloned.append(r) 

100 # Once installed all with no deps, install deps again 

101 for repo in cloned: 

102 repo.install(codebase=self, deps=True) 

103 return cloned 

104 

105 def create_repo(self, name, with_recipe: Optional[str] = None, **recipe_kwargs): 

106 """Create a repo in the codebase. 

107 

108 with_recipe will instantiate it with a specific recipe - the kwargs need to be provided to the call. 

109 """ 

110 repo_dir = self.root_dir/Path(*self.namespace_validator.to_parts(name)) 

111 r = Repo( 

112 name=name, 

113 local_dir=repo_dir, 

114 namespace_validation_repo=self.namespace_validation_repo, 

115 validation_level=self.settings.validation_level, 

116 provider_client=self.settings.provider_settings.repo_client) 

117 if r.exists or r.exists_locally: 

118 raise ValueError(f'Repo {name} already exists') 

119 r.create() 

120 if with_recipe is not None: 

121 repo = recipe_kwargs.get('repo', {}) 

122 repo['url'] = repo.get('url', r.url) 

123 repo['repo_separator'] = repo.get('repo_separator', self.namespace_validator.repo_separator) 

124 recipe_kwargs['repo'] = repo 

125 recipe = Recipe.load( 

126 with_recipe, 

127 name=repo['repo_separator'].join(self.namespace_validator.to_parts(r.name)), 

128 **recipe_kwargs 

129 ) 

130 created = recipe.create( 

131 base_path=r.local_dir.parent, 

132 override_path=self.namespace_validator.to_parts(r.name)[-1] 

133 ) 

134 r.commit('Initial commit', hooks=False) 

135 r.push() 

136 r.install(codebase=self, deps=True) 

137 return created 

138 

139 def delete_repo(self, name): 

140 """Delete a repo from the codebase.""" 

141 repo_dir = self.root_dir/Path(*self.namespace_validator.to_parts(name)) 

142 r = Repo( 

143 name=name, 

144 local_dir=repo_dir, 

145 namespace_validation_repo=self.namespace_validation_repo, 

146 validation_level=self.settings.validation_level, 

147 provider_client=self.settings.provider_settings.repo_client) 

148 r.delete() 

149 

150 def create_namespace_repo( 

151 self, 

152 name: str | None = None, 

153 *, 

154 namespace_options: NamespaceOptionsType | NamespaceValidator, 

155 delimiters: List[str] | None = None, 

156 repo_separator: str | None = None, 

157 namespaces_filename: str | Path = 'namespaces.yaml'): 

158 """Create and populate the validator repo.""" 

159 if name is None: 

160 name = self.namespaces_dir.name 

161 self.namespace_validation_repo = NamespaceValidationRepo( 

162 name=name, 

163 namespaces_filename=namespaces_filename, 

164 local_dir=self.root_dir/self.namespaces_dir 

165 ) 

166 self.namespace_validation_repo.create( 

167 namespace_options=namespace_options, 

168 delimiters=delimiters, 

169 repo_separator=repo_separator 

170 )