Skip to content

nskit.vcs

VCS and Codebase handlers for creating repo (and build) infrastructure.

NamespaceOptionsType = TypeAliasType('NamespaceOptionsType', List[Union[str, Dict[str, 'NamespaceOptionsType']]]) module-attribute

Codebase

Bases: BaseConfiguration

Object for managing a codebase.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/codebase.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
class Codebase(BaseConfiguration):
    """Object for managing a codebase."""
    root_dir: Path = Field(default_factory=Path.cwd)
    namespaces_dir: Path = Path('.namespaces')
    settings: Annotated[CodebaseSettings, Field(validate_default=True)] = None
    namespace_validation_repo: Optional[NamespaceValidationRepo] = None

    @field_validator('settings', mode='before')
    @classmethod
    def _validate_settings(cls, value, info: ValidationInfo):
        if value is None:
            namespace_validation_repo = None
            if (info.data.get('root_dir')/info.data.get('namespaces_dir')).exists():
                # Namespaces repo exists
                namespace_validation_repo = NamespaceValidationRepo(local_dir=info.data.get('root_dir')/info.data.get('namespaces_dir'))
            value = CodebaseSettings(namespace_validation_repo=namespace_validation_repo)
        return value

    @field_validator('namespace_validation_repo', mode='before')
    @classmethod
    def _validate_namespace_validation_repo_from_settings(cls, value, info: ValidationInfo):
        if value is None:
            try:
                value = info.data.get('settings').namespace_validation_repo
            except (AttributeError) as e:
                raise ValueError(e) from None
        return value

    @field_validator('namespace_validation_repo', mode='after')
    @classmethod
    def _validate_namespace_validation_repo(cls, value, info: ValidationInfo):
        if value:
            value.local_dir = info.data.get('root_dir')/info.data.get('namespaces_dir')
        return value

    def model_post_init(self, *args, **kwargs):
        """Set the settings namespace validation repo to the same as the codebase."""
        super().model_post_init(*args, **kwargs)
        self.settings.namespace_validation_repo = self.namespace_validation_repo

    @property
    def namespace_validator(self):
        """Get the namespace validator object."""
        if self.namespace_validation_repo:
            return self.namespace_validation_repo.validator
        else:
            return NamespaceValidator(options=None)

    def list_repos(self):
        """Get the repo names that are validated by the namespace_validator if provided."""
        potential_repos = self.settings.provider_settings.repo_client.list()
        sdk_repos = []
        for repo in potential_repos:
            result, _ = self.namespace_validator.validate_name(repo)
            if result:
                sdk_repos.append(repo)
        return sdk_repos

    def clone(self):
        """Clone all repos that match the codebase to a local (nested) directory."""
        # List repos
        root = self.root_dir
        root.mkdir(exist_ok=True, parents=True)
        repos = self.list_repos()
        # Create folder structure based on namespacing
        cloned = []
        for repo in repos:
            repo_dir = self.root_dir/Path(*self.namespace_validator.to_parts(repo))
            r = Repo(
                name=repo,
                local_dir=repo_dir,
                namespace_validation_repo=self.namespace_validation_repo,
                validation_level=self.settings.validation_level,
                provider_client=self.settings.provider_settings.repo_client)
            if not r.exists_locally:
                r.clone()
            r.install(codebase=self, deps=False)
            cloned.append(r)
        # Once installed all with no deps, install deps again
        for repo in cloned:
            repo.install(codebase=self, deps=True)
        return cloned

    def create_repo(self, name, with_recipe: Optional[str] = None, **recipe_kwargs):
        """Create a repo in the codebase.

        with_recipe will instantiate it with a specific recipe - the kwargs need to be provided to the call.
        """
        repo_dir = self.root_dir/Path(*self.namespace_validator.to_parts(name))
        r = Repo(
            name=name,
            local_dir=repo_dir,
            namespace_validation_repo=self.namespace_validation_repo,
            validation_level=self.settings.validation_level,
            provider_client=self.settings.provider_settings.repo_client)
        if r.exists or r.exists_locally:
            raise ValueError(f'Repo {name} already exists')
        r.create()
        if with_recipe is not None:
            repo = recipe_kwargs.get('repo', {})
            repo['url'] = repo.get('url', r.url)
            repo['repo_separator'] = repo.get('repo_separator', self.namespace_validator.repo_separator)
            recipe_kwargs['repo'] = repo
            recipe = Recipe.load(
                with_recipe,
                name=repo['repo_separator'].join(self.namespace_validator.to_parts(r.name)),
                **recipe_kwargs
            )
            created = recipe.create(
                base_path=r.local_dir.parent,
                override_path=self.namespace_validator.to_parts(r.name)[-1]
            )
            r.commit('Initial commit', hooks=False)
            r.push()
            r.install(codebase=self, deps=True)
            return created

    def delete_repo(self, name):
        """Delete a repo from the codebase."""
        repo_dir = self.root_dir/Path(*self.namespace_validator.to_parts(name))
        r = Repo(
            name=name,
            local_dir=repo_dir,
            namespace_validation_repo=self.namespace_validation_repo,
            validation_level=self.settings.validation_level,
            provider_client=self.settings.provider_settings.repo_client)
        r.delete()

    def create_namespace_repo(
            self,
            name: str | None = None,
            *,
            namespace_options: NamespaceOptionsType | NamespaceValidator,
            delimiters: List[str] | None = None,
            repo_separator: str | None = None,
            namespaces_filename: str | Path = 'namespaces.yaml'):
        """Create and populate the validator repo."""
        if name is None:
            name = self.namespaces_dir.name
        self.namespace_validation_repo = NamespaceValidationRepo(
            name=name,
            namespaces_filename=namespaces_filename,
            local_dir=self.root_dir/self.namespaces_dir
        )
        self.namespace_validation_repo.create(
            namespace_options=namespace_options,
            delimiters=delimiters,
            repo_separator=repo_separator
        )

