Coverage for app/crud/deployments.py: 80.79%
151 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-25 15:51 +0000
1import json
2from datetime import datetime
3from typing import Any
4from uuid import UUID
6from sqlalchemy import and_, asc, desc, exists, func
7from sqlalchemy.orm import Session, joinedload
9from app.models import (
10 Deployment,
11 Task,
12 TaskStatus,
13 TaskType,
14 Team,
15 User,
16 UserToDeployment,
17 UserToTeam,
18)
19from app.schemas import DeploymentCreate
20from app.utils.time import utcnow
23def _is_destroyed_subq():
24 """Correlated EXISTS: deployment has a successful DESTROY task."""
25 return exists().where(
26 (Task.deploymentId == Deployment.deploymentId)
27 & (Task.type == TaskType.DESTROY)
28 & (Task.status == TaskStatus.SUCCESS)
29 )
32def count_active_user_deployments(db: Session, user_id: UUID) -> int:
33 """Number of *active* deployments owned by ``user_id``.
35 "Active" here matches what the user sees on the Deployments page:
37 * Owned by them (``Deployment.userId == user_id``).
38 * Not soft-deleted (``deleted_at IS NULL``).
39 * Has not been fully destroyed (no successful DESTROY task) —
40 soft-delete happens after destroy completes, so the check
41 covers the destroy-in-flight window too.
42 """
43 return (
44 db.query(Deployment)
45 .filter(Deployment.userId == user_id)
46 .filter(Deployment.deleted_at.is_(None))
47 .filter(~_is_destroyed_subq())
48 .count()
49 )
52def get_deployment(
53 db: Session,
54 deployment_id: UUID,
55 include_deleted: bool = False,
56) -> Deployment | None:
57 """Get deployment by ID. Hides soft-deleted rows by default.
59 ``include_deleted=True`` is for the rare audit/restore lookup; the
60 HTTP API never sets it.
61 """
62 q = db.query(Deployment).filter(Deployment.deploymentId == deployment_id)
63 if not include_deleted:
64 q = q.filter(Deployment.deleted_at.is_(None))
65 return q.first()
68def get_deployment_with_details(
69 db: Session,
70 deployment_id: UUID,
71 include_deleted: bool = False,
72) -> Deployment | None:
73 """Get deployment by ID with all relations loaded. Hides soft-deleted by default."""
74 q = (
75 db.query(Deployment)
76 .options(
77 joinedload(Deployment.user),
78 joinedload(Deployment.app),
79 joinedload(Deployment.teams),
80 )
81 .filter(Deployment.deploymentId == deployment_id)
82 )
83 if not include_deleted:
84 q = q.filter(Deployment.deleted_at.is_(None))
85 return q.first()
88def get_latest_task(db: Session, deployment_id: UUID) -> Task | None:
89 """Get the most recent task for a deployment"""
90 return (
91 db.query(Task)
92 .filter(Task.deploymentId == deployment_id)
93 .order_by(desc(Task.created_at))
94 .first()
95 )
98def get_first_task(db: Session, deployment_id: UUID) -> Task | None:
99 """Get the first task for a deployment (when deployment was created)"""
100 return (
101 db.query(Task)
102 .filter(Task.deploymentId == deployment_id)
103 .order_by(asc(Task.created_at))
104 .first()
105 )
108def derive_status(
109 task_status: TaskStatus | None,
110 task_type: TaskType | None,
111) -> str | None:
112 """Synthesize the effective deployment status from the latest task's
113 ``(status, type)`` pair.
115 ``task.status`` alone isn't enough: a destroy in flight surfaces as
116 ``destroying`` and a finished destroy as ``destroyed`` (neither is a
117 stored enum value). Pause/resume follow the same pattern:
119 * ``(PAUSE, pending|running)`` → ``pausing``
120 * ``(PAUSE, success)`` → ``paused``
121 * ``(RESUME, pending|running)`` → ``resuming``
122 * ``(RESUME, success)`` → falls through to ``success``
124 PAUSE/RESUME ``failed``/``cancelled`` bleed through unchanged so the
125 user sees the pause/resume itself broke. Returns ``None`` when the
126 deployment has no tasks yet.
127 """
128 if task_status is None:
129 return None
131 raw_status = task_status.value
132 raw_type = task_type.value if task_type else None
134 if raw_type == "destroy":
135 if raw_status in ("pending", "running"):
136 return "destroying"
137 if raw_status == "success":
138 return "destroyed"
139 # failed/cancelled bleed through unchanged so the user sees that
140 # the destroy itself broke (vs. the original deploy succeeded).
141 elif raw_type == "pause":
142 if raw_status in ("pending", "running"):
143 return "pausing"
144 if raw_status == "success":
145 return "paused"
146 if raw_status == "failed":
147 # Distinguish a pause failure from a deploy failure — the
148 # resources are still running, only the stop pass broke.
149 return "pause_failed"
150 # cancelled bleeds through unchanged.
151 elif raw_type == "resume":
152 if raw_status in ("pending", "running"):
153 return "resuming"
154 if raw_status == "failed":
155 # Same as pause_failed: instances are still SHUTOFF, only
156 # the start pass tripped.
157 return "resume_failed"
158 # On resume success the deployment is running again; let
159 # "success" pass through so the lifecycle matrix treats it like
160 # a fresh successful deploy.
161 return raw_status
164def get_deployment_status(db: Session, deployment_id: UUID) -> str | None:
165 """Effective deployment status for a single deployment.
167 Thin wrapper around ``derive_status`` for the per-deployment path
168 (detail endpoint, single-row callers). The list endpoint goes
169 through ``bulk_get_task_summary`` instead so it doesn't fan out
170 one query per row.
172 Returns ``None`` if the deployment has no tasks yet.
173 """
174 task = get_latest_task(db, deployment_id)
175 if task is None:
176 return None
177 return derive_status(task.status, task.type)
180def get_deployment_created_at(db: Session, deployment_id: UUID):
181 """Get deployment creation time from first task"""
182 task = get_first_task(db, deployment_id)
183 return task.created_at if task else None
186def bulk_get_task_summary(
187 db: Session, deployment_ids: list[UUID]
188) -> dict[UUID, tuple[TaskStatus | None, TaskType | None, datetime | None]]:
189 """Fetch the latest-task ``(status, type)`` and the first-task
190 ``created_at`` for every deployment in ``deployment_ids`` — in two
191 queries regardless of how many deployments are passed.
193 Returns a dict keyed by ``deploymentId``. Deployments with no tasks
194 are absent from the map; callers use
195 ``.get(deployment_id, (None, None, None))``.
196 """
198 if not deployment_ids:
199 return {}
201 # Latest task per deployment via row_number() over (PARTITION BY ...
202 # ORDER BY created_at DESC).
203 latest_rn = (
204 func.row_number()
205 .over(partition_by=Task.deploymentId, order_by=desc(Task.created_at))
206 .label("rn")
207 )
208 latest_subq = (
209 db.query(
210 Task.deploymentId.label("did"),
211 Task.status.label("status"),
212 Task.type.label("type"),
213 latest_rn,
214 )
215 .filter(Task.deploymentId.in_(deployment_ids))
216 .subquery()
217 )
218 latest_rows = (
219 db.query(latest_subq.c.did, latest_subq.c.status, latest_subq.c.type)
220 .filter(latest_subq.c.rn == 1)
221 .all()
222 )
224 # First task per deployment (ascending) for created_at.
225 first_rn = (
226 func.row_number()
227 .over(partition_by=Task.deploymentId, order_by=asc(Task.created_at))
228 .label("rn")
229 )
230 first_subq = (
231 db.query(
232 Task.deploymentId.label("did"),
233 Task.created_at.label("created_at"),
234 first_rn,
235 )
236 .filter(Task.deploymentId.in_(deployment_ids))
237 .subquery()
238 )
239 first_rows = (
240 db.query(first_subq.c.did, first_subq.c.created_at)
241 .filter(first_subq.c.rn == 1)
242 .all()
243 )
245 first_map = {row.did: row.created_at for row in first_rows}
246 return {
247 row.did: (row.status, row.type, first_map.get(row.did))
248 for row in latest_rows
249 }
252def get_team_members(db: Session, team_id: UUID) -> list[User]:
253 """Get all users in a team"""
254 user_ids = (
255 db.query(UserToTeam.userId)
256 .filter(UserToTeam.teamId == team_id)
257 .all()
258 )
259 user_ids = [uid[0] for uid in user_ids]
261 if not user_ids:
262 return []
264 return db.query(User).filter(User.userId.in_(user_ids)).all()
267def get_deployment_teams_with_members(db: Session, deployment_id: UUID) -> list[dict[str, Any]]:
268 """Get all teams for a deployment with their members"""
269 teams = db.query(Team).filter(Team.deploymentId == deployment_id).all()
271 result = []
272 for team in teams:
273 members = get_team_members(db, team.teamId)
274 result.append({
275 "teamId": team.teamId,
276 "name": team.name,
277 "members": [
278 {
279 "userId": member.userId,
280 "email": member.email,
281 "username": member.username
282 }
283 for member in members
284 ]
285 })
287 return result
290def get_deployment_outputs(db: Session, deployment_id: UUID) -> dict[str, Any] | None:
291 """Get parsed Terraform outputs from the latest successful task"""
292 task = (
293 db.query(Task)
294 .filter(Task.deploymentId == deployment_id)
295 .filter(Task.outputs.isnot(None))
296 .order_by(desc(Task.created_at))
297 .first()
298 )
300 if task and task.outputs:
301 try:
302 return json.loads(task.outputs)
303 except json.JSONDecodeError:
304 return None
305 return None
308def get_latest_successful_deploy_outputs(
309 db: Session, deployment_id: UUID
310) -> dict[str, Any] | None:
311 """Parsed Terraform outputs of the most recent *successful DEPLOY* task.
313 Stricter than :func:`get_deployment_outputs`, which returns the latest
314 task carrying any non-null ``outputs`` regardless of type or status — a
315 DESTROY task's (empty) outputs could win there. Credential extraction
316 must read the outputs of an actual successful deploy, so this helper
317 filters to ``type == DEPLOY`` and ``status == SUCCESS``.
319 Returns ``None`` when no such task exists or its ``outputs`` are absent
320 / unparseable. Shared by ``resend_user_access`` and the per-member
321 ``/my-access`` endpoint so both agree on which outputs are authoritative.
322 """
323 task = (
324 db.query(Task)
325 .filter(
326 Task.deploymentId == deployment_id,
327 Task.type == TaskType.DEPLOY,
328 Task.status == TaskStatus.SUCCESS,
329 )
330 .order_by(desc(Task.created_at))
331 .first()
332 )
334 if task and task.outputs:
335 try:
336 return json.loads(task.outputs) if isinstance(task.outputs, str) else task.outputs
337 except json.JSONDecodeError:
338 return None
339 return None
342def get_deployments(
343 db: Session,
344 skip: int = 0,
345 limit: int = 100,
346 user_id: UUID | None = None,
347 member_user_id: UUID | None = None,
348 app_id: UUID | None = None,
349 status: str | None = None,
350 include_deleted: bool = False,
351) -> list[Deployment]:
352 """Get deployments with optional filters. Hides soft-deleted by default.
354 ``user_id`` filters by deployment owner (``Deployment.userId``).
356 ``member_user_id`` filters by owner or membership: a deployment
357 matches when the user is the creator OR appears in any team's
358 ``UserToTeam`` row OR has a direct ``UserToDeployment`` mapping.
359 Mutually exclusive with ``user_id``; if both are set ``user_id``
360 wins.
361 """
362 query = db.query(Deployment)
364 if not include_deleted:
365 # Backed by the partial index ix_deployments_live so this stays
366 # cheap even with many rows.
367 query = query.filter(Deployment.deleted_at.is_(None))
368 if user_id:
369 query = query.filter(Deployment.userId == user_id)
370 elif member_user_id:
371 # Owner OR team member OR direct mapping, via a subquery on the
372 # user's teamIds so the OR doesn't explode into a cartesian.
373 member_team_ids = (
374 db.query(UserToTeam.teamId).filter(UserToTeam.userId == member_user_id)
375 )
376 member_deployment_ids_via_teams = (
377 db.query(Team.deploymentId).filter(Team.teamId.in_(member_team_ids))
378 )
379 member_deployment_ids_direct = (
380 db.query(UserToDeployment.deploymentId)
381 .filter(UserToDeployment.userId == member_user_id)
382 )
383 query = query.filter(
384 (Deployment.userId == member_user_id)
385 | (Deployment.deploymentId.in_(member_deployment_ids_via_teams))
386 | (Deployment.deploymentId.in_(member_deployment_ids_direct))
387 )
388 if app_id:
389 query = query.filter(Deployment.appId == app_id)
391 # Filter by effective status. The exposed status is derived from the
392 # LATEST task per deployment (see ``derive_status``), so we join a
393 # window-function subquery pinning the latest task and apply the
394 # equivalent predicate here — before offset/limit — so the page size
395 # stays correct.
396 if status:
397 latest_rn = (
398 func.row_number()
399 .over(partition_by=Task.deploymentId, order_by=desc(Task.created_at))
400 .label("rn")
401 )
402 latest_subq = (
403 db.query(
404 Task.deploymentId.label("did"),
405 Task.status.label("status"),
406 Task.type.label("type"),
407 latest_rn,
408 ).subquery()
409 )
410 query = query.join(
411 latest_subq,
412 and_(
413 latest_subq.c.did == Deployment.deploymentId,
414 latest_subq.c.rn == 1,
415 ),
416 )
418 if status == "destroying":
419 query = query.filter(
420 latest_subq.c.type == TaskType.DESTROY,
421 latest_subq.c.status.in_((TaskStatus.PENDING, TaskStatus.RUNNING)),
422 )
423 elif status == "destroyed":
424 query = query.filter(
425 latest_subq.c.type == TaskType.DESTROY,
426 latest_subq.c.status == TaskStatus.SUCCESS,
427 )
428 else:
429 # Plain task statuses, mirroring ``derive_status``:
430 # - ``pending``/``running``/``success`` match deploy-typed
431 # tasks only (destroy surfaces as destroying/destroyed).
432 # - ``failed``/``cancelled`` bleed through both task types.
433 try:
434 status_enum = TaskStatus(status)
435 except ValueError:
436 # Unknown status string → empty result.
437 return []
438 query = query.filter(latest_subq.c.status == status_enum)
439 if status_enum in (
440 TaskStatus.PENDING,
441 TaskStatus.RUNNING,
442 TaskStatus.SUCCESS,
443 ):
444 query = query.filter(latest_subq.c.type != TaskType.DESTROY)
446 # Order by deploymentId (UUID)
447 query = query.order_by(desc(Deployment.deploymentId))
449 return query.offset(skip).limit(limit).all()
452def create_deployment(db: Session, deployment: DeploymentCreate, user_id: UUID) -> Deployment:
453 """Insert a deployment row in the current transaction.
455 Does NOT commit — the caller is expected to also insert teams/tasks
456 in the same TX and commit once at the end. This is necessary so the
457 advisory lock acquired at the start of the request stays held across
458 all related inserts.
459 """
460 # Convert userInputVar dict to JSON string for database storage
461 user_input_var_json = None
462 if deployment.userInputVar is not None:
463 user_input_var_json = json.dumps(deployment.userInputVar)
465 db_deployment = Deployment(
466 name=deployment.name,
467 appId=deployment.appId,
468 userId=user_id,
469 releaseTag=deployment.releaseTag,
470 userInputVar=user_input_var_json,
471 )
472 db.add(db_deployment)
473 db.flush()
474 db.refresh(db_deployment)
475 return db_deployment
477def soft_delete_deployment(db: Session, deployment_id: UUID) -> bool:
478 """Mark a deployment as deleted without removing the row.
480 Sets ``deleted_at = utcnow()`` so default queries skip it. The
481 related tasks/teams/user-mappings are intentionally untouched —
482 they're useful for audit and the partial-unique index on active
483 tasks already prevents the deployment from accepting new work.
485 Returns ``False`` if the deployment doesn't exist (or was already
486 deleted), ``True`` on a successful soft-delete.
487 """
488 db_deployment = get_deployment(db, deployment_id)
489 if not db_deployment:
490 return False
491 db_deployment.deleted_at = utcnow()
492 db.commit()
493 return True
496def create_user_to_deployments(
497 db: Session,
498 deployment_id: UUID,
499 user_ids: set[UUID],
500) -> list[UserToDeployment]:
501 """
502 Create UserToDeployment entries for multiple users
503 """
504 user_to_deployments = []
506 for user_id in user_ids:
507 user_to_deployment = UserToDeployment(
508 userId=user_id,
509 deploymentId=deployment_id
510 )
511 db.add(user_to_deployment)
512 user_to_deployments.append(user_to_deployment)
514 return user_to_deployments