Skip to content

Github

zenml.integrations.github

Initialization of the GitHub ZenML integration.

Attributes

GITHUB = 'github' module-attribute

GITHUB_EVENT_FLAVOR = 'github' module-attribute

Classes

BasePluginFlavor

Bases: ABC

Base Class for all PluginFlavors.

Functions
get_flavor_response_model(hydrate: bool) -> BasePluginFlavorResponse[Any, Any, Any] abstractmethod classmethod

Convert the Flavor into a Response Model.

Parameters:

Name Type Description Default
hydrate bool

Whether the model should be hydrated.

 required
Source code in src/zenml/plugins/base_plugin_flavor.py 
79
80
81
82
83
84
85
86
87
88
@classmethod
@abstractmethod
def get_flavor_response_model(
    cls, hydrate: bool
) -> BasePluginFlavorResponse[Any, Any, Any]:
    """Convert the Flavor into a Response Model.

    Args:
        hydrate: Whether the model should be hydrated.
    """

GitHubIntegration

Bases: Integration

Definition of GitHub integration for ZenML.

Functions
plugin_flavors() -> List[Type[BasePluginFlavor]] classmethod

Declare the event flavors for the github integration.

Returns:

Type Description
List[Type[BasePluginFlavor]]

List of stack component flavors for this integration.
Source code in src/zenml/integrations/github/__init__.py 
31
32
33
34
35
36
37
38
39
40
@classmethod
def plugin_flavors(cls) -> List[Type[BasePluginFlavor]]:
    """Declare the event flavors for the github integration.

    Returns:
        List of stack component flavors for this integration.
    """
    from zenml.integrations.github.plugins import GithubWebhookEventSourceFlavor

    return [GithubWebhookEventSourceFlavor]

Integration

Base class for integration in ZenML.

Functions
activate() -> None classmethod

Abstract method to activate the integration.

Source code in src/zenml/integrations/integration.py 
175
176
177
@classmethod
def activate(cls) -> None:
    """Abstract method to activate the integration."""
check_installation() -> bool classmethod

Method to check whether the required packages are installed.

Returns:

Type Description
bool

True if all required packages are installed, False otherwise.
Source code in src/zenml/integrations/integration.py 
 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
@classmethod
def check_installation(cls) -> bool:
    """Method to check whether the required packages are installed.

    Returns:
        True if all required packages are installed, False otherwise.
    """
    for r in cls.get_requirements():
        try:
            # First check if the base package is installed
            dist = pkg_resources.get_distribution(r)

            # Next, check if the dependencies (including extras) are
            # installed
            deps: List[Requirement] = []

            _, extras = parse_requirement(r)
            if extras:
                extra_list = extras[1:-1].split(",")
                for extra in extra_list:
                    try:
                        requirements = dist.requires(extras=[extra])  # type: ignore[arg-type]
                    except pkg_resources.UnknownExtra as e:
                        logger.debug(f"Unknown extra: {str(e)}")
                        return False
                    deps.extend(requirements)
            else:
                deps = dist.requires()

            for ri in deps:
                try:
                    # Remove the "extra == ..." part from the requirement string
                    cleaned_req = re.sub(
                        r"; extra == \"\w+\"", "", str(ri)
                    )
                    pkg_resources.get_distribution(cleaned_req)
                except pkg_resources.DistributionNotFound as e:
                    logger.debug(
                        f"Unable to find required dependency "
                        f"'{e.req}' for requirement '{r}' "
                        f"necessary for integration '{cls.NAME}'."
                    )
                    return False
                except pkg_resources.VersionConflict as e:
                    logger.debug(
                        f"Package version '{e.dist}' does not match "
                        f"version '{e.req}' required by '{r}' "
                        f"necessary for integration '{cls.NAME}'."
                    )
                    return False

        except pkg_resources.DistributionNotFound as e:
            logger.debug(
                f"Unable to find required package '{e.req}' for "
                f"integration {cls.NAME}."
            )
            return False
        except pkg_resources.VersionConflict as e:
            logger.debug(
                f"Package version '{e.dist}' does not match version "
                f"'{e.req}' necessary for integration {cls.NAME}."
            )
            return False

    logger.debug(
        f"Integration {cls.NAME} is installed correctly with "
        f"requirements {cls.get_requirements()}."
    )
    return True