namespace_validator property

Get the namespace validator object.

clone()

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

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/codebase.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def clone(self):
    """Clone all repos that match the codebase to a local (nested) directory."""
    # List repos
    root = self.root_dir
    root.mkdir(exist_ok=True, parents=True)
    repos = self.list_repos()
    # Create folder structure based on namespacing
    cloned = []
    for repo in repos:
        repo_dir = self.root_dir/Path(*self.namespace_validator.to_parts(repo))
        r = Repo(
            name=repo,
            local_dir=repo_dir,
            namespace_validation_repo=self.namespace_validation_repo,
            validation_level=self.settings.validation_level,
            provider_client=self.settings.provider_settings.repo_client)
        if not r.exists_locally:
            r.clone()
        r.install(codebase=self, deps=False)
        cloned.append(r)
    # Once installed all with no deps, install deps again
    for repo in cloned:
        repo.install(codebase=self, deps=True)
    return cloned

create_namespace_repo(name=None, *, namespace_options, delimiters=None, repo_separator=None, namespaces_filename='namespaces.yaml')

Create and populate the validator repo.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/codebase.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def create_namespace_repo(
        self,
        name: str | None = None,
        *,
        namespace_options: NamespaceOptionsType | NamespaceValidator,
        delimiters: List[str] | None = None,
        repo_separator: str | None = None,
        namespaces_filename: str | Path = 'namespaces.yaml'):
    """Create and populate the validator repo."""
    if name is None:
        name = self.namespaces_dir.name
    self.namespace_validation_repo = NamespaceValidationRepo(
        name=name,
        namespaces_filename=namespaces_filename,
        local_dir=self.root_dir/self.namespaces_dir
    )
    self.namespace_validation_repo.create(
        namespace_options=namespace_options,
        delimiters=delimiters,
        repo_separator=repo_separator
    )

create_repo(name, with_recipe=None, **recipe_kwargs)

Create a repo in the codebase.

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

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/codebase.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def create_repo(self, name, with_recipe: Optional[str] = None, **recipe_kwargs):
    """Create a repo in the codebase.

    with_recipe will instantiate it with a specific recipe - the kwargs need to be provided to the call.
    """
    repo_dir = self.root_dir/Path(*self.namespace_validator.to_parts(name))
    r = Repo(
        name=name,
        local_dir=repo_dir,
        namespace_validation_repo=self.namespace_validation_repo,
        validation_level=self.settings.validation_level,
        provider_client=self.settings.provider_settings.repo_client)
    if r.exists or r.exists_locally:
        raise ValueError(f'Repo {name} already exists')
    r.create()
    if with_recipe is not None:
        repo = recipe_kwargs.get('repo', {})
        repo['url'] = repo.get('url', r.url)
        repo['repo_separator'] = repo.get('repo_separator', self.namespace_validator.repo_separator)
        recipe_kwargs['repo'] = repo
        recipe = Recipe.load(
            with_recipe,
            name=repo['repo_separator'].join(self.namespace_validator.to_parts(r.name)),
            **recipe_kwargs
        )
        created = recipe.create(
            base_path=r.local_dir.parent,
            override_path=self.namespace_validator.to_parts(r.name)[-1]
        )
        r.commit('Initial commit', hooks=False)
        r.push()
        r.install(codebase=self, deps=True)
        return created

delete_repo(name)

Delete a repo from the codebase.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/codebase.py 
139
140
141
142
143
144
145
146
147
148
def delete_repo(self, name):
    """Delete a repo from the codebase."""
    repo_dir = self.root_dir/Path(*self.namespace_validator.to_parts(name))
    r = Repo(
        name=name,
        local_dir=repo_dir,
        namespace_validation_repo=self.namespace_validation_repo,
        validation_level=self.settings.validation_level,
        provider_client=self.settings.provider_settings.repo_client)
    r.delete()

