Coverage for open_webui/routers/automations.py: 51%

1import asyncio 

2import logging 

3from typing import Optional 

4 

5from fastapi import APIRouter, Depends, HTTPException, Request, status 

6from open_webui.constants import ERROR_MESSAGES 

7from open_webui.events import EVENTS, publish_event 

8from open_webui.internal.db import get_async_session 

9from open_webui.models.automations import ( 

10 AutomationForm, 

11 AutomationListResponse, 

12 AutomationModel, 

13 AutomationResponse, 

14 AutomationRunModel, 

15 AutomationRuns, 

16 Automations, 

17) 

18from open_webui.models.access_grants import AccessGrants, has_public_write_access_grant 

19from open_webui.models.channels import Channels 

20from open_webui.models.config import Config 

21from open_webui.models.folders import Folders 

22from open_webui.utils.access_control import has_permission 

23from open_webui.utils.auth import get_admin_user, get_verified_user 

24from open_webui.utils.automations import ( 

25 execute_automation, 

26 next_n_runs_ns, 

27 next_run_ns, 

28 rrule_interval_seconds, 

29 validate_rrule, 

30) 

31from sqlalchemy.ext.asyncio import AsyncSession 

32 

33log = logging.getLogger(__name__) 

34 

35router = APIRouter() 

36 

37PAGE_ITEM_COUNT = 30 

38 

39 

40############################ 

41# Helpers 

42############################ 

43 

44 

45async def check_automations_permission(request, user): 

46 config = await Config.get_many('automations.enable', 'user.permissions') 

47 if not config.get('automations.enable'): 47 ↛ 48line 47 didn't jump to line 48 because the condition on line 47 was never true

48 raise HTTPException( 

49 status_code=status.HTTP_403_FORBIDDEN, 

50 detail=ERROR_MESSAGES.UNAUTHORIZED, 

51 ) 

52 if user.role != 'admin' and not await has_permission( 52 ↛ 55line 52 didn't jump to line 55 because the condition on line 52 was never true

53 user.id, 'features.automations', config.get('user.permissions') 

54 ): 

55 raise HTTPException( 

56 status_code=status.HTTP_403_FORBIDDEN, 

57 detail=ERROR_MESSAGES.UNAUTHORIZED, 

58 ) 

59 

60 

61def check_automation_access(automation, user): 

62 if not automation or user.id != automation.user_id: 62 ↛ exitline 62 didn't return from function 'check_automation_access' because the condition on line 62 was always true

63 raise HTTPException( 

64 status_code=status.HTTP_404_NOT_FOUND, 

65 detail=ERROR_MESSAGES.NOT_FOUND, 

66 ) 

67 

68 

69async def check_automation_limits(request, user, rrule_str: str, db, is_create: bool = False): 

70 """Enforce global automation limits. Admins bypass all checks.""" 

71 if user.role == 'admin': 

72 return 

73 

74 # Max count (create only) 

75 if is_create: 

76 max_count = await Config.get('automations.max_count') 

77 if max_count: 

78 max_count = int(max_count) 

79 if max_count > 0 and await Automations.count_by_user(user.id, db=db) >= max_count: 

80 raise HTTPException( 

81 status_code=status.HTTP_403_FORBIDDEN, 

82 detail=ERROR_MESSAGES.AUTOMATION_LIMIT_EXCEEDED(max_count), 

83 ) 

84 

85 # Min interval (create + update) 

86 min_interval = await Config.get('automations.min_interval') 

87 if min_interval: 

88 min_interval = int(min_interval) 

89 if min_interval > 0: 

90 interval = await rrule_interval_seconds(rrule_str) 

91 if interval is not None and interval < min_interval: 

92 raise HTTPException( 

93 status_code=status.HTTP_400_BAD_REQUEST, 

94 detail=ERROR_MESSAGES.AUTOMATION_TOO_FREQUENT(min_interval), 

95 ) 

96 

97 

98async def check_automation_folder_access(folder_id: Optional[str], user, db: AsyncSession): 

99 if folder_id is None: 

100 return 

101 folder = await Folders.get_folder_by_id_and_user_id(folder_id, user.id, db=db) 

102 if not folder: 102 ↛ exitline 102 didn't return from function 'check_automation_folder_access' because the condition on line 102 was always true

103 raise HTTPException( 

104 status_code=status.HTTP_404_NOT_FOUND, 

105 detail=ERROR_MESSAGES.NOT_FOUND, 

106 ) 

107 

108 

109async def check_automation_channel_access(form_data: AutomationForm, user, db: AsyncSession): 

110 target = form_data.data.target 

111 if not target or target.type != 'channel': 

112 return 

113 

114 if not target.channel_id or not await Config.get('channels.enable'): 114 ↛ 120line 114 didn't jump to line 120 because the condition on line 114 was always true

115 raise HTTPException( 

116 status_code=status.HTTP_404_NOT_FOUND, 

117 detail=ERROR_MESSAGES.NOT_FOUND, 

118 ) 

