Skip to content

Code Repositories

zenml.code_repositories

Initialization of the ZenML code repository base abstraction.

Attributes

__all__ = ['BaseCodeRepository', 'LocalRepositoryContext'] module-attribute

Classes

BaseCodeRepository(id: UUID, name: str, config: Dict[str, Any])

Bases: ABC

Base class for code repositories.

Code repositories are used to connect to a remote code repository and store information about the repository, such as the URL, the owner, the repository name, and the host. They also provide methods to download files from the repository when a pipeline is run remotely.

Initializes a code repository.

Parameters:

Name Type Description Default
id UUID

The ID of the code repository.

 required
name str

The name of the code repository.

 required
config Dict[str, Any]

The config of the code repository.

 required
Source code in src/zenml/code_repositories/base_code_repository.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(
    self,
    id: UUID,
    name: str,
    config: Dict[str, Any],
) -> None:
    """Initializes a code repository.

    Args:
        id: The ID of the code repository.
        name: The name of the code repository.
        config: The config of the code repository.
    """
    self._id = id
    self._name = name
    self._config = config
    self.login()
Attributes
config: BaseCodeRepositoryConfig property

Config class for Code Repository.

Returns:

Type Description
BaseCodeRepositoryConfig

The config class.
id: UUID property

ID of the code repository.

Returns:

Type Description
UUID

The ID of the code repository.
name: str property

Name of the code repository.

Returns:

Type Description
str

The name of the code repository.
requirements: Set[str] property

Set of PyPI requirements for the repository.

Returns:

Type Description
Set[str]

A set of PyPI requirements for the repository.
Functions
download_files(commit: str, directory: str, repo_sub_directory: Optional[str]) -> None abstractmethod

Downloads files from the code repository to a local directory.

Parameters:

Name Type Description Default
commit str

The commit hash to download files from.

 required
directory str

The directory to download files to.

 required
repo_sub_directory Optional[str]

The subdirectory in the repository to download files from.

 required

Raises:

Type Description
RuntimeError

If the download fails.
Source code in src/zenml/code_repositories/base_code_repository.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@abstractmethod
def download_files(
    self, commit: str, directory: str, repo_sub_directory: Optional[str]
) -> None:
    """Downloads files from the code repository to a local directory.

    Args:
        commit: The commit hash to download files from.
        directory: The directory to download files to.
        repo_sub_directory: The subdirectory in the repository to
            download files from.

    Raises:
        RuntimeError: If the download fails.
    """
    pass
from_model(model: CodeRepositoryResponse) -> BaseCodeRepository classmethod

Loads a code repository from a model.

Parameters:

Name Type Description Default
model CodeRepositoryResponse

The CodeRepositoryResponseModel to load from.

 required

Returns:

Type Description
BaseCodeRepository

The loaded code repository object.
Source code in src/zenml/code_repositories/base_code_repository.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@classmethod
def from_model(cls, model: CodeRepositoryResponse) -> "BaseCodeRepository":
    """Loads a code repository from a model.

    Args:
        model: The CodeRepositoryResponseModel to load from.

    Returns:
        The loaded code repository object.
    """
    class_: Type[BaseCodeRepository] = (
        source_utils.load_and_validate_class(
            source=model.source, expected_class=BaseCodeRepository
        )
    )
    return class_(id=model.id, name=model.name, config=model.config)
get_local_context(path: str) -> Optional[LocalRepositoryContext] abstractmethod

Gets a local repository context from a path.

Parameters:

Name Type Description Default
path str

The path to the local repository.

 required

Returns:

Type Description
Optional[LocalRepositoryContext]

The local repository context object.
Source code in src/zenml/code_repositories/base_code_repository.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
@abstractmethod
def get_local_context(
    self, path: str
) -> Optional["LocalRepositoryContext"]:
    """Gets a local repository context from a path.

    Args:
        path: The path to the local repository.

    Returns:
        The local repository context object.
    """
    pass
login() -> None abstractmethod

Logs into the code repository.

This method is called when the code repository is initialized. It should be used to authenticate with the code repository.

Raises:

Type Description
RuntimeError

If the login fails.
Source code in src/zenml/code_repositories/base_code_repository.py 
133
134
135
136
137
138
139
140
141
142
143
@abstractmethod
def login(self) -> None:
    """Logs into the code repository.

    This method is called when the code repository is initialized.
    It should be used to authenticate with the code repository.

    Raises:
        RuntimeError: If the login fails.
    """
    pass