list_repos()

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

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/codebase.py 
70
71
72
73
74
75
76
77
78
def list_repos(self):
    """Get the repo names that are validated by the namespace_validator if provided."""
    potential_repos = self.settings.provider_settings.repo_client.list()
    sdk_repos = []
    for repo in potential_repos:
        result, _ = self.namespace_validator.validate_name(repo)
        if result:
            sdk_repos.append(repo)
    return sdk_repos

model_post_init(*args, **kwargs)

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

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/codebase.py 
57
58
59
60
def model_post_init(self, *args, **kwargs):
    """Set the settings namespace validation repo to the same as the codebase."""
    super().model_post_init(*args, **kwargs)
    self.settings.namespace_validation_repo = self.namespace_validation_repo

Repo

Bases: _Repo

Repo with namespace validator.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/repo.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
class Repo(_Repo):
    """Repo with namespace validator."""

    namespace_validation_repo: Optional[NamespaceValidationRepo] = None
    validation_level: ValidationEnum = ValidationEnum.none
    name: str

    @model_validator(mode='after')
    def _validate_name(self):
        value = self.name
        if self.namespace_validation_repo and self.validation_level in [ValidationEnum.strict, ValidationEnum.warn]:
            namespace_validator = self.namespace_validation_repo.validator
            result, message = namespace_validator.validate_name(value)
            if not result:
                message = (f'{value} {message.format(key="<root>")}')
            value = namespace_validator.to_repo_name(value)
            if self.validation_level == ValidationEnum.strict and not result:
                raise ValueError(message)
            elif not result:
                warnings.warn(message, stacklevel=2)
        self.name = value

NamespaceValidator

Bases: BaseConfiguration

Namespace Validator object.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/namespace_validator.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
class NamespaceValidator(BaseConfiguration):
    """Namespace Validator object."""
    options: Optional[NamespaceOptionsType]
    repo_separator: str = REPO_SEPARATOR
    delimiters: List[str] = _DELIMITERS

    __delimiters_regexp = None
    # Validate delimiters to add repo_separator

    @field_validator('delimiters', mode='after')
    @classmethod
    def _validate_repo_separator_in_delimiters(cls, v: List[str], info: ValidationInfo):
        if info.data['repo_separator'] not in v:
            v.append(info.data['repo_separator'])
        return v

    @property
    def _delimiters_regexp(self):
        if self.__delimiters_regexp is None:
            self.__delimiters_regexp = '|'.join(map(re.escape, self.delimiters))
        return self.__delimiters_regexp

    def to_parts(self, name: str):
        """Break the name into the namespace parts."""
        if self.options:
            return re.split(self._delimiters_regexp, name)
        return [name]

    def to_repo_name(self, name: str):
        """Convert the name to the appropriate name with a given repo separator."""
        return self.repo_separator.join(self.to_parts(name))

    def validate_name(self, proposed_name: str):
        """Validate a proposed name."""
        name_parts = self.to_parts(proposed_name)
        if self.options:
            result, message = self._validate_level(name_parts, self.options)
            message = message.format(key='<root>')
        else:
            result = True
            message = 'no constraints set'
        return result, message

    def _validate_level(
            self,
            name_parts: List[str],
            partial_namespace: List[Union[str, Dict]]):
        not_matched = []
        for key in partial_namespace:
            # If it is a dict, then there are mappings of <section>: [<subsection 1>, <subsection 2>]
            if isinstance(key, dict):
                for sub_key, new_partial_namespace in key.items():
                    if sub_key == name_parts[0]:
                        # This maps to a section with subsections, so we need to validate those
                        result, message = self._validate_level(name_parts[1:], new_partial_namespace)
                        if not result:
                            message = message.format(key=sub_key)
                        return result, message
                    not_matched.append(sub_key)
            # Otherwise it is a string
            elif key == name_parts[0]:
                return True, 'ok'
            else:
                not_matched.append(key)
        return False, f'Does not match valid names for {{key}}: {", ".join(not_matched)}, with delimiters: {self.delimiters}'

to_parts(name)

Break the name into the namespace parts.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/namespace_validator.py 
51
52
53
54
55
def to_parts(self, name: str):
    """Break the name into the namespace parts."""
    if self.options:
        return re.split(self._delimiters_regexp, name)
    return [name]

to_repo_name(name)

Convert the name to the appropriate name with a given repo separator.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/namespace_validator.py 
57
58
59
def to_repo_name(self, name: str):
    """Convert the name to the appropriate name with a given repo separator."""
    return self.repo_separator.join(self.to_parts(name))

validate_name(proposed_name)

Validate a proposed name.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/namespace_validator.py 
61
62
63
64
65
66
67
68
69
70
def validate_name(self, proposed_name: str):
    """Validate a proposed name."""
    name_parts = self.to_parts(proposed_name)
    if self.options:
        result, message = self._validate_level(name_parts, self.options)
        message = message.format(key='<root>')
    else:
        result = True
        message = 'no constraints set'
    return result, message

Installers

Repository installers.

Installer

Bases: ABC, BaseConfiguration

Abstract class for language installer.

