Coverage for app/crud/apps.py: 100.00%

1from uuid import UUID 

2 

3from sqlalchemy.orm import Session 

4 

5from app.models import App, AppVersionApproval, AppVersionApprovalStatus 

6from app.schemas import AppCreate, AppUpdate 

7from app.utils.time import utcnow 

8 

9 

10def get_app( 

11 db: Session, 

12 app_id: UUID, 

13 include_deleted: bool = False, 

14) -> App | None: 

15 """Get app by ID. Hides soft-deleted apps by default. 

16 

17 ``include_deleted=True`` is reserved for the rare audit lookup — 

18 the HTTP API never sets it. 

19 """ 

20 q = db.query(App).filter(App.appId == app_id) 

21 if not include_deleted: 

22 q = q.filter(App.deleted_at.is_(None)) 

23 return q.first() 

24 

25 

26def get_apps( 

27 db: Session, 

28 skip: int = 0, 

29 limit: int = 100, 

30 user_id: UUID | None = None, 

31 include_deleted: bool = False, 

32) -> list[App]: 

33 """Get apps with optional user filter. Hides soft-deleted by default.""" 

34 query = db.query(App) 

35 

36 if not include_deleted: 

37 query = query.filter(App.deleted_at.is_(None)) 

38 if user_id: 

39 query = query.filter(App.userId == user_id) 

40 

41 return query.offset(skip).limit(limit).all() 

42 

43 

44def get_visible_apps( 

45 db: Session, 

46 requesting_user_id: UUID, 

47 skip: int = 0, 

48 limit: int = 100, 

49) -> list[App]: 

50 """Return apps visible to the requesting user. 

51 

52 Visibility rules: 

53 - Always: apps owned by the requesting user (regardless of is_private) 

54 - Additionally: public apps (is_private=False) that have at least one 

55 APPROVED version 

56 """ 

57 approved_app_ids = ( 

58 db.query(AppVersionApproval.appId) 

59 .filter(AppVersionApproval.status == AppVersionApprovalStatus.APPROVED) 

60 .distinct() 

61 .scalar_subquery() 

62 ) 

63 

64 return ( 

65 db.query(App) 

66 .filter(App.deleted_at.is_(None)) 

67 .filter( 

68 (App.userId == requesting_user_id) 

69 | ( 

70 App.is_private.is_(False) 

71 & App.appId.in_(approved_app_ids) 

72 ) 

73 ) 

74 .offset(skip) 

75 .limit(limit) 

76 .all() 

77 ) 

78 

79 

80def create_app(db: Session, app: AppCreate, user_id: UUID) -> App: 

81 """Create a new app.""" 

82 db_app = App( 

83 name=app.name, 

84 description=app.description, 

85 git_link=app.git_link, 

86 is_private=app.is_private, 

87 userId=user_id, 

88 ) 

89 db.add(db_app) 

90 db.commit() 

91 db.refresh(db_app) 

92 return db_app 

93 

94 

95def update_app(db: Session, app_id: UUID, app_update: AppUpdate) -> App | None: 

96 """Update app information. 

97 

98 The ``image`` field is intentionally NOT applied here — the router 

99 decodes the data-URL via ``parse_image_data_url`` and calls 

100 ``set_app_image`` directly. ``model_dump`` would otherwise pass 

101 the data-URL string straight into the ``LargeBinary`` column. 

102 

103 ``git_link`` is intentionally excluded as well: once an app has 

104 deployments, changing the repo would make existing deployments 

105 point at a different repo than they originally deployed. 

106 ``AppUpdate`` already drops the field from the schema; this 

107 exclude is defense-in-depth in case the schema is replaced or 

108 extended later. 

109 """ 

110 db_app = get_app(db, app_id) 

111 if not db_app: 

112 return None 

113 

114 update_data = app_update.model_dump( 

115 exclude_unset=True, exclude={"image", "git_link"} 

116 ) 

117 for field, value in update_data.items(): 

118 setattr(db_app, field, value) 

119 

120 db.commit() 

121 db.refresh(db_app) 

122 return db_app 

123 

124 

125def set_app_image( 

126 db: Session, 

127 app_id: UUID, 

128 image_bytes: bytes | None, 

129 image_mime: str | None, 

130) -> App | None: 

131 """Persist the image bytes + mime atomically. 

132 

133 Both args ``None`` clears the image. Otherwise both must be set — 

134 the router enforces that via ``parse_image_data_url`` before 

135 calling here, so this function trusts its inputs. 

136 """ 

137 db_app = get_app(db, app_id) 

138 if not db_app: 

139 return None 

140 db_app.image = image_bytes 

141 db_app.image_mime = image_mime 

142 db.commit() 

143 db.refresh(db_app) 

144 return db_app 

145 

146 

147def soft_delete_app(db: Session, app_id: UUID) -> bool: 

148 """Mark an app as deleted without removing the row. 

149 

150 The row stays so existing ``deployments.appId`` foreign keys keep 

151 resolving, but list queries skip it. Restoring an app means clearing 

152 ``deleted_at`` directly via SQL. 

153 """ 

154 db_app = get_app(db, app_id) 

155 if not db_app: 

156 return False 

157 db_app.deleted_at = utcnow() 

158 db.commit() 

159 return True