Coverage for src/ezcompiler/services/pipeline_service.py: 65.79%
32 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-03 15:46 +0000
1# ///////////////////////////////////////////////////////////////
2# PIPELINE_SERVICE - Build pipeline orchestration helpers
3# Project: ezcompiler
4# ///////////////////////////////////////////////////////////////
6"""
7Pipeline service - Compilation, ZIP and upload orchestration.
9This service extracts the compile->zip->upload workflow from interfaces
10so the orchestration logic remains reusable and testable.
11"""
13from __future__ import annotations
15# ///////////////////////////////////////////////////////////////
16# IMPORTS
17# ///////////////////////////////////////////////////////////////
18# Standard library imports
19from collections.abc import Callable
20from pathlib import Path
21from typing import Any, Literal, cast
23# Local imports
24from ..shared import CompilationResult, CompilerConfig
25from .compiler_service import CompilerService
26from .uploader_service import UploaderService
28# ///////////////////////////////////////////////////////////////
29# CLASSES
30# ///////////////////////////////////////////////////////////////
33class PipelineService:
34 """Service that coordinates compile, zip and upload stages."""
36 def __init__(
37 self,
38 compiler_service_factory: (
39 Callable[[CompilerConfig], CompilerService] | None
40 ) = None,
41 ) -> None:
42 """Initialise the pipeline service.
44 Args:
45 compiler_service_factory: Optional factory to create a CompilerService
46 from a CompilerConfig. Defaults to ``CompilerService`` constructor.
47 Inject a custom factory in tests to avoid triggering real compilation.
48 """
49 self._compiler_service_factory: Callable[[CompilerConfig], CompilerService] = (
50 compiler_service_factory or CompilerService
51 )
53 def compile_project(
54 self,
55 config: CompilerConfig,
56 console: bool = True,
57 compiler: str | None = None,
58 ) -> tuple[CompilerService, CompilationResult]:
59 """Compile a project and return service + result."""
60 compiler_service = self._compiler_service_factory(config)
61 compilation_result = compiler_service.compile(
62 console=console,
63 compiler=cast(
64 Literal["Cx_Freeze", "PyInstaller", "Nuitka", "auto"] | None,
65 compiler,
66 ),
67 )
68 return compiler_service, compilation_result
70 def zip_artifact(
71 self,
72 config: CompilerConfig,
73 compiler_service: CompilerService,
74 compilation_result: CompilationResult | None,
75 progress_callback: Callable[[str, int], None] | None = None,
76 ) -> bool:
77 """Create ZIP artifact when required and return True when created."""
78 zip_needed = (
79 compilation_result.zip_needed if compilation_result else config.zip_needed
80 )
81 if not zip_needed:
82 return False
84 compiler_service._zip_artifact(
85 output_path=str(config.zip_file_path),
86 progress_callback=progress_callback,
87 )
88 return True
90 @staticmethod
91 def build_stages(
92 config: CompilerConfig,
93 should_zip: bool = False,
94 should_upload: bool = False,
95 ) -> list[dict[str, Any]]:
96 """
97 Build the stage list for dynamic_layered_progress.
99 Args:
100 config: Compiler configuration (used for display labels)
101 should_zip: Whether a ZIP stage should be included
102 should_upload: Whether an upload stage should be included
104 Returns:
105 list[dict]: Stage configuration list ready for dynamic_layered_progress
106 """
107 stages: list[dict[str, Any]] = [
108 {
109 "name": "main",
110 "type": "main",
111 "description": f"Building {config.project_name} v{config.version}",
112 },
113 {
114 "name": "version",
115 "type": "spinner",
116 "description": "Generating version file",
117 },
118 {
119 "name": "compile",
120 "type": "spinner",
121 "description": f"Compiling with {config.compiler}",
122 },
123 ]
124 if should_zip:
125 stages.append(
126 {
127 "name": "zip",
128 "type": "progress",
129 "description": "Creating ZIP archive",
130 "total": 100,
131 }
132 )
133 if should_upload:
134 stages.append(
135 {
136 "name": "upload",
137 "type": "spinner",
138 "description": "Uploading artifacts",
139 }
140 )
141 return stages
143 def upload_artifact(
144 self,
145 config: CompilerConfig,
146 structure: str,
147 destination: str,
148 compilation_result: CompilationResult | None,
149 upload_config: dict[str, Any] | None = None,
150 ) -> None:
151 """Upload project artifact to a destination."""
152 zip_needed = (
153 compilation_result.zip_needed if compilation_result else config.zip_needed
154 )
155 source_file = (
156 str(config.zip_file_path) if zip_needed else str(config.output_folder)
157 )
159 UploaderService.upload(
160 source_path=Path(source_file),
161 upload_type=cast(Literal["disk", "server"], structure),
162 destination=destination,
163 upload_config=upload_config,
164 )