validate_config(config: Dict[str, Any]) -> None classmethod

Validate the code repository config.

This method should check that the config/credentials are valid and the configured repository exists.

Parameters:

Name Type Description Default
config Dict[str, Any]

The configuration.

 required
Source code in src/zenml/code_repositories/base_code_repository.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@classmethod
def validate_config(cls, config: Dict[str, Any]) -> None:
    """Validate the code repository config.

    This method should check that the config/credentials are valid and
    the configured repository exists.

    Args:
        config: The configuration.
    """
    # The initialization calls the login to verify the credentials
    code_repo = cls(id=uuid4(), name="", config=config)

    # Explicitly access the config for pydantic validation
    _ = code_repo.config

LocalRepositoryContext(code_repository: BaseCodeRepository)

Bases: ABC

Base class for local repository contexts.

This class is used to represent a local repository. It is used to track the current state of the repository and to provide information about the repository, such as the root path, the current commit, and whether the repository is dirty.

Initializes a local repository context.

Parameters:

Name Type Description Default
code_repository BaseCodeRepository

The code repository.

 required
Source code in src/zenml/code_repositories/local_repository_context.py 
36
37
38
39
40
41
42
def __init__(self, code_repository: "BaseCodeRepository") -> None:
    """Initializes a local repository context.

    Args:
        code_repository: The code repository.
    """
    self._code_repository = code_repository
Attributes
code_repository: BaseCodeRepository property

Returns the code repository.

Returns:

Type Description
BaseCodeRepository

The code repository.
current_commit: str abstractmethod property

Returns the current commit of the local repository.

Returns:

Type Description
str

The current commit of the local repository.
has_local_changes: bool abstractmethod property

Returns whether the local repository has local changes.

A repository has local changes if it is dirty or there are some commits which have not been pushed yet.

Returns:

Type Description
bool

Whether the local repository has local changes.
is_dirty: bool abstractmethod property

Returns whether the local repository is dirty.

A repository counts as dirty if it has any untracked or uncommitted changes.

Returns:

Type Description
bool

Whether the local repository is dirty.
root: str abstractmethod property

Returns the root path of the local repository.

Returns:

Type Description
str

The root path of the local repository.
Functions

Modules

base_code_repository

Base class for code repositories.

Classes
BaseCodeRepository(id: UUID, name: str, config: Dict[str, Any])

Bases: ABC

Base class for code repositories.

Code repositories are used to connect to a remote code repository and store information about the repository, such as the URL, the owner, the repository name, and the host. They also provide methods to download files from the repository when a pipeline is run remotely.

Initializes a code repository.

Parameters:

Name Type Description Default
id UUID

The ID of the code repository.

 required
name str

The name of the code repository.

 required
config Dict[str, Any]

The config of the code repository.

 required
Source code in src/zenml/code_repositories/base_code_repository.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(
    self,
    id: UUID,
    name: str,
    config: Dict[str, Any],
) -> None:
    """Initializes a code repository.

    Args:
        id: The ID of the code repository.
        name: The name of the code repository.
        config: The config of the code repository.
    """
    self._id = id
    self._name = name
    self._config = config
    self.login()
Attributes
config: BaseCodeRepositoryConfig property

Config class for Code Repository.

Returns:

Type Description
BaseCodeRepositoryConfig

The config class.
id: UUID property

ID of the code repository.

Returns:

Type Description
UUID

The ID of the code repository.
name: str property

Name of the code repository.

Returns:

Type Description
str

The name of the code repository.
requirements: Set[str] property

Set of PyPI requirements for the repository.

Returns:

Type Description
Set[str]

A set of PyPI requirements for the repository.
Functions
download_files(commit: str, directory: str, repo_sub_directory: Optional[str]) -> None abstractmethod

Downloads files from the code repository to a local directory.

Parameters:

Name Type Description Default
commit str

The commit hash to download files from.

 required
directory str

The directory to download files to.

 required
repo_sub_directory Optional[str]

The subdirectory in the repository to download files from.

 required

Raises:

Type Description
RuntimeError

If the download fails.
Source code in src/zenml/code_repositories/base_code_repository.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@abstractmethod
def download_files(
    self, commit: str, directory: str, repo_sub_directory: Optional[str]
) -> None:
    """Downloads files from the code repository to a local directory.

    Args:
        commit: The commit hash to download files from.
        directory: The directory to download files to.
        repo_sub_directory: The subdirectory in the repository to
            download files from.

    Raises:
        RuntimeError: If the download fails.
    """
    pass