flavors() -> List[Type[Flavor]] classmethod

Abstract method to declare new stack component flavors.

Returns:

Type Description
List[Type[Flavor]]

A list of new stack component flavors.
Source code in src/zenml/integrations/integration.py 
179
180
181
182
183
184
185
186
@classmethod
def flavors(cls) -> List[Type[Flavor]]:
    """Abstract method to declare new stack component flavors.

    Returns:
        A list of new stack component flavors.
    """
    return []
get_requirements(target_os: Optional[str] = None, python_version: Optional[str] = None) -> List[str] classmethod

Method to get the requirements for the integration.

Parameters:

Name Type Description Default
target_os Optional[str]

The target operating system to get the requirements for.

 None
python_version Optional[str]

The Python version to use for the requirements.

 None

Returns:

Type Description
List[str]

A list of requirements.
Source code in src/zenml/integrations/integration.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
@classmethod
def get_requirements(
    cls,
    target_os: Optional[str] = None,
    python_version: Optional[str] = None,
) -> List[str]:
    """Method to get the requirements for the integration.

    Args:
        target_os: The target operating system to get the requirements for.
        python_version: The Python version to use for the requirements.

    Returns:
        A list of requirements.
    """
    return cls.REQUIREMENTS
get_uninstall_requirements(target_os: Optional[str] = None) -> List[str] classmethod

Method to get the uninstall requirements for the integration.

Parameters:

Name Type Description Default
target_os Optional[str]

The target operating system to get the requirements for.

 None

Returns:

Type Description
List[str]

A list of requirements.
Source code in src/zenml/integrations/integration.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
@classmethod
def get_uninstall_requirements(
    cls, target_os: Optional[str] = None
) -> List[str]:
    """Method to get the uninstall requirements for the integration.

    Args:
        target_os: The target operating system to get the requirements for.

    Returns:
        A list of requirements.
    """
    ret = []
    for each in cls.get_requirements(target_os=target_os):
        is_ignored = False
        for ignored in cls.REQUIREMENTS_IGNORED_ON_UNINSTALL:
            if each.startswith(ignored):
                is_ignored = True
                break
        if not is_ignored:
            ret.append(each)
    return ret
plugin_flavors() -> List[Type[BasePluginFlavor]] classmethod

Abstract method to declare new plugin flavors.

Returns:

Type Description
List[Type[BasePluginFlavor]]

A list of new plugin flavors.
Source code in src/zenml/integrations/integration.py 
188
189
190
191
192
193
194
195
@classmethod
def plugin_flavors(cls) -> List[Type["BasePluginFlavor"]]:
    """Abstract method to declare new plugin flavors.

    Returns:
        A list of new plugin flavors.
    """
    return []

Modules

code_repositories

Initialization of the ZenML GitHub code repository.

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

Bases: BaseCodeRepository

GitHub code repository.

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: GitHubCodeRepositoryConfig property

Returns the GitHubCodeRepositoryConfig config.

Returns:

Type Description
GitHubCodeRepositoryConfig

The configuration.
github_repo: Repository property

The GitHub repository object from the GitHub API.

Returns:

Type Description
Repository

The GitHub repository.

Raises:

Type Description
RuntimeError

If the repository cannot be found.
Functions
check_github_repo_public(owner: str, repo: str) -> None

Checks if a GitHub repository is public.

Parameters:

Name Type Description Default
owner str

The owner of the repository.

 required
repo str

The name of the repository.

 required

Raises:

Type Description
RuntimeError

