Coverage for /usr/local/lib/python3.12/site-packages/prefect/docker/docker_image.py: 24%

1from pathlib import Path 1 ctx1a

2from typing import Any, Optional 1 ctx1a

3 

4from prefect.settings import ( 1 ctx1a

5 PREFECT_DEFAULT_DOCKER_BUILD_NAMESPACE, 

6) 

7from prefect.types._datetime import now 1 ctx1a

8from prefect.utilities.dockerutils import ( 1 ctx1a

9 PushError, 

10 build_image, 

11 docker_client, 

12 generate_default_dockerfile, 

13 parse_image_tag, 

14 split_repository_path, 

15) 

16from prefect.utilities.slugify import slugify 1 ctx1a

17 

18 

19class DockerImage: 1 ctx1a

20 """ 

21 Configuration used to build and push a Docker image for a deployment. 

22 

23 Attributes: 

24 name: The name of the Docker image to build, including the registry and 

25 repository. 

26 tag: The tag to apply to the built image. 

27 dockerfile: The path to the Dockerfile to use for building the image. If 

28 not provided, a default Dockerfile will be generated. 

29 **build_kwargs: Additional keyword arguments to pass to the Docker build request. 

30 See the [`docker-py` documentation](https://docker-py.readthedocs.io/en/stable/images.html#docker.models.images.ImageCollection.build) 

31 for more information. 

32 

33 """ 

34 

35 def __init__( 1 ctx1a

36 self, 

37 name: str, 

38 tag: Optional[str] = None, 

39 dockerfile: str = "auto", 

40 **build_kwargs: Any, 

41 ): 

42 image_name, image_tag = parse_image_tag(name) 

43 if tag and image_tag: 

44 raise ValueError( 

45 f"Only one tag can be provided - both {image_tag!r} and {tag!r} were" 

46 " provided as tags." 

47 ) 

48 namespace, repository = split_repository_path(image_name) 

49 # if the provided image name does not include a namespace (registry URL or user/org name), 

50 # use the default namespace 

51 if not namespace: 

52 namespace = PREFECT_DEFAULT_DOCKER_BUILD_NAMESPACE.value() 

53 # join the namespace and repository to create the full image name 

54 # ignore namespace if it is None 

55 self.name: str = "/".join(filter(None, [namespace, repository])) 

56 self.tag: str = tag or image_tag or slugify(now("UTC").isoformat()) 

57 self.dockerfile: str = dockerfile 

58 self.build_kwargs: dict[str, Any] = build_kwargs 

59 

60 @property 1 ctx1a

61 def reference(self) -> str: 1 ctx1a

62 return f"{self.name}:{self.tag}" 

63 

64 def build(self) -> None: 1 ctx1a

65 full_image_name = self.reference 

66 build_kwargs = self.build_kwargs.copy() 

67 if "context" not in build_kwargs: 

68 build_kwargs["context"] = Path.cwd() 

69 build_kwargs["tag"] = full_image_name 

70 build_kwargs["pull"] = build_kwargs.get("pull", True) 

71 

72 if self.dockerfile == "auto": 

73 with generate_default_dockerfile(): 

74 build_image(**build_kwargs) 

75 else: 

76 build_kwargs["dockerfile"] = self.dockerfile 

77 build_image(**build_kwargs) 

78 

79 def push(self) -> None: 1 ctx1a

80 with docker_client() as client: 

81 events = client.api.push( 

82 repository=self.name, tag=self.tag, stream=True, decode=True 

83 ) 

84 for event in events: 

85 if "error" in event: 

86 raise PushError(event["error"])