from_model(model: CodeRepositoryResponse) -> BaseCodeRepository classmethod

Loads a code repository from a model.

Parameters:

Name Type Description Default
model CodeRepositoryResponse

The CodeRepositoryResponseModel to load from.

 required

Returns:

Type Description
BaseCodeRepository

The loaded code repository object.
Source code in src/zenml/code_repositories/base_code_repository.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@classmethod
def from_model(cls, model: CodeRepositoryResponse) -> "BaseCodeRepository":
    """Loads a code repository from a model.

    Args:
        model: The CodeRepositoryResponseModel to load from.

    Returns:
        The loaded code repository object.
    """
    class_: Type[BaseCodeRepository] = (
        source_utils.load_and_validate_class(
            source=model.source, expected_class=BaseCodeRepository
        )
    )
    return class_(id=model.id, name=model.name, config=model.config)
get_local_context(path: str) -> Optional[LocalRepositoryContext] abstractmethod

Gets a local repository context from a path.

Parameters:

Name Type Description Default
path str

The path to the local repository.

 required

Returns:

Type Description
Optional[LocalRepositoryContext]

The local repository context object.
Source code in src/zenml/code_repositories/base_code_repository.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
@abstractmethod
def get_local_context(
    self, path: str
) -> Optional["LocalRepositoryContext"]:
    """Gets a local repository context from a path.

    Args:
        path: The path to the local repository.

    Returns:
        The local repository context object.
    """
    pass
login() -> None abstractmethod

Logs into the code repository.

This method is called when the code repository is initialized. It should be used to authenticate with the code repository.

Raises:

Type Description
RuntimeError

If the login fails.
Source code in src/zenml/code_repositories/base_code_repository.py 
133
134
135
136
137
138
139
140
141
142
143
@abstractmethod
def login(self) -> None:
    """Logs into the code repository.

    This method is called when the code repository is initialized.
    It should be used to authenticate with the code repository.

    Raises:
        RuntimeError: If the login fails.
    """
    pass
validate_config(config: Dict[str, Any]) -> None classmethod

Validate the code repository config.

This method should check that the config/credentials are valid and the configured repository exists.

Parameters:

Name Type Description Default
config Dict[str, Any]

The configuration.

 required
Source code in src/zenml/code_repositories/base_code_repository.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@classmethod
def validate_config(cls, config: Dict[str, Any]) -> None:
    """Validate the code repository config.

    This method should check that the config/credentials are valid and
    the configured repository exists.

    Args:
        config: The configuration.
    """
    # The initialization calls the login to verify the credentials
    code_repo = cls(id=uuid4(), name="", config=config)

    # Explicitly access the config for pydantic validation
    _ = code_repo.config
BaseCodeRepositoryConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: SecretReferenceMixin, ABC

Base config for code repositories.

