|
3 | 3 | # SPDX-License-Identifier: Apache-2.0 |
4 | 4 | from base64 import b64decode, b64encode |
5 | 5 | from enum import Enum |
6 | | -from typing import Any, Optional, Union |
| 6 | +from typing import Any |
7 | 7 |
|
8 | 8 | import requests |
9 | 9 | from haystack import component, default_from_dict, default_to_dict, logging |
@@ -79,7 +79,7 @@ def __init__( |
79 | 79 | self, |
80 | 80 | *, |
81 | 81 | github_token: Secret = Secret.from_env_var("GITHUB_TOKEN"), |
82 | | - repo: Optional[str] = None, |
| 82 | + repo: str | None = None, |
83 | 83 | branch: str = "main", |
84 | 84 | raise_on_failure: bool = True, |
85 | 85 | ): |
@@ -145,7 +145,7 @@ def _update_file(self, owner: str, repo: str, path: str, content: str, message: |
145 | 145 | def _check_last_commit(self, owner: str, repo: str, branch: str) -> bool: |
146 | 146 | """Check if last commit was made by the current token user.""" |
147 | 147 | url = f"https://api.github.com/repos/{owner}/{repo}/commits" |
148 | | - params: dict[str, Union[str, int]] = {"per_page": 1, "sha": branch} |
| 148 | + params: dict[str, str | int] = {"per_page": 1, "sha": branch} |
149 | 149 | response = requests.get(url, headers=self._get_request_headers(), params=params, timeout=10) |
150 | 150 | response.raise_for_status() |
151 | 151 | last_commit = response.json()[0] |
@@ -191,7 +191,7 @@ def _undo_changes(self, owner: str, repo: str, payload: dict[str, Any], branch: |
191 | 191 | commits_url = f"https://api.github.com/repos/{owner}/{repo}/commits" |
192 | 192 |
|
193 | 193 | # Get the previous commit SHA |
194 | | - params: dict[str, Union[str, int]] = {"per_page": 2, "sha": branch} |
| 194 | + params: dict[str, str | int] = {"per_page": 2, "sha": branch} |
195 | 195 | commits = requests.get(commits_url, headers=self._get_request_headers(), params=params, timeout=10).json() |
196 | 196 | previous_sha = commits[1]["sha"] |
197 | 197 |
|
@@ -244,10 +244,10 @@ def _delete_file(self, owner: str, repo: str, payload: dict[str, str], branch: s |
244 | 244 | @component.output_types(result=str) |
245 | 245 | def run( |
246 | 246 | self, |
247 | | - command: Union[Command, str], |
| 247 | + command: Command | str, |
248 | 248 | payload: dict[str, Any], |
249 | | - repo: Optional[str] = None, |
250 | | - branch: Optional[str] = None, |
| 249 | + repo: str | None = None, |
| 250 | + branch: str | None = None, |
251 | 251 | ) -> dict[str, str]: |
252 | 252 | """ |
253 | 253 | Process GitHub file operations. |
|
0 commit comments