119 

120 channel = await Channels.get_channel_by_id(target.channel_id, db=db) 

121 if not channel: 

122 raise HTTPException( 

123 status_code=status.HTTP_404_NOT_FOUND, 

124 detail=ERROR_MESSAGES.NOT_FOUND, 

125 ) 

126 

127 if user.role == 'admin': 

128 return 

129 if not await has_permission(user.id, 'features.channels', await Config.get('user.permissions')): 

130 raise HTTPException( 

131 status_code=status.HTTP_403_FORBIDDEN, 

132 detail=ERROR_MESSAGES.DEFAULT(), 

133 ) 

134 if channel.type in ['group', 'dm']: 

135 allowed = await Channels.is_user_channel_member(channel.id, user.id, db=db) 

136 else: 

137 allowed = has_public_write_access_grant(channel.access_grants) or await AccessGrants.has_access( 

138 user_id=user.id, resource_type='channel', resource_id=channel.id, permission='write', db=db 

139 ) 

140 if not allowed: 

141 raise HTTPException( 

142 status_code=status.HTTP_403_FORBIDDEN, 

143 detail=ERROR_MESSAGES.DEFAULT(), 

144 ) 

145 

146 

147async def enrich_automation(automation: AutomationModel, db: AsyncSession, tz: str = None) -> AutomationResponse: 

148 """Full enrichment for single-item views (includes next_runs computation).""" 

149 last_run = await AutomationRuns.get_latest(automation.id, db=db) 

150 return AutomationResponse( 

151 **automation.model_dump(), 

152 last_run=last_run, 

153 next_runs=await next_n_runs_ns(automation.data['rrule'], tz=tz), 

154 ) 

155 

156 

157############################ 

158# GetAutomationItems (paginated) 

159############################ 

160 

161 

162@router.get('/list') 

163async def get_automation_items( 

164 request: Request, 

165 query: Optional[str] = None, 

166 status: Optional[str] = None, 

167 folder_id: Optional[str] = None, 

168 page: Optional[int] = 1, 

169 user=Depends(get_verified_user), 

170 db: AsyncSession = Depends(get_async_session), 

171): 

172 await check_automations_permission(request, user) 

173 limit = PAGE_ITEM_COUNT 

174 page = max(1, page) 

175 skip = (page - 1) * limit 

176 

177 result = await Automations.search_automations( 

178 user_id=user.id, 

179 query=query, 

180 status=status, 

181 folder_id=folder_id, 

182 skip=skip, 

183 limit=limit, 

184 db=db, 

185 ) 

186 

187 # Batch-fetch latest runs in a single query instead of N+1 

188 ids = [item.id for item in result.items] 

189 latest_runs = await AutomationRuns.get_latest_batch(ids, db=db) if ids else {} 

190 

191 return { 

192 'items': [ 

193 AutomationResponse( 

194 **item.model_dump(), 

195 last_run=latest_runs.get(item.id), 

196 ) 

197 for item in result.items 

198 ], 

199 'total': result.total, 

200 } 

201 

202 

203############################ 

204# CreateNewAutomation 

205############################ 

206 

207 

208@router.post('/create', response_model=AutomationResponse) 

209async def create_new_automation( 

210 request: Request, 

211 form_data: AutomationForm, 

212 user=Depends(get_verified_user), 

213 db: AsyncSession = Depends(get_async_session), 

214): 

215 await check_automations_permission(request, user) 

216 await check_automation_folder_access(form_data.folder_id, user, db) 

217 await check_automation_channel_access(form_data, user, db) 

218 try: 

219 await validate_rrule(form_data.data.rrule, tz=user.timezone) 

220 except ValueError as e: 

221 raise HTTPException( 

222 status_code=status.HTTP_400_BAD_REQUEST, 

223 detail=str(e), 

224 ) 

225 

226 await check_automation_limits(request, user, form_data.data.rrule, db, is_create=True) 

227 

228 tz = user.timezone 

229 automation = await Automations.insert(user.id, form_data, await next_run_ns(form_data.data.rrule, tz=tz), db=db) 

230 response = await enrich_automation(automation, db, tz=tz) 

231 await publish_event( 

232 request, 

233 EVENTS.AUTOMATION_CREATED, 

234 actor=user, 

235 subject_id=automation.id, 

236 data={'name': automation.name, 'is_active': automation.is_active, 'folder_id': automation.folder_id}, 

237 ) 

238 return response 

239 

240 

241############################ 

242# GetAutomationById 

243############################ 

244 

245 

246@router.get('/{id}', response_model=AutomationResponse) 

247async def get_automation_by_id( 

248 request: Request, 

249 id: str, 

250 user=Depends(get_verified_user), 

251 db: AsyncSession = Depends(get_async_session), 

252): 

253 await check_automations_permission(request, user) 

254 automation = await Automations.get_by_id(id, db=db) 

255 check_automation_access(automation, user) 

