Create a new deployment.
(
self,
project: Project,
branch: str,
commit: dict,
db: AsyncSession,
redis_client: Redis,
trigger: str = "user",
current_user: User | None = None,
)
| 389 | pass |
| 390 | |
| 391 | async def create( |
| 392 | self, |
| 393 | project: Project, |
| 394 | branch: str, |
| 395 | commit: dict, |
| 396 | db: AsyncSession, |
| 397 | redis_client: Redis, |
| 398 | trigger: str = "user", |
| 399 | current_user: User | None = None, |
| 400 | ) -> Deployment: |
| 401 | """Create a new deployment.""" |
| 402 | |
| 403 | environment = get_environment_for_branch(branch, project.active_environments) |
| 404 | if not environment: |
| 405 | raise ValueError("No environment found for this branch.") |
| 406 | |
| 407 | config = project.config or {} |
| 408 | runner_slug = config.get("runner") or config.get("image") |
| 409 | if not runner_slug: |
| 410 | raise ValueError("Runner not set in project config.") |
| 411 | registry_state = RegistryService( |
| 412 | Path(get_settings().data_dir) / "registry" |
| 413 | ).state |
| 414 | runner_entry = next( |
| 415 | ( |
| 416 | runner |
| 417 | for runner in registry_state.runners |
| 418 | if runner.get("slug") == runner_slug |
| 419 | ), |
| 420 | None, |
| 421 | ) |
| 422 | if not runner_entry: |
| 423 | raise ValueError(f"Runner '{runner_slug}' not found in registry.") |
| 424 | if runner_entry.get("enabled") is not True: |
| 425 | raise ValueError(f"Runner '{runner_slug}' is disabled.") |
| 426 | runner_image = runner_entry.get("image") |
| 427 | if not runner_image: |
| 428 | raise ValueError(f"Runner '{runner_slug}' has no image configured.") |
| 429 | |
| 430 | commit_user_author = commit.get("author") or {} |
| 431 | commit_user_committer = commit.get("committer") or {} |
| 432 | commit_payload = commit.get("commit") or {} |
| 433 | commit_payload_author = commit_payload.get("author") or {} |
| 434 | commit_payload_committer = commit_payload.get("committer") or {} |
| 435 | |
| 436 | author = ( |
| 437 | commit_user_author.get("login") |
| 438 | or commit_user_committer.get("login") |
| 439 | or commit_payload_author.get("name") |
| 440 | or commit_payload_committer.get("name") |
| 441 | or "" |
| 442 | ) |
| 443 | message = commit_payload.get("message") or "" |
| 444 | date_raw = ( |
| 445 | commit_payload_author.get("date") |
| 446 | or commit_payload_committer.get("date") |
| 447 | or datetime.now(timezone.utc).isoformat() |
| 448 | ) |
no test coverage detected