If the repository is not public.
Source code in src/zenml/integrations/github/code_repositories/github_code_repository.py 
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 check_github_repo_public(self, owner: str, repo: str) -> None:
    """Checks if a GitHub repository is public.

    Args:
        owner: The owner of the repository.
        repo: The name of the repository.

    Raises:
        RuntimeError: If the repository is not public.
    """
    url = f"{Consts.DEFAULT_BASE_URL}/repos/{owner}/{repo}"
    response = requests.get(url, timeout=7)

    try:
        if response.status_code == 200:
            pass
        else:
            raise RuntimeError(
                "It is not possible to access this repository as it does "
                "not appear to be public. Access to private repositories "
                "is only possible when a token is provided. Please provide "
                "a token and try again"
            )
    except Exception as e:
        raise RuntimeError(
            "An error occurred while checking if repository is public: "
            f"{str(e)}"
        )
check_remote_url(url: str) -> bool

Checks whether the remote url matches the code repository.

Parameters:

Name Type Description Default
url str

The remote url.

 required

Returns:

Type Description
bool

Whether the remote url is correct.
Source code in src/zenml/integrations/github/code_repositories/github_code_repository.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def check_remote_url(self, url: str) -> bool:
    """Checks whether the remote url matches the code repository.

    Args:
        url: The remote url.

    Returns:
        Whether the remote url is correct.
    """
    host = self.config.host or "github.com"
    host = host.rstrip("/")

    https_url = (
        f"https://{host}/{self.config.owner}/{self.config.repository}.git"
    )
    if url == https_url:
        return True

    ssh_regex = re.compile(
        f".*@{host}:{self.config.owner}/{self.config.repository}.git"
    )
    if ssh_regex.fullmatch(url):
        return True

    return False
download_files(commit: str, directory: str, repo_sub_directory: Optional[str]) -> None

Downloads files from a commit to a local directory.

Parameters:

Name Type Description Default
commit str

The commit to download.

 required
directory str

The directory to download to.

 required
repo_sub_directory Optional[str]

The sub directory to download from.

 required

Raises:

Type Description
RuntimeError

If the repository sub directory is invalid.
Source code in src/zenml/integrations/github/code_repositories/github_code_repository.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def download_files(
    self, commit: str, directory: str, repo_sub_directory: Optional[str]
) -> None:
    """Downloads files from a commit to a local directory.

    Args:
        commit: The commit to download.
        directory: The directory to download to.
        repo_sub_directory: The sub directory to download from.

    Raises:
        RuntimeError: If the repository sub directory is invalid.
    """
    contents = self.github_repo.get_contents(
        repo_sub_directory or "", ref=commit
    )
    if not isinstance(contents, List):
        raise RuntimeError("Invalid repository subdirectory.")

    os.makedirs(directory, exist_ok=True)

    for content in contents:
        local_path = os.path.join(directory, content.name)
        if content.type == "dir":
            self.download_files(
                commit=commit,
                directory=local_path,
                repo_sub_directory=content.path,
            )
        # For symlinks, content.type is initially wrongly set to "file",
        # which is why we need to read it from the raw data instead.
        elif content.raw_data["type"] == "symlink":
            try:
                os.symlink(src=content.raw_data["target"], dst=local_path)
            except Exception as e:
                logger.error(
                    "Failed to create symlink `%s` (%s): %s",
                    content.path,
                    content.html_url,
                    e,
                )
        else:
            try:
                with open(local_path, "wb") as f:
                    f.write(content.decoded_content)
            except (GithubException, IOError, AssertionError) as e:
                logger.error(
                    "Error processing `%s` (%s): %s",
                    content.path,
                    content.html_url,
                    e,
                )
get_local_context(path: str) -> Optional[LocalRepositoryContext]

Gets the local repository context.

Parameters:

Name Type Description Default
path str

The path to the local repository.

 required

Returns:

Type Description
Optional[LocalRepositoryContext]