Source code in src/zenml/config/secret_reference_mixin.py 
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
def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references are only passed for valid fields.

    This method ensures that secret references are not passed for fields
    that explicitly prevent them or require pydantic validation.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using plain-text secrets.
        **kwargs: Arguments to initialize this object.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            or an attribute which explicitly disallows secret references
            is passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}`. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure values with secrets "
                    "here: https://docs.zenml.io/getting-started/deploying-zenml/secret-management"
                )
            continue

        if secret_utils.is_clear_text_field(field):
            raise ValueError(
                f"Passing the `{key}` attribute as a secret reference is "
                "not allowed."
            )

        requires_validation = has_validators(
            pydantic_class=self.__class__, field_name=key
        )
        if requires_validation:
            raise ValueError(
                f"Passing the attribute `{key}` as a secret reference is "
                "not allowed as additional validation is required for "
                "this attribute."
            )

    super().__init__(**kwargs)
Functions
Modules

git

Initialization of the local git repository context.

Classes
LocalGitRepositoryContext(code_repository: BaseCodeRepository, git_repo: Repo, remote_name: str)

Bases: LocalRepositoryContext

Local git repository context.

Initializes a local git repository context.

Parameters:

Name Type Description Default
code_repository BaseCodeRepository

The code repository.

 required
git_repo Repo

The git repo.

 required
remote_name str

Name of the remote.

 required
Source code in src/zenml/code_repositories/git/local_git_repository_context.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    code_repository: "BaseCodeRepository",
    git_repo: "Repo",
    remote_name: str,
):
    """Initializes a local git repository context.

    Args:
        code_repository: The code repository.
        git_repo: The git repo.
        remote_name: Name of the remote.
    """
    super().__init__(code_repository=code_repository)
    self._git_repo = git_repo
    self._remote = git_repo.remote(name=remote_name)
Attributes
current_commit: str property

The current commit.

Returns:

Type Description
str

The current commit sha.
git_repo: Repo property

The git repo.

Returns:

Type Description
Repo

The git repo object of the local git repository.
has_local_changes: bool property

Whether the git repo has local changes.

A repository has local changes if it is dirty or there are some commits which have not been pushed yet.

Returns:

Type Description
bool

True if the git repo has local changes, False otherwise.

Raises:

Type Description
RuntimeError

If the git repo is in a detached head state.
is_dirty: bool property

Whether the git repo is dirty.

By default, a repository counts as dirty if it has any untracked or uncommitted changes. Users can use an environment variable to ignore untracked files.

Returns:

Type Description
bool

True if the git repo is dirty, False otherwise.
remote: Remote property

The git remote.

Returns:

Type Description
Remote

The remote of the git repo object of the local git repository.
root: str property

The root of the git repo.

Returns:

Type Description
str

The root of the git repo.
Functions
at(path: str, code_repository: BaseCodeRepository, remote_url_validation_callback: Callable[[str], bool]) -> Optional[LocalGitRepositoryContext] classmethod

Returns a local git repository at the given path.

Parameters:

Name Type Description Default
path str

The path to the local git repository.

 required
code_repository BaseCodeRepository

The code repository.

 required
remote_url_validation_callback Callable[[str], bool]

A callback that validates the remote URL of the git repository.

 required

Returns:

Type Description
Optional[LocalGitRepositoryContext]

A local git repository if the path is a valid git repository
Optional[LocalGitRepositoryContext]

and the remote URL is valid, otherwise None.
Source code in src/zenml/code_repositories/git/local_git_repository_context.py 
 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
@classmethod
def at(
    cls,
    path: str,
    code_repository: "BaseCodeRepository",
    remote_url_validation_callback: Callable[[str], bool],
) -> Optional["LocalGitRepositoryContext"]:
    """Returns a local git repository at the given path.

    Args:
        path: The path to the local git repository.
        code_repository: The code repository.
        remote_url_validation_callback: A callback that validates the
            remote URL of the git repository.

    Returns:
        A local git repository if the path is a valid git repository
        and the remote URL is valid, otherwise None.
    """
    try:
        # These imports fail when git is not installed on the machine
        from git.exc import InvalidGitRepositoryError
        from git.repo.base import Repo
    except ImportError:
        logger.debug("Failed to import git library.")
        return None

    try:
        git_repo = Repo(path=path, search_parent_directories=True)
    except InvalidGitRepositoryError:
        logger.debug("No git repository exists at path %s.", path)
        return None

    remote_name = None
    for remote in git_repo.remotes:
        if remote_url_validation_callback(remote.url):
            remote_name = remote.name
            break

    if not remote_name:
        return None

    return LocalGitRepositoryContext(
        code_repository=code_repository,
        git_repo=git_repo,
        remote_name=remote_name,
    )
Modules
local_git_repository_context

Implementation of the Local git repository context.

Classes
LocalGitRepositoryContext(code_repository: BaseCodeRepository, git_repo: Repo, remote_name: str)

Bases: LocalRepositoryContext

Local git repository context.

Initializes a local git repository context.

Parameters:

Name Type Description Default
code_repository BaseCodeRepository

The code repository.

 required
git_repo Repo

The git repo.

 required
remote_name str

Name of the remote.

 required
Source code in src/zenml/code_repositories/git/local_git_repository_context.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    code_repository: "BaseCodeRepository",
    git_repo: "Repo",
    remote_name: str,
):
    """Initializes a local git repository context.

    Args:
        code_repository: The code repository.
        git_repo: The git repo.
        remote_name: Name of the remote.
    """
    super().__init__(code_repository=code_repository)
    self._git_repo = git_repo
    self._remote = git_repo.remote(name=remote_name)
Attributes
current_commit: str property

The current commit.

Returns:

Type Description
str

The current commit sha.
git_repo: Repo property

The git repo.

Returns:

Type Description
Repo

The git repo object of the local git repository.
has_local_changes: bool property

Whether the git repo has local changes.

A repository has local changes if it is dirty or there are some commits which have not been pushed yet.

Returns:

Type Description
bool

True if the git repo has local changes, False otherwise.

Raises:

Type Description
RuntimeError

If the git repo is in a detached head state.
is_dirty: bool property

Whether the git repo is dirty.

By default, a repository counts as dirty if it has any untracked or uncommitted changes. Users can use an environment variable to ignore untracked files.

Returns:

Type Description
bool

True if the git repo is dirty, False otherwise.
remote: Remote property

The git remote.

Returns:

Type Description
Remote

The remote of the git repo object of the local git repository.
root: str property

The root of the git repo.

Returns:

Type Description
str

The root of the git repo.
Functions
at(path: str, code_repository: BaseCodeRepository, remote_url_validation_callback: Callable[[str], bool]) -> Optional[LocalGitRepositoryContext] classmethod

Returns a local git repository at the given path.

Parameters:

Name Type Description Default
path str

The path to the local git repository.

 required
code_repository BaseCodeRepository

The code repository.

 required
remote_url_validation_callback Callable[[str], bool]

A callback that validates the remote URL of the git repository.

 required

Returns:

Type Description
Optional[LocalGitRepositoryContext]

A local git repository if the path is a valid git repository
Optional[LocalGitRepositoryContext]

and the remote URL is valid, otherwise None.
Source code in src/zenml/code_repositories/git/local_git_repository_context.py 
 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
@classmethod
def at(
    cls,
    path: str,
    code_repository: "BaseCodeRepository",
    remote_url_validation_callback: Callable[[str], bool],
) -> Optional["LocalGitRepositoryContext"]:
    """Returns a local git repository at the given path.

    Args:
        path: The path to the local git repository.
        code_repository: The code repository.
        remote_url_validation_callback: A callback that validates the
            remote URL of the git repository.

    Returns:
        A local git repository if the path is a valid git repository
        and the remote URL is valid, otherwise None.
    """
    try:
        # These imports fail when git is not installed on the machine
        from git.exc import InvalidGitRepositoryError
        from git.repo.base import Repo
    except ImportError:
        logger.debug("Failed to import git library.")
        return None

    try:
        git_repo = Repo(path=path, search_parent_directories=True)
    except InvalidGitRepositoryError:
        logger.debug("No git repository exists at path %s.", path)
        return None

    remote_name = None
    for remote in git_repo.remotes:
        if remote_url_validation_callback(remote.url):
            remote_name = remote.name
            break

    if not remote_name:
        return None

    return LocalGitRepositoryContext(
        code_repository=code_repository,
        git_repo=git_repo,
        remote_name=remote_name,
    )
Functions

local_repository_context

Base class for local code repository contexts.

Classes
LocalRepositoryContext(code_repository: BaseCodeRepository)

Bases: ABC

Base class for local repository contexts.

This class is used to represent a local repository. It is used to track the current state of the repository and to provide information about the repository, such as the root path, the current commit, and whether the repository is dirty.

Initializes a local repository context.

Parameters:

Name Type Description Default
code_repository BaseCodeRepository

The code repository.

 required
Source code in src/zenml/code_repositories/local_repository_context.py 
36
37
38
39
40
41
42
def __init__(self, code_repository: "BaseCodeRepository") -> None:
    """Initializes a local repository context.

    Args:
        code_repository: The code repository.
    """
    self._code_repository = code_repository
Attributes
code_repository: BaseCodeRepository property

Returns the code repository.

Returns:

Type Description
BaseCodeRepository

The code repository.
current_commit: str abstractmethod property

Returns the current commit of the local repository.

Returns:

Type Description
str

The current commit of the local repository.
has_local_changes: bool abstractmethod property

Returns whether the local repository has local changes.

A repository has local changes if it is dirty or there are some commits which have not been pushed yet.

Returns:

Type Description
bool

Whether the local repository has local changes.
is_dirty: bool abstractmethod property

Returns whether the local repository is dirty.

A repository counts as dirty if it has any untracked or uncommitted changes.

Returns:

Type Description
bool

Whether the local repository is dirty.
root: str abstractmethod property

Returns the root path of the local repository.

Returns:

Type Description
str

The root path of the local repository.
Functions
Functions