Can be enabled or disabled using the boolean flag and environment variables.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/installer.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class Installer(ABC, BaseConfiguration):
    """Abstract class for language installer.

    Can be enabled or disabled using the boolean flag and environment variables.
    """

    enabled: bool = True

    def check(self, path: Path, **kwargs):
        """Check if the installer is enabled, and the repo matches the criteria."""
        return self.enabled and self.check_repo(path=path, **kwargs)

    @abstractmethod
    def check_repo(self, path: Path, **kwargs):
        """Check if the repo matches the installer language."""
        raise NotImplementedError('Implement in a language specific installer. It should check for appropriate files to signal that it is a repo of that type')

    @abstractmethod
    def install(self, path: Path, *, codebase: Codebase | None = None, deps: bool = True, **kwargs):
        """Install the repo into the appropriate environment."""
        raise NotImplementedError('Implement in language specific installer. It should take in any language specific environment/executables')

check(path, **kwargs)

Check if the installer is enabled, and the repo matches the criteria.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/installer.py 
36
37
38
def check(self, path: Path, **kwargs):
    """Check if the installer is enabled, and the repo matches the criteria."""
    return self.enabled and self.check_repo(path=path, **kwargs)

check_repo(path, **kwargs) abstractmethod

Check if the repo matches the installer language.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/installer.py 
40
41
42
43
@abstractmethod
def check_repo(self, path: Path, **kwargs):
    """Check if the repo matches the installer language."""
    raise NotImplementedError('Implement in a language specific installer. It should check for appropriate files to signal that it is a repo of that type')

install(path, *, codebase=None, deps=True, **kwargs) abstractmethod

Install the repo into the appropriate environment.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/installer.py 
45
46
47
48
@abstractmethod
def install(self, path: Path, *, codebase: Codebase | None = None, deps: bool = True, **kwargs):
    """Install the repo into the appropriate environment."""
    raise NotImplementedError('Implement in language specific installer. It should take in any language specific environment/executables')

PythonInstaller

Bases: Installer

Python language installer.

Can be enabled or disabled using the boolean flag and environment variables. The virtualenv config can be updated (to a custom dir/path relative to the codebase root)

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/installer.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class PythonInstaller(Installer):
    """Python language installer.

    Can be enabled or disabled using the boolean flag and environment variables. The virtualenv config can be updated (to a custom dir/path relative to the codebase root)
    """

    model_config = SettingsConfigDict(env_prefix='NSKIT_PYTHON_INSTALLER_', env_file='.env')
    virtualenv_dir: Path = Path('.venv')
    # Include Azure DevOps seeder
    virtualenv_args: List[str] = []
    # For Azure Devops could set this to something like: ['--seeder', 'azdo-pip']

    def check_repo(self, path: Path):
        """Check if this is a python repo."""
        logger.debug(f'{self.__class__} enabled, checking for match.')
        result = (path/'setup.py').exists() or (path/'pyproject.toml').exists() or (path/'requirements.txt').exists()
        logger.info(f'Matched repo to {self.__class__}.')
        return result

    def install(self, path: Path, *, codebase: Codebase | None = None, executable: str = 'venv', deps: bool = True):
        """Install the repo.

        executable can override the executable to use (e.g. a virtualenv)
        deps controls whether dependencies are installed or not.
        """
        executable = self._get_executable(path, codebase, executable)
        logger.info(f'Installing using {executable}.')
        args = []
        if not deps:
            args.append('--no-deps')
        with ChDir(path):
            if Path('setup.py').exists() or Path('pyproject.toml').exists():
                subprocess.check_call([str(executable), '-m', 'pip', 'install', '-e', '.[dev]']+args)  # nosec B603, B607
            elif deps and Path('requirements.txt').exists():
                subprocess.check_call([str(executable), '-m', 'pip', 'install', '-r', 'requirements.txt'])  # nosec B603, B607

    def _get_virtualenv(self, full_virtualenv_dir: Path):
        """Get the virtualenv executable.

        Create's it if it doesn't exist.
        """
        if not full_virtualenv_dir.exists():
            virtualenv.cli_run([str(full_virtualenv_dir)]+self.virtualenv_args)
        if sys.platform.startswith('win'):
            executable = full_virtualenv_dir/'Scripts'/'python.exe'
        else:
            executable = full_virtualenv_dir/'bin'/'python'
        return executable.absolute()

    def _get_executable(self, path: Path, codebase: Codebase | None = None, executable: str | None = 'venv'):
        # Install in the current environment
        if self.virtualenv_dir.is_absolute():
            full_virtualenv_dir = self.virtualenv_dir
        elif codebase:
            full_virtualenv_dir = codebase.root_dir/self.virtualenv_dir
        else:
            full_virtualenv_dir = path/self.virtualenv_dir
        if executable is None:
            executable = sys.executable
        elif executable == 'venv':
            executable = self._get_virtualenv(full_virtualenv_dir=full_virtualenv_dir)
        return executable

check_repo(path)