The local repository context.
Source code in src/zenml/integrations/github/code_repositories/github_code_repository.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def get_local_context(self, path: str) -> Optional[LocalRepositoryContext]:
    """Gets the local repository context.

    Args:
        path: The path to the local repository.

    Returns:
        The local repository context.
    """
    return LocalGitRepositoryContext.at(
        path=path,
        code_repository=self,
        remote_url_validation_callback=self.check_remote_url,
    )
login() -> None

Logs in to GitHub using the token provided in the config.

Raises:

Type Description
RuntimeError

If the login fails.
Source code in src/zenml/integrations/github/code_repositories/github_code_repository.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def login(
    self,
) -> None:
    """Logs in to GitHub using the token provided in the config.

    Raises:
        RuntimeError: If the login fails.
    """
    try:
        self._github_session = Github(
            base_url=self.config.api_url or Consts.DEFAULT_BASE_URL,
            auth=Token(self.config.token) if self.config.token else None,
        )

        if self.config.token:
            user = self._github_session.get_user().login
            logger.debug(f"Logged in as {user}")
        else:
            self.check_github_repo_public(
                self.config.owner, self.config.repository
            )
    except Exception as e:
        raise RuntimeError(f"An error occurred while logging in: {str(e)}")
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/integrations/github/code_repositories/github_code_repository.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
@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.
    """
    code_repo = cls(id=uuid4(), name="", config=config)
    # Try to access the project to make sure it exists
    _ = code_repo.github_repo
Modules
github_code_repository

GitHub code repository.

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

Bases: BaseCodeRepository

GitHub code repository.

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: GitHubCodeRepositoryConfig property

Returns the GitHubCodeRepositoryConfig config.

Returns:

Type Description
GitHubCodeRepositoryConfig

The configuration.
github_repo: Repository property

The GitHub repository object from the GitHub API.

Returns:

Type Description
Repository

The GitHub repository.

Raises:

Type Description
RuntimeError

If the repository cannot be found.
Functions
check_github_repo_public(owner: str, repo: str) -> None

Checks if a GitHub repository is public.

Parameters:

Name Type Description Default
owner str

The owner of the repository.

 required
repo str

The name of the repository.

 required

Raises:

Type Description
RuntimeError

If the repository is not public.
Source code in src/zenml/integrations/github/code_repositories/github_code_repository.py 
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 check_github_repo_public(self, owner: str, repo: str) -> None:
    """Checks if a GitHub repository is public.

    Args:
        owner: The owner of the repository.
        repo: The name of the repository.

    Raises:
        RuntimeError: If the repository is not public.
    """
    url = f"{Consts.DEFAULT_BASE_URL}/repos/{owner}/{repo}"
    response = requests.get(url, timeout=7)

    try:
        if response.status_code == 200:
            pass
        else:
            raise RuntimeError(
                "It is not possible to access this repository as it does "
                "not appear to be public. Access to private repositories "
                "is only possible when a token is provided. Please provide "
                "a token and try again"
            )
    except Exception as e:
        raise RuntimeError(
            "An error occurred while checking if repository is public: "
            f"{str(e)}"
        )
check_remote_url(url: str) -> bool

Checks whether the remote url matches the code repository.

Parameters:

Name Type Description Default
url str

The remote url.

 required

Returns:

Type Description
bool

Whether the remote url is correct.
Source code in src/zenml/integrations/github/code_repositories/github_code_repository.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def check_remote_url(self, url: str) -> bool:
    """Checks whether the remote url matches the code repository.

    Args:
        url: The remote url.

    Returns:
        Whether the remote url is correct.
    """
    host = self.config.host or "github.com"
    host = host.rstrip("/")

    https_url = (
        f"https://{host}/{self.config.owner}/{self.config.repository}.git"
    )
    if url == https_url:
        return True

    ssh_regex = re.compile(
        f".*@{host}:{self.config.owner}/{self.config.repository}.git"
    )
    if ssh_regex.fullmatch(url):
        return True

    return False
download_files(commit: str, directory: str, repo_sub_directory: Optional[str]) -> None

Downloads files from a commit to a local directory.

Parameters:

Name Type Description Default
commit str

The commit to download.

 required
directory str

The directory to download to.

 required
repo_sub_directory Optional[str]

The sub directory to download from.

 required

Raises:

Type Description
RuntimeError

If the repository sub directory is invalid.
Source code in src/zenml/integrations/github/code_repositories/github_code_repository.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def download_files(
    self, commit: str, directory: str, repo_sub_directory: Optional[str]
) -> None:
    """Downloads files from a commit to a local directory.

    Args:
        commit: The commit to download.
        directory: The directory to download to.
        repo_sub_directory: The sub directory to download from.

    Raises:
        RuntimeError: If the repository sub directory is invalid.
    """
    contents = self.github_repo.get_contents(
        repo_sub_directory or "", ref=commit
    )
    if not isinstance(contents, List):
        raise RuntimeError("Invalid repository subdirectory.")

    os.makedirs(directory, exist_ok=True)

    for content in contents:
        local_path = os.path.join(directory, content.name)
        if content.type == "dir":
            self.download_files(
                commit=commit,
                directory=local_path,
                repo_sub_directory=content.path,
            )
        # For symlinks, content.type is initially wrongly set to "file",
        # which is why we need to read it from the raw data instead.
        elif content.raw_data["type"] == "symlink":
            try:
                os.symlink(src=content.raw_data["target"], dst=local_path)
            except Exception as e:
                logger.error(
                    "Failed to create symlink `%s` (%s): %s",
                    content.path,
                    content.html_url,
                    e,
                )
        else:
            try:
                with open(local_path, "wb") as f:
                    f.write(content.decoded_content)
            except (GithubException, IOError, AssertionError) as e:
                logger.error(
                    "Error processing `%s` (%s): %s",
                    content.path,
                    content.html_url,
                    e,
                )
get_local_context(path: str) -> Optional[LocalRepositoryContext]

Gets the local repository context.

Parameters:

Name Type Description Default
path str

The path to the local repository.

 required

Returns:

Type Description
Optional[LocalRepositoryContext]

The local repository context.
Source code in src/zenml/integrations/github/code_repositories/github_code_repository.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def get_local_context(self, path: str) -> Optional[LocalRepositoryContext]:
    """Gets the local repository context.

    Args:
        path: The path to the local repository.

    Returns:
        The local repository context.
    """
    return LocalGitRepositoryContext.at(
        path=path,
        code_repository=self,
        remote_url_validation_callback=self.check_remote_url,
    )
login() -> None

Logs in to GitHub using the token provided in the config.

Raises:

Type Description
RuntimeError

If the login fails.
Source code in src/zenml/integrations/github/code_repositories/github_code_repository.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def login(
    self,
) -> None:
    """Logs in to GitHub using the token provided in the config.

    Raises:
        RuntimeError: If the login fails.
    """
    try:
        self._github_session = Github(
            base_url=self.config.api_url or Consts.DEFAULT_BASE_URL,
            auth=Token(self.config.token) if self.config.token else None,
        )

        if self.config.token:
            user = self._github_session.get_user().login
            logger.debug(f"Logged in as {user}")
        else:
            self.check_github_repo_public(
                self.config.owner, self.config.repository
            )
    except Exception as e:
        raise RuntimeError(f"An error occurred while logging in: {str(e)}")
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/integrations/github/code_repositories/github_code_repository.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
@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.
    """
    code_repo = cls(id=uuid4(), name="", config=config)
    # Try to access the project to make sure it exists
    _ = code_repo.github_repo
GitHubCodeRepositoryConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: BaseCodeRepositoryConfig

Config for GitHub code repositories.

Parameters:

Name Type Description Default
api_url

The GitHub API URL.

 required
owner

The owner of the repository.

 required
repository

The name of the repository.

 required
host

The host of the repository.

 required
token

The token to access the repository.

 required
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

plugins

Github event flavors.

Classes
GithubWebhookEventSourceFlavor

Bases: BaseWebhookEventSourceFlavor

Enables users to configure github event sources.

Modules
event_sources
Modules
github_webhook_event_source

Implementation of the github webhook event source.

Classes
Commit

Bases: BaseModel

Github Event.

GithubEvent

Bases: BaseEvent

Push Event from Github.

Attributes
branch: Optional[str] property

The branch the event happened on.

Returns:

Type Description
Optional[str]

The branch name.
event_type: Union[GithubEventType, str] property

The type of github event.

Returns:

Type Description
Union[GithubEventType, str]

The type of the event.
GithubEventType

Bases: StrEnum

Collection of all possible Github Events.

GithubWebhookEventFilterConfiguration

Bases: WebhookEventFilterConfig

Configuration for github event filters.

Functions
event_matches_filter(event: BaseEvent) -> bool

Checks the filter against the inbound event.

Parameters:

