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

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 settings: Annotated[CodebaseSettings, Field(validate_default=True)] = None 

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

27 namespace_validation_repo: Optional[NamespaceValidationRepo] = None 

28 

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

30 @classmethod 

31 def _validate_settings(cls, value): 

32 if value is None: 

33 value = CodebaseSettings() 

34 return value 

35 

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

37 @classmethod 

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

39 if value is None: 

40 try: 

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

42 except (AttributeError) as e: 

43 raise ValueError(e) from None 

44 return value 

45 

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

47 @classmethod 

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

49 if value: 

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

51 return value 

52 

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

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

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

56 self.settings.namespace_validation_repo = self.namespace_validation_repo 

57 

58 @property 

59 def namespace_validator(self): 

60 """Get the namespace validator object.""" 

61 if self.namespace_validation_repo: 

62 return self.namespace_validation_repo.validator 

63 else: 

64 return NamespaceValidator(options=None) 

65 

66 def list_repos(self): 

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

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

69 sdk_repos = [] 

70 for repo in potential_repos: 

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

72 if result: 

73 sdk_repos.append(repo) 

74 return sdk_repos 

75 

76 def clone(self): 

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

78 # List repos 

79 root = self.root_dir 

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

81 repos = self.list_repos() 

82 # Create folder structure based on namespacing 

83 cloned = [] 

84 for repo in repos: 

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

86 r = Repo( 

87 name=repo, 

88 local_dir=repo_dir, 

89 namespace_validation_repo=self.namespace_validation_repo, 

90 validation_level=self.settings.validation_level, 

91 provider_client=self.settings.provider_settings.repo_client) 

92 if not r.exists_locally: 

93 r.clone() 

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

95 cloned.append(r) 

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

97 for repo in cloned: 

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

99 return cloned 

100 

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

102 """Create a repo in the codebase. 

103 

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

105 """ 

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

107 r = Repo( 

108 name=name, 

109 local_dir=repo_dir, 

110 namespace_validation_repo=self.namespace_validation_repo, 

111 validation_level=self.settings.validation_level, 

112 provider_client=self.settings.provider_settings.repo_client) 

113 if r.exists or r.exists_locally: 

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

115 r.create() 

116 if with_recipe is not None: 

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

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

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

120 recipe_kwargs['repo'] = repo 

121 recipe = Recipe.load( 

122 with_recipe, 

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

124 **recipe_kwargs 

125 ) 

126 created = recipe.create( 

127 base_path=r.local_dir.parent, 

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

129 ) 

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

131 r.push() 

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

133 return created 

134 

135 def delete_repo(self, name): 

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

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

138 r = Repo( 

139 name=name, 

140 local_dir=repo_dir, 

141 namespace_validation_repo=self.namespace_validation_repo, 

142 validation_level=self.settings.validation_level, 

143 provider_client=self.settings.provider_settings.repo_client) 

144 r.delete() 

145 

146 def create_namespace_repo( 

147 self, 

148 name: str | None = None, 

149 *, 

150 namespace_options: NamespaceOptionsType | NamespaceValidator, 

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

152 repo_separator: str | None = None, 

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

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

155 if name is None: 

156 name = self.namespaces_dir.name 

157 self.namespace_validation_repo = NamespaceValidationRepo( 

158 name=name, 

159 namespaces_filename=namespaces_filename, 

160 local_dir=self.namespaces_dir 

161 ) 

162 self.namespace_validation_repo.create( 

163 namespace_options=namespace_options, 

164 delimiters=delimiters, 

165 repo_separator=repo_separator 

166 )