Check if this is a python repo.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/installer.py 
63
64
65
66
67
68
def check_repo(self, path: Path):
    """Check if this is a python repo."""
    logger.debug(f'{self.__class__} enabled, checking for match.')
    result = (path/'setup.py').exists() or (path/'pyproject.toml').exists() or (path/'requirements.txt').exists()
    logger.info(f'Matched repo to {self.__class__}.')
    return result

install(path, *, codebase=None, executable='venv', deps=True)

Install the repo.

executable can override the executable to use (e.g. a virtualenv) deps controls whether dependencies are installed or not.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/installer.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def install(self, path: Path, *, codebase: Codebase | None = None, executable: str = 'venv', deps: bool = True):
    """Install the repo.

    executable can override the executable to use (e.g. a virtualenv)
    deps controls whether dependencies are installed or not.
    """
    executable = self._get_executable(path, codebase, executable)
    logger.info(f'Installing using {executable}.')
    args = []
    if not deps:
        args.append('--no-deps')
    with ChDir(path):
        if Path('setup.py').exists() or Path('pyproject.toml').exists():
            subprocess.check_call([str(executable), '-m', 'pip', 'install', '-e', '.[dev]']+args)  # nosec B603, B607
        elif deps and Path('requirements.txt').exists():
            subprocess.check_call([str(executable), '-m', 'pip', 'install', '-r', 'requirements.txt'])  # nosec B603, B607

Providers

VCS Providers, accessed using entrypoints.

Abstract Client

Abstract classes for the provider.

RepoClient

Bases: ABC

Repo management client.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/abstract.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class RepoClient(ABC):
    """Repo management client."""

    @abstractmethod
    def create(self, repo_name: str):
        """Create a repo."""
        raise NotImplementedError()

    @abstractmethod
    def get_remote_url(self, repo_name: str) -> HttpUrl:
        """Get the remote url for a repo."""
        raise NotImplementedError()

    def get_clone_url(self, repo_name: str) -> HttpUrl:
        """Get the clone URL.

        This defaults to the remote url unless specifically implemented.
        """
        return self.get_remote_url(repo_name)

    @abstractmethod
    def delete(self, repo_name: str):
        """Delete a repo."""
        raise NotImplementedError()

    @abstractmethod
    def check_exists(self, repo_name: str) -> bool:
        """Check if the repo exists on the remote."""
        raise NotImplementedError()

    @abstractmethod
    def list(self) -> List[str]:
        """List all repos on the remote."""
        raise NotImplementedError()
check_exists(repo_name) abstractmethod

Check if the repo exists on the remote.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/abstract.py 
35
36
37
38
@abstractmethod
def check_exists(self, repo_name: str) -> bool:
    """Check if the repo exists on the remote."""
    raise NotImplementedError()
create(repo_name) abstractmethod

Create a repo.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/abstract.py 
13
14
15
16
@abstractmethod
def create(self, repo_name: str):
    """Create a repo."""
    raise NotImplementedError()
delete(repo_name) abstractmethod

Delete a repo.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/abstract.py 
30
31
32
33
@abstractmethod
def delete(self, repo_name: str):
    """Delete a repo."""
    raise NotImplementedError()
get_clone_url(repo_name)

Get the clone URL.

This defaults to the remote url unless specifically implemented.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/abstract.py 
23
24
25
26
27
28
def get_clone_url(self, repo_name: str) -> HttpUrl:
    """Get the clone URL.

    This defaults to the remote url unless specifically implemented.
    """
    return self.get_remote_url(repo_name)
get_remote_url(repo_name) abstractmethod

Get the remote url for a repo.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/abstract.py 
18
19
20
21
@abstractmethod
def get_remote_url(self, repo_name: str) -> HttpUrl:
    """Get the remote url for a repo."""
    raise NotImplementedError()
list() abstractmethod

List all repos on the remote.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/abstract.py 
40
41
42
43
@abstractmethod
def list(self) -> List[str]:
    """List all repos on the remote."""
    raise NotImplementedError()

VCSProviderSettings

Bases: ABC, BaseConfiguration

Settings for VCS Provider.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/abstract.py 
46
47
48
49
50
51
52
class VCSProviderSettings(ABC, BaseConfiguration):
    """Settings for VCS Provider."""

    @abstractproperty
    def repo_client(self) -> RepoClient:
        """Return the instantiated repo client object for the provider."""
        raise NotImplementedError()
repo_client()

Return the instantiated repo client object for the provider.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/abstract.py 
49
50
51
52
@abstractproperty
def repo_client(self) -> RepoClient:
    """Return the instantiated repo client object for the provider."""
    raise NotImplementedError()

Azure Devops

Azure Devops provider using azure-cli to manage it.

AzureDevOpsRepoClient

Bases: RepoClient

