Coverage for /usr/local/lib/python3.12/site-packages/prefect/_versioning.py: 28%
178 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:38 +0000
1from __future__ import annotations 1a
3import os 1a
4from enum import Enum 1a
5from typing import Any, Callable, Coroutine, Dict, Literal, Optional 1a
6from urllib.parse import urlparse 1a
8from anyio import run_process 1a
9from pydantic import Field 1a
11from prefect.client.schemas.objects import VersionInfo 1a
14async def get_commit_message_first_line() -> str: 1a
15 result = await run_process(["git", "log", "-1", "--pretty=%B"])
16 return result.stdout.decode().strip().splitlines()[0]
19class SimpleVersionInfo(VersionInfo): 1a
20 type: Literal["prefect:simple"] = "prefect:simple" 1a
21 version: str = Field(default="") 1a
22 branch: Optional[str] = Field(default=None) 1a
23 url: Optional[str] = Field(default=None) 1a
26class GithubVersionInfo(VersionInfo): 1a
27 type: Literal["vcs:github"] = "vcs:github" 1a
28 version: str 1a
29 commit_sha: str 1a
30 message: str 1a
31 branch: str 1a
32 repository: str 1a
33 url: str 1a
36class GitlabVersionInfo(VersionInfo): 1a
37 type: Literal["vcs:gitlab"] = "vcs:gitlab" 1a
38 version: str 1a
39 commit_sha: str 1a
40 message: str 1a
41 branch: str 1a
42 repository: str 1a
43 url: str 1a
46class BitbucketVersionInfo(VersionInfo): 1a
47 type: Literal["vcs:bitbucket"] = "vcs:bitbucket" 1a
48 version: str 1a
49 commit_sha: str 1a
50 message: str 1a
51 branch: str 1a
52 repository: str 1a
53 url: str 1a
56class AzureDevopsVersionInfo(VersionInfo): 1a
57 type: Literal["vcs:azuredevops"] = "vcs:azuredevops" 1a
58 version: str 1a
59 commit_sha: str 1a
60 message: str 1a
61 branch: str 1a
62 repository: str 1a
63 url: str 1a
66class GitVersionInfo(VersionInfo): 1a
67 type: Literal["vcs:git"] = "vcs:git" 1a
68 version: str 1a
69 commit_sha: str 1a
70 message: str 1a
71 branch: str 1a
72 repository: str 1a
73 url: str 1a
76async def get_github_version_info( 1a
77 commit_sha: Optional[str] = None,
78 message: Optional[str] = None,
79 branch: Optional[str] = None,
80 repository: Optional[str] = None,
81 url: Optional[str] = None,
82) -> GithubVersionInfo:
83 """Create a GithubVersionInfo object from provided values or environment variables.
85 Args:
86 commit_sha: The commit SHA, falls back to GITHUB_SHA env var
87 message: The commit message, falls back to git log -1 --pretty=%B
88 branch: The git branch, falls back to GITHUB_REF_NAME env var
89 repository: The repository name, falls back to GITHUB_REPOSITORY env var
90 url: The repository URL, constructed from GITHUB_SERVER_URL/GITHUB_REPOSITORY/tree/GITHUB_SHA if not provided
92 Returns:
93 A GithubVersionInfo
95 Raises:
96 ValueError: If any required fields cannot be determined
97 """
98 try:
99 commit_sha = commit_sha or os.getenv("GITHUB_SHA")
100 branch = branch or os.getenv("GITHUB_REF_NAME")
101 repository = repository or os.getenv("GITHUB_REPOSITORY")
102 url = url or f"{os.getenv('GITHUB_SERVER_URL')}/{repository}/tree/{commit_sha}"
104 if not message:
105 message = await get_commit_message_first_line()
107 if not commit_sha:
108 raise ValueError(
109 "commit_sha is required - must be provided or set in GITHUB_SHA"
110 )
111 if not branch:
112 raise ValueError(
113 "branch is required - must be provided or set in GITHUB_REF_NAME"
114 )
115 if not repository:
116 raise ValueError(
117 "repository is required - must be provided or set in GITHUB_REPOSITORY"
118 )
120 except Exception as e:
121 raise ValueError(
122 f"Error getting git version info: {e}. You may not be in a Github repository."
123 )
125 return GithubVersionInfo(
126 type="vcs:github",
127 version=commit_sha[:8],
128 commit_sha=commit_sha,
129 message=message,
130 branch=branch,
131 repository=repository,
132 url=url,
133 )
136async def get_gitlab_version_info( 1a
137 commit_sha: Optional[str] = None,
138 message: Optional[str] = None,
139 branch: Optional[str] = None,
140 repository: Optional[str] = None,
141 url: Optional[str] = None,
142) -> GitlabVersionInfo:
143 """Create a GitlabVersionInfo object from provided values or environment variables.
145 Args:
146 commit_sha: The commit SHA, falls back to CI_COMMIT_SHA env var
147 message: The commit message, falls back to git log -1 --pretty=%B
148 branch: The git branch, falls back to CI_COMMIT_REF_NAME env var
149 repository: The repository name, falls back to CI_PROJECT_NAME env var
150 url: The repository URL, constructed from CI_PROJECT_URL/-/tree/CI_COMMIT_SHA if not provided
152 Returns:
153 A GitlabVersionInfo
155 Raises:
156 ValueError: If any required fields cannot be determined
157 """
158 try:
159 commit_sha = commit_sha or os.getenv("CI_COMMIT_SHA")
160 branch = branch or os.getenv("CI_COMMIT_REF_NAME")
161 repository = repository or os.getenv("CI_PROJECT_NAME")
162 url = url or f"{os.getenv('CI_PROJECT_URL')}/-/tree/{commit_sha}"
164 if not message:
165 message = await get_commit_message_first_line()
167 if not commit_sha:
168 raise ValueError(
169 "commit_sha is required - must be provided or set in CI_COMMIT_SHA"
170 )
171 if not branch:
172 raise ValueError(
173 "branch is required - must be provided or set in CI_COMMIT_REF_NAME"
174 )
175 if not repository:
176 raise ValueError(
177 "repository is required - must be provided or set in CI_PROJECT_NAME"
178 )
179 if not url:
180 raise ValueError(
181 "url is required - must be provided or set in CI_PROJECT_URL"
182 )
184 except Exception as e:
185 raise ValueError(
186 f"Error getting git version info: {e}. You may not be in a Gitlab repository."
187 )
189 return GitlabVersionInfo(
190 type="vcs:gitlab",
191 version=commit_sha[:8],
192 commit_sha=commit_sha,
193 message=message,
194 branch=branch,
195 repository=repository,
196 url=url,
197 )
200async def get_bitbucket_version_info( 1a
201 commit_sha: Optional[str] = None,
202 message: Optional[str] = None,
203 branch: Optional[str] = None,
204 repository: Optional[str] = None,
205 url: Optional[str] = None,
206) -> BitbucketVersionInfo:
207 """Create a BitbucketVersionInfo object from provided values or environment variables.
209 Args:
210 commit_sha: The commit SHA, falls back to BITBUCKET_COMMIT env var
211 message: The commit message, falls back to git log -1 --pretty=%B
212 branch: The git branch, falls back to BITBUCKET_BRANCH env var
213 repository: The repository name, falls back to BITBUCKET_REPO_SLUG env var
214 url: The repository URL, constructed from BITBUCKET_GIT_HTTP_ORIGIN/BITBUCKET_REPO_SLUG/src/BITBUCKET_COMMIT if not provided
216 Returns:
217 A BitbucketVersionInfo
219 Raises:
220 ValueError: If any required fields cannot be determined
221 """
222 try:
223 commit_sha = commit_sha or os.getenv("BITBUCKET_COMMIT")
224 branch = branch or os.getenv("BITBUCKET_BRANCH")
225 repository = repository or os.getenv("BITBUCKET_REPO_SLUG")
226 url = url or f"{os.getenv('BITBUCKET_GIT_HTTP_ORIGIN')}/src/{commit_sha}"
228 if not message:
229 message = await get_commit_message_first_line()
231 if not commit_sha:
232 raise ValueError(
233 "commit_sha is required - must be provided or set in BITBUCKET_COMMIT"
234 )
235 if not branch:
236 raise ValueError(
237 "branch is required - must be provided or set in BITBUCKET_BRANCH"
238 )
239 if not repository:
240 raise ValueError(
241 "repository is required - must be provided or set in BITBUCKET_REPO_SLUG"
242 )
243 if not url:
244 raise ValueError(
245 "url is required - must be provided or set in BITBUCKET_GIT_HTTP_ORIGIN"
246 )
248 except Exception as e:
249 raise ValueError(
250 f"Error getting git version info: {e}. You may not be in a Bitbucket repository."
251 )
253 return BitbucketVersionInfo(
254 type="vcs:bitbucket",
255 version=commit_sha[:8],
256 commit_sha=commit_sha,
257 message=message,
258 branch=branch,
259 repository=repository,
260 url=url,
261 )
264async def get_azuredevops_version_info( 1a
265 commit_sha: Optional[str] = None,
266 message: Optional[str] = None,
267 branch: Optional[str] = None,
268 repository: Optional[str] = None,
269 url: Optional[str] = None,
270) -> AzureDevopsVersionInfo:
271 """Create an AzureDevopsVersionInfo object from provided values or environment variables.
273 Args:
274 commit_sha: The commit SHA, falls back to BUILD_SOURCEVERSION env var
275 message: The commit message, falls back to git log -1 --pretty=%B
276 branch: The git branch, falls back to BUILD_SOURCEBRANCHNAME env var
277 repository: The repository name, falls back to BUILD_REPOSITORY_NAME env var
278 url: The repository URL, constructed from BUILD_REPOSITORY_URI?version=GCBUILD_SOURCEVERSION if not provided
280 Returns:
281 An AzureDevopsVersionInfo
283 Raises:
284 ValueError: If any required fields cannot be determined
285 """
286 try:
287 commit_sha = commit_sha or os.getenv("BUILD_SOURCEVERSION")
288 branch = branch or os.getenv("BUILD_SOURCEBRANCHNAME")
289 repository = repository or os.getenv("BUILD_REPOSITORY_NAME")
290 url = url or f"{os.getenv('BUILD_REPOSITORY_URI')}?version=GC{commit_sha}"
292 if not message:
293 message = await get_commit_message_first_line()
295 if not commit_sha:
296 raise ValueError(
297 "commit_sha is required - must be provided or set in BUILD_SOURCEVERSION"
298 )
299 if not branch:
300 raise ValueError(
301 "branch is required - must be provided or set in BUILD_SOURCEBRANCHNAME"
302 )
303 if not repository:
304 raise ValueError(
305 "repository is required - must be provided or set in BUILD_REPOSITORY_NAME"
306 )
307 if not url:
308 raise ValueError(
309 "url is required - must be provided or set in BUILD_REPOSITORY_URI"
310 )
312 except Exception as e:
313 raise ValueError(
314 f"Error getting git version info: {e}. You may not be in an Azure DevOps repository."
315 )
317 return AzureDevopsVersionInfo(
318 type="vcs:azuredevops",
319 version=commit_sha[:8],
320 commit_sha=commit_sha,
321 message=message,
322 branch=branch,
323 repository=repository,
324 url=url,
325 )
328async def get_git_version_info( 1a
329 commit_sha: Optional[str] = None,
330 message: Optional[str] = None,
331 branch: Optional[str] = None,
332 url: Optional[str] = None,
333 repository: Optional[str] = None,
334) -> GitVersionInfo:
335 try:
336 if not commit_sha:
337 # Run git command and get stdout
338 result = await run_process(["git", "rev-parse", "HEAD"])
339 # Decode bytes to string and strip whitespace
340 commit_sha = result.stdout.decode().strip()
342 if not branch:
343 result = await run_process(["git", "rev-parse", "--abbrev-ref", "HEAD"])
344 branch = result.stdout.decode().strip()
346 if not repository:
347 result = await run_process(["git", "config", "--get", "remote.origin.url"])
348 remote_url = result.stdout.decode().strip()
350 # Extract just the repository name (last part of the path)
351 repo_url = urlparse(remote_url)
352 repository = repo_url.path.strip("/")
353 if repository.endswith(".git"):
354 repository = repository[:-4]
356 if not message:
357 message = await get_commit_message_first_line()
359 if not url and repository:
360 # Use the full remote URL as the URL
361 result = await run_process(["git", "config", "--get", "remote.origin.url"])
362 url = result.stdout.decode().strip()
363 except Exception as e:
364 raise ValueError(
365 f"Error getting git version info: {e}. You may not be in a git repository."
366 )
368 if not url:
369 raise ValueError("Could not determine git repository URL")
371 return GitVersionInfo(
372 type="vcs:git",
373 version=commit_sha[:8],
374 branch=branch,
375 url=url,
376 repository=repository,
377 commit_sha=commit_sha,
378 message=message,
379 )
382class VersionType(str, Enum): 1a
383 SIMPLE = "prefect:simple" 1a
384 GITHUB = "vcs:github" 1a
385 GITLAB = "vcs:gitlab" 1a
386 BITBUCKET = "vcs:bitbucket" 1a
387 AZUREDEVOPS = "vcs:azuredevops" 1a
388 GIT = "vcs:git" 1a
391async def get_inferred_version_info( 1a
392 version_type: Optional[str] = None,
393) -> VersionInfo | None:
394 """
395 Attempts to infer version information from the environment.
397 Args:
398 version_type: Optional type of version info to get. If provided, only that
399 type will be attempted.
401 Returns:
402 VersionInfo: The inferred version information
404 Raises:
405 ValueError: If unable to infer version info from any source
406 """
407 # Map version types to their getter functions
408 type_to_getter: Dict[str, Callable[..., Coroutine[Any, Any, Any]]] = {
409 VersionType.GITHUB: get_github_version_info,
410 VersionType.GITLAB: get_gitlab_version_info,
411 VersionType.BITBUCKET: get_bitbucket_version_info,
412 VersionType.AZUREDEVOPS: get_azuredevops_version_info,
413 VersionType.GIT: get_git_version_info,
414 }
416 # Default order of getters to try
417 default_getters = [
418 get_github_version_info,
419 get_gitlab_version_info,
420 get_bitbucket_version_info,
421 get_azuredevops_version_info,
422 get_git_version_info,
423 ]
425 if version_type is VersionType.SIMPLE:
426 return None
427 if version_type:
428 if version_type not in type_to_getter:
429 raise ValueError(f"Unknown version type: {version_type}")
430 getters = [type_to_getter[version_type]]
431 else:
432 getters = default_getters
434 for getter in getters:
435 try:
436 return await getter()
437 except ValueError:
438 continue
440 return None