Name Type Description Default
event BaseEvent

The incoming event

 required

Returns:

Type Description
bool

Whether the event matches the filter
Source code in src/zenml/integrations/github/plugins/event_sources/github_webhook_event_source.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def event_matches_filter(self, event: BaseEvent) -> bool:
    """Checks the filter against the inbound event.

    Args:
        event: The incoming event

    Returns:
        Whether the event matches the filter
    """
    if not isinstance(event, GithubEvent):
        return False
    if self.event_type and event.event_type != self.event_type:
        # Mismatch for the action
        return False
    if self.repo and event.repository.full_name != self.repo:
        # Mismatch for the repository
        return False
    if self.branch and event.branch != self.branch:
        # Mismatch for the branch
        return False
    return True
GithubWebhookEventSourceConfiguration

Bases: WebhookEventSourceConfig

Configuration for github source filters.

GithubWebhookEventSourceHandler(event_hub: Optional[BaseEventHub] = None)

Bases: BaseWebhookEventSourceHandler

Handler for all github events.

Source code in src/zenml/event_sources/base_event_source.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def __init__(self, event_hub: Optional[BaseEventHub] = None) -> None:
    """Event source handler initialization.

    Args:
        event_hub: Optional event hub to use to initialize the event source
            handler. If not set during initialization, it may be set
            at a later time by calling `set_event_hub`. An event hub must
            be configured before the event handler needs to dispatch events.
    """
    super().__init__()
    if event_hub is None:
        from zenml.event_hub.event_hub import (
            event_hub as default_event_hub,
        )

        # TODO: for now, we use the default internal event hub. In
        # the future, this should be configurable.
        event_hub = default_event_hub

    self.set_event_hub(event_hub)
Attributes
config_class: Type[GithubWebhookEventSourceConfiguration] property

Returns the webhook event source configuration class.

Returns:

Type Description
Type[GithubWebhookEventSourceConfiguration]

The configuration.
filter_class: Type[GithubWebhookEventFilterConfiguration] property

Returns the webhook event filter configuration class.

Returns:

Type Description
Type[GithubWebhookEventFilterConfiguration]

The event filter configuration class.
flavor_class: Type[BaseWebhookEventSourceFlavor] property

Returns the flavor class of the plugin.

Returns:

Type Description
Type[BaseWebhookEventSourceFlavor]

The flavor class of the plugin.
PullRequest

Bases: BaseModel

Github Pull Request.

Repository

Bases: BaseModel

Github Repository.

Tag

Bases: BaseModel

Github Tag.

User

Bases: BaseModel

Github User.

Functions
github_webhook_event_source_flavor

Github webhook event source flavor.

Classes
GithubWebhookEventSourceFlavor

Bases: BaseWebhookEventSourceFlavor

Enables users to configure github event sources.