Client for managing Azure DevOps repos using azure-cli.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/azure_devops.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class AzureDevOpsRepoClient(RepoClient):
    """Client for managing Azure DevOps repos using azure-cli."""

    def __init__(self, config: AzureDevOpsSettings):
        """Initialise the client."""
        self._cli = get_default_cli()
        self._config = config

    def _invoke(self, command, out_file=None):
        return self._cli.invoke(command, out_file=out_file)

    def check_exists(self, repo_name: str) -> bool:
        """Check if the repo exists in the project."""
        output = StringIO()
        return not self._invoke(['repos',
                                 'show',
                                 '--organization',
                                 self._config.organisation_url,
                                 '--project',
                                 self._config.project,
                                 '-r',
                                 repo_name],
                                out_file=output)

    def create(self, repo_name: str):
        """Create the repo in the project."""
        output = StringIO()
        return self._invoke(['repos',
                             'create',
                             '--organization',
                             self._config.organisation_url,
                             '--project',
                             self._config.project,
                             '--name',
                             repo_name],
                            out_file=output)

    def delete(self, repo_name: str):
        """Delete the repo if it exists in the project."""
        # We need to get the ID
        show_output = StringIO()
        result = self._invoke(['repos',
                               'show',
                               '--organization',
                               self._config.organisation_url,
                               '--project',
                               self._config.project,
                               '-r',
                               repo_name],
                              out_file=show_output)
        if not result:
            # Exists
            repo_info = json.loads(show_output.getvalue())
            repo_id = repo_info['id']
            output = StringIO()
            return self._invoke(['repos',
                                 'delete',
                                 '--organization',
                                 self._config.organisation_url,
                                 '--project',
                                 self._config.project,
                                 '--id',
                                 repo_id],
                                out_file=output)

    def get_remote_url(self, repo_name: str) -> HttpUrl:
        """Get the remote url for the repo."""
        output = StringIO()
        result = self._invoke(['repos',
                               'show',
                               '--organization',
                               self._config.organisation_url,
                               '--project',
                               self._config.project,
                               '-r',
                               repo_name],
                              out_file=output)
        if not result:
            # Exists
            repo_info = json.loads(output.getvalue())
            return repo_info['remoteUrl']

    def list(self) -> List[str]:
        """List the repos in the project."""
        output = StringIO()
        result = self._invoke(['repos',
                               'list',
                               '--organization',
                               self._config.organisation_url,
                               '--project',
                               self._config.project],
                              out_file=output)
        if not result:
            # Exists
            repo_list = [u['name'] for u in json.loads(output.getvalue())]
            return repo_list
__init__(config)

Initialise the client.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/azure_devops.py 
46
47
48
49
def __init__(self, config: AzureDevOpsSettings):
    """Initialise the client."""
    self._cli = get_default_cli()
    self._config = config
check_exists(repo_name)

Check if the repo exists in the project.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/azure_devops.py 
54
55
56
57
58
59
60
61
62
63
64
65
def check_exists(self, repo_name: str) -> bool:
    """Check if the repo exists in the project."""
    output = StringIO()
    return not self._invoke(['repos',
                             'show',
                             '--organization',
                             self._config.organisation_url,
                             '--project',
                             self._config.project,
                             '-r',
                             repo_name],
                            out_file=output)
create(repo_name)

Create the repo in the project.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/azure_devops.py 
67
68
69
70
71
72
73
74
75
76
77
78
def create(self, repo_name: str):
    """Create the repo in the project."""
    output = StringIO()
    return self._invoke(['repos',
                         'create',
                         '--organization',
                         self._config.organisation_url,
                         '--project',
                         self._config.project,
                         '--name',
                         repo_name],
                        out_file=output)
delete(repo_name)

Delete the repo if it exists in the project.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/azure_devops.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def delete(self, repo_name: str):
    """Delete the repo if it exists in the project."""
    # We need to get the ID
    show_output = StringIO()
    result = self._invoke(['repos',
                           'show',
                           '--organization',
                           self._config.organisation_url,
                           '--project',
                           self._config.project,
                           '-r',
                           repo_name],
                          out_file=show_output)
    if not result:
        # Exists
        repo_info = json.loads(show_output.getvalue())
        repo_id = repo_info['id']
        output = StringIO()
        return self._invoke(['repos',
                             'delete',
                             '--organization',
                             self._config.organisation_url,
                             '--project',
                             self._config.project,
                             '--id',
                             repo_id],
                            out_file=output)
get_remote_url(repo_name)

Get the remote url for the repo.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/azure_devops.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def get_remote_url(self, repo_name: str) -> HttpUrl:
    """Get the remote url for the repo."""
    output = StringIO()
    result = self._invoke(['repos',
                           'show',
                           '--organization',
                           self._config.organisation_url,
                           '--project',
                           self._config.project,
                           '-r',
                           repo_name],
                          out_file=output)
    if not result:
        # Exists
        repo_info = json.loads(output.getvalue())
        return repo_info['remoteUrl']
list()

List the repos in the project.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/azure_devops.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def list(self) -> List[str]:
    """List the repos in the project."""
    output = StringIO()
    result = self._invoke(['repos',
                           'list',
                           '--organization',
                           self._config.organisation_url,
                           '--project',
                           self._config.project],
                          out_file=output)
    if not result:
        # Exists
        repo_list = [u['name'] for u in json.loads(output.getvalue())]
        return repo_list