256 return await enrich_automation(automation, db, tz=user.timezone) 

257 

258 

259############################ 

260# UpdateAutomationById 

261############################ 

262 

263 

264@router.post('/{id}/update', response_model=AutomationResponse) 

265async def update_automation_by_id( 

266 request: Request, 

267 id: str, 

268 form_data: AutomationForm, 

269 user=Depends(get_verified_user), 

270 db: AsyncSession = Depends(get_async_session), 

271): 

272 await check_automations_permission(request, user) 

273 automation = await Automations.get_by_id(id, db=db) 

274 check_automation_access(automation, user) 

275 await check_automation_folder_access(form_data.folder_id, user, db) 

276 await check_automation_channel_access(form_data, user, db) 

277 

278 try: 

279 await validate_rrule(form_data.data.rrule, tz=user.timezone) 

280 except ValueError as e: 

281 raise HTTPException( 

282 status_code=status.HTTP_400_BAD_REQUEST, 

283 detail=str(e), 

284 ) 

285 

286 await check_automation_limits(request, user, form_data.data.rrule, db, is_create=False) 

287 

288 tz = user.timezone 

289 updated = await Automations.update_by_id(id, form_data, await next_run_ns(form_data.data.rrule, tz=tz), db=db) 

290 response = await enrich_automation(updated, db, tz=tz) 

291 await publish_event( 

292 request, 

293 EVENTS.AUTOMATION_UPDATED, 

294 actor=user, 

295 subject_id=updated.id, 

296 data={'name': updated.name, 'is_active': updated.is_active, 'folder_id': updated.folder_id}, 

297 ) 

298 return response 

299 

300 

301############################ 

302# ToggleAutomationById 

303############################ 

304 

305 

306@router.post('/{id}/toggle', response_model=AutomationResponse) 

307async def toggle_automation_by_id( 

308 request: Request, 

309 id: str, 

310 user=Depends(get_verified_user), 

311 db: AsyncSession = Depends(get_async_session), 

312): 

313 await check_automations_permission(request, user) 

314 automation = await Automations.get_by_id(id, db=db) 

315 check_automation_access(automation, user) 

316 toggled = await Automations.toggle(id, await next_run_ns(automation.data['rrule'], tz=user.timezone), db=db) 

317 response = await enrich_automation(toggled, db, tz=user.timezone) 

318 await publish_event( 

319 request, 

320 EVENTS.AUTOMATION_ENABLED if toggled.is_active else EVENTS.AUTOMATION_DISABLED, 

321 actor=user, 

322 subject_id=toggled.id, 

323 subject_type='automation', 

324 data={'name': toggled.name}, 

325 ) 

326 return response 

327 

328 

329############################ 

330# RunAutomationById 

331############################ 

332 

333 

334@router.post('/{id}/run') 

335async def run_automation_by_id( 

336 request: Request, 

337 id: str, 

338 user=Depends(get_verified_user), 

339 db: AsyncSession = Depends(get_async_session), 

340): 

341 await check_automations_permission(request, user) 

342 automation = await Automations.get_by_id(id, db=db) 

343 check_automation_access(automation, user) 

344 asyncio.create_task(execute_automation(request.app, automation)) 

345 await publish_event( 

346 request, 

347 EVENTS.AUTOMATION_RUN_STARTED, 

348 actor=user, 

349 subject_id=automation.id, 

350 data={'name': automation.name}, 

351 ) 

352 return await enrich_automation(automation, db, tz=user.timezone) 

353 

354 

355############################ 

356# DeleteAutomationById 

357############################ 

358 

359 

360@router.delete('/{id}/delete') 

361async def delete_automation_by_id( 

362 request: Request, 

363 id: str, 

364 user=Depends(get_verified_user), 

365 db: AsyncSession = Depends(get_async_session), 

366): 

367 await check_automations_permission(request, user) 

368 automation = await Automations.get_by_id(id, db=db) 

369 check_automation_access(automation, user) 

370 await AutomationRuns.delete_by_automation(id, db=db) 

371 result = await Automations.delete(id, db=db) 

372 if result: 

373 await publish_event( 

374 request, 

375 EVENTS.AUTOMATION_DELETED, 

376 actor=user, 

377 subject_id=id, 

378 data={'name': automation.name}, 

379 ) 

380 return result 

381 

382 

383############################ 

384# GetAutomationRuns 

385############################ 

386 

387 

388@router.get('/{id}/runs', response_model=list[AutomationRunModel]) 

389async def get_automation_runs( 

390 request: Request, 

391 id: str, 

392 skip: int = 0, 

393 limit: int = 50, 

394 user=Depends(get_verified_user), 

395 db: AsyncSession = Depends(get_async_session), 

396): 

397 await check_automations_permission(request, user) 

398 automation = await Automations.get_by_id(id, db=db) 

399 check_automation_access(automation, user) 

400 return await AutomationRuns.get_by_automation(id, skip=skip, limit=limit, db=db)