AzureDevOpsSettings

Bases: VCSProviderSettings

Azure DevOps settings.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/azure_devops.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class AzureDevOpsSettings(VCSProviderSettings):
    """Azure DevOps settings."""
    model_config = SettingsConfigDict(env_prefix='AZURE_DEVOPS_', env_file='.env', dotenv_extra='ignore')

    url: HttpUrl = "https://dev.azure.com"
    organisation: str
    project: str

    @property
    def organisation_url(self):
        """Get the organistion Url."""
        return f'{self.url}/{self.organisation}'

    @property
    def project_url(self):
        """Get the project url."""
        return f'{self.organisation_url}/{self.project}'

    @property
    def repo_client(self) -> 'AzureDevOpsRepoClient':
        """Get the instantiated repo client."""
        return AzureDevOpsRepoClient(self)
organisation_url property

Get the organistion Url.

project_url property

Get the project url.

repo_client: AzureDevOpsRepoClient property

Get the instantiated repo client.

Github

Github provider using ghapi.

GithubOrgType

Bases: Enum

Org type, user or org.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/github.py 
62
63
64
65
class GithubOrgType(Enum):
    """Org type, user or org."""
    user = 'User'
    org = 'Org'

GithubRepoClient

Bases: RepoClient

Client for managing github repos.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/github.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
class GithubRepoClient(RepoClient):
    """Client for managing github repos."""

    def __init__(self, config: GithubSettings):
        """Initialise the client."""
        self._config = config
        self._github = GhApi(token=self._config.token.get_secret_value(), gh_host=str(self._config.url).rstrip('/'))
        # If the organisation is set, we get it, and assume that the token is valid
        # Otherwise default to the user
        if self._config.organisation:
            try:
                self._github.orgs.get(self._config.organisation)
                self._org_type = GithubOrgType.org
            except HTTP404NotFoundError:
                self._github.users.get_by_username(self._config.organisation)
                self._org_type = GithubOrgType.user
        else:
            self._config.organisation = self._github.users.get_authenticated()['login']
            self._org_type = GithubOrgType.user

    def create(self, repo_name: str):
        """Create the repo in the user/organisation."""
        kwargs = {
            'name': repo_name,
            'private': self._config.repo.private,
            'has_issues': self._config.repo.has_issues,
            'has_wiki': self._config.repo.has_wiki,
            'has_downloads': self._config.repo.has_downloads,
            'has_projects': self._config.repo.has_projects,
            'allow_squash_merge': self._config.repo.allow_squash_merge,
            'allow_merge_commit': self._config.repo.allow_merge_commit,
            'allow_rebase_merge': self._config.repo.allow_rebase_merge,
            'auto_init': self._config.repo.auto_init,
            'delete_branch_on_merge': self._config.repo.delete_branch_on_merge
        }
        if self._org_type == GithubOrgType.org:
            self._github.repos.create_in_org(self._config.organisation, **kwargs)
        else:
            self._github.repos.create_for_authenticated_user(**kwargs)

    def get_remote_url(self, repo_name: str) -> HttpUrl:
        """Get the remote url for the repo."""
        if self.check_exists(repo_name):
            return self._github.repos.get(self._config.organisation, repo_name)['html_url']

    def get_clone_url(self, repo_name: str) -> HttpUrl:
        """Get the clone url for the repo."""
        if self.check_exists(repo_name):
            return self._github.repos.get(self._config.organisation, repo_name)['clone_url']

    def delete(self, repo_name: str):
        """Delete the repo if it exists in the organisation/user."""
        if self.check_exists(repo_name):
            return self._github.repos.delete(self._config.organisation, repo_name)

    def check_exists(self, repo_name: str) -> bool:
        """Check if the repo exists in the organisation/user."""
        try:
            self._github.repos.get(self._config.organisation, repo_name)
            return True
        except HTTP404NotFoundError:
            return False

    def list(self) -> List[str]:
        """List the repos in the project."""
        repos = []
        if self._org_type == GithubOrgType.org:
            get_method = self._github.repos.list_for_org
        else:
            get_method = self._github.repos.list_for_user
        for u in paged(get_method, self._config.organisation, per_page=100):
            repos += [x['name'] for x in u]
        return repos
__init__(config)

Initialise the client.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/github.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def __init__(self, config: GithubSettings):
    """Initialise the client."""
    self._config = config
    self._github = GhApi(token=self._config.token.get_secret_value(), gh_host=str(self._config.url).rstrip('/'))
    # If the organisation is set, we get it, and assume that the token is valid
    # Otherwise default to the user
    if self._config.organisation:
        try:
            self._github.orgs.get(self._config.organisation)
            self._org_type = GithubOrgType.org
        except HTTP404NotFoundError:
            self._github.users.get_by_username(self._config.organisation)
            self._org_type = GithubOrgType.user
    else:
        self._config.organisation = self._github.users.get_authenticated()['login']
        self._org_type = GithubOrgType.user
check_exists(repo_name)

Check if the repo exists in the organisation/user.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/github.py 
123
124
125
126
127
128
129
def check_exists(self, repo_name: str) -> bool:
    """Check if the repo exists in the organisation/user."""
    try:
        self._github.repos.get(self._config.organisation, repo_name)
        return True
    except HTTP404NotFoundError:
        return False
create(repo_name)

Create the repo in the user/organisation.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/github.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def create(self, repo_name: str):
    """Create the repo in the user/organisation."""
    kwargs = {
        'name': repo_name,
        'private': self._config.repo.private,
        'has_issues': self._config.repo.has_issues,
        'has_wiki': self._config.repo.has_wiki,
        'has_downloads': self._config.repo.has_downloads,
        'has_projects': self._config.repo.has_projects,
        'allow_squash_merge': self._config.repo.allow_squash_merge,
        'allow_merge_commit': self._config.repo.allow_merge_commit,
        'allow_rebase_merge': self._config.repo.allow_rebase_merge,
        'auto_init': self._config.repo.auto_init,
        'delete_branch_on_merge': self._config.repo.delete_branch_on_merge
    }
    if self._org_type == GithubOrgType.org:
        self._github.repos.create_in_org(self._config.organisation, **kwargs)
    else:
        self._github.repos.create_for_authenticated_user(**kwargs)
delete(repo_name)

Delete the repo if it exists in the organisation/user.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/github.py 
118
119
120
121
def delete(self, repo_name: str):
    """Delete the repo if it exists in the organisation/user."""
    if self.check_exists(repo_name):
        return self._github.repos.delete(self._config.organisation, repo_name)
get_clone_url(repo_name)

Get the clone url for the repo.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/github.py 
113
114
115
116
def get_clone_url(self, repo_name: str) -> HttpUrl:
    """Get the clone url for the repo."""
    if self.check_exists(repo_name):
        return self._github.repos.get(self._config.organisation, repo_name)['clone_url']
get_remote_url(repo_name)

Get the remote url for the repo.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/github.py 
108
109
110
111
def get_remote_url(self, repo_name: str) -> HttpUrl:
    """Get the remote url for the repo."""
    if self.check_exists(repo_name):
        return self._github.repos.get(self._config.organisation, repo_name)['html_url']
list()

List the repos in the project.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/github.py 
131
132
133
134
135
136
137
138
139
140
def list(self) -> List[str]:
    """List the repos in the project."""
    repos = []
    if self._org_type == GithubOrgType.org:
        get_method = self._github.repos.list_for_org
    else:
        get_method = self._github.repos.list_for_user
    for u in paged(get_method, self._config.organisation, per_page=100):
        repos += [x['name'] for x in u]
    return repos

GithubRepoSettings

Bases: BaseConfiguration

Github Repo settings.

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/github.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class GithubRepoSettings(BaseConfiguration):
    """Github Repo settings."""
    model_config = SettingsConfigDict(env_prefix='GITHUB_REPO_', env_file='.env', dotenv_extra='ignore')

    private: bool = True
    has_issues: Optional[bool] = None
    has_wiki: Optional[bool] = None
    has_downloads: Optional[bool] = None
    has_projects: Optional[bool] = None
    allow_squash_merge: Optional[bool] = None
    allow_merge_commit: Optional[bool] = None
    allow_rebase_merge: Optional[bool] = None
    delete_branch_on_merge: Optional[bool] = None
    auto_init: bool = False

GithubSettings

Bases: VCSProviderSettings

Github settings.

Uses PAT token for auth (set in environment variables as GITHUB_TOKEN)

Source code in .venv-docs/lib/python3.12/site-packages/nskit/vcs/providers/github.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class GithubSettings(VCSProviderSettings):
    """Github settings.

    Uses PAT token for auth (set in environment variables as GITHUB_TOKEN)
    """
    model_config = SettingsConfigDict(env_prefix='GITHUB_', env_file='.env', dotenv_extra='ignore')

    interactive: bool = Field(False, description='Use Interactive Validation for token')
    url: HttpUrl = "https://api.github.com"
    organisation: Optional[str] = Field(None, description='Organisation to work in, otherwise uses the user for the token')
    token: SecretStr = Field(None, validate_default=True, description='Token to use for authentication, falls back to interactive device authentication if not provided')
    repo: GithubRepoSettings = Field(default_factory=GithubRepoSettings)

    @property
    def repo_client(self) -> 'GithubRepoClient':
        """Get the instantiated repo client."""
        return GithubRepoClient(self)

    @field_validator('token', mode='before')
    @classmethod
    def _validate_token(cls, value, info: ValidationInfo):
        if value is None and info.data.get('interactive', False):
            ghauth = GhDeviceAuth(_def_clientid, Scope.repo, Scope.delete_repo)
            print(ghauth.url_docs())
            ghauth.open_browser()
            value = ghauth.wait()
        return value
repo_client: GithubRepoClient property

Get the instantiated repo client.