mirror of
https://github.com/open-webui/open-webui.git
synced 2026-11-15 18:55:07 +03:00
Timothy Jaeryang Baek
656 lines
24 KiB
Python
656 lines
24 KiB
Python
"""
|
||
Automation utilities and unified scheduler.
|
||
|
||
RRULE helpers, scheduler worker loop, and execution logic.
|
||
Follows the utils/<feature>.py pattern (cf. utils/channels.py, utils/task.py).
|
||
|
||
The scheduler_worker_loop handles all time-based background work:
|
||
- Automation execution (claim_due → execute)
|
||
- Calendar event alerts (upcoming events → socket + webhook notifications)
|
||
- One-shot chat timers
|
||
|
||
Environment:
|
||
SCHEDULER_POLL_INTERVAL – seconds between polls (default: 10)
|
||
TIMER_POLL_INTERVAL – seconds between timer polls (default: 1)
|
||
CALENDAR_ALERT_LOOKAHEAD_MINUTES – default alert window (default: 5)
|
||
"""
|
||
|
||
import asyncio
|
||
import logging
|
||
import os
|
||
import random
|
||
import time
|
||
from datetime import timedelta
|
||
from typing import Optional
|
||
from uuid import uuid4
|
||
|
||
from fastapi import HTTPException, Request
|
||
from fastapi.security import HTTPAuthorizationCredentials
|
||
from open_webui.constants import ERROR_MESSAGES
|
||
from open_webui.events import EVENTS, build_event, publish_event
|
||
from open_webui.internal.db import get_async_db
|
||
from open_webui.models.automations import AutomationModel, AutomationRuns, Automations
|
||
from open_webui.models.chats import ChatForm, Chats
|
||
from open_webui.models.config import Config
|
||
from open_webui.models.folders import Folders
|
||
from open_webui.utils.access_control.folders import has_folder_write_access
|
||
from open_webui.models.messages import MessageForm
|
||
from open_webui.models.users import Users
|
||
from open_webui.utils.auth import create_token
|
||
from open_webui.utils.misc import parse_duration
|
||
from open_webui.utils.models import get_all_models
|
||
from open_webui.utils.recurrence import (
|
||
_resolve_tz,
|
||
next_n_runs_ns,
|
||
next_run_ns,
|
||
rrule_interval_seconds,
|
||
validate_rrule,
|
||
)
|
||
from open_webui.utils.task import prompt_template
|
||
from open_webui.utils.agents.discovery import get_models as get_agent_models
|
||
from open_webui.utils.terminals import initialize_terminal_directory, resolve_terminal_connection
|
||
from starlette.datastructures import Headers
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
SCHEDULER_POLL_INTERVAL = int(os.getenv('SCHEDULER_POLL_INTERVAL', os.getenv('AUTOMATION_POLL_INTERVAL', '10')))
|
||
TIMER_POLL_INTERVAL = int(os.getenv('TIMER_POLL_INTERVAL', '1'))
|
||
CALENDAR_ALERT_LOOKAHEAD_MINUTES = int(os.getenv('CALENDAR_ALERT_LOOKAHEAD_MINUTES', '10'))
|
||
|
||
|
||
# Worker Loop
|
||
|
||
|
||
# Keep the old name as an alias so any stale imports still work.
|
||
async def automation_worker_loop(app) -> None:
|
||
"""Deprecated alias — use scheduler_worker_loop."""
|
||
await scheduler_worker_loop(app)
|
||
|
||
|
||
async def scheduler_worker_loop(app) -> None:
|
||
"""Unified background scheduler for all time-based work.
|
||
|
||
Handles:
|
||
1. Automation execution (ENABLE_AUTOMATIONS)
|
||
2. Calendar event alerts (ENABLE_CALENDAR)
|
||
|
||
Runs on every instance. Poll interval is configurable via
|
||
SCHEDULER_POLL_INTERVAL env var (default: 10 seconds).
|
||
"""
|
||
log.info(
|
||
'Scheduler worker started (timer poll interval: %ss, scheduler poll interval: %ss)',
|
||
TIMER_POLL_INTERVAL,
|
||
SCHEDULER_POLL_INTERVAL,
|
||
)
|
||
next_scheduler_poll = 0.0
|
||
|
||
while True:
|
||
try:
|
||
now = time.monotonic()
|
||
# Timers
|
||
try:
|
||
from open_webui.utils.timers import claim_due_timers, execute_due_timer
|
||
|
||
for timer_id, claim_id in await claim_due_timers(int(time.time_ns()), limit=10):
|
||
asyncio.create_task(execute_due_timer(app, timer_id, claim_id))
|
||
except Exception:
|
||
log.exception('Scheduler: timer error')
|
||
|
||
if now < next_scheduler_poll:
|
||
await asyncio.sleep(max(1, TIMER_POLL_INTERVAL))
|
||
continue
|
||
# Jitter to spread automation/calendar load across instances; timers keep a tight poll.
|
||
next_scheduler_poll = now + SCHEDULER_POLL_INTERVAL + random.uniform(0, 2)
|
||
|
||
# Automations
|
||
if await Config.get('automations.enable'):
|
||
try:
|
||
async with get_async_db() as db:
|
||
batch = await Automations.claim_due(int(time.time_ns()), limit=10, db=db)
|
||
if batch:
|
||
log.info('Claimed %s due automation(s)', len(batch))
|
||
for automation in batch:
|
||
asyncio.create_task(execute_automation(app, automation))
|
||
except Exception:
|
||
log.exception('Scheduler: automation error')
|
||
|
||
# Calendar alerts
|
||
if await Config.get('calendar.enable'):
|
||
try:
|
||
await _check_calendar_alerts(app)
|
||
except Exception:
|
||
log.exception('Scheduler: calendar alert error')
|
||
|
||
except Exception:
|
||
log.exception('Scheduler worker error')
|
||
|
||
await asyncio.sleep(max(1, TIMER_POLL_INTERVAL))
|
||
|
||
|
||
# Execute
|
||
|
||
|
||
def _build_request(
|
||
app,
|
||
token: Optional[str] = None,
|
||
) -> Request:
|
||
"""Build a minimal ASGI Request for chat_completion.
|
||
|
||
Mirrors the mock-request pattern used in main.py lifespan
|
||
(model pre-fetch, tool server init) for consistency.
|
||
|
||
When token is provided, attach it as
|
||
request.state.token so session-auth tool servers and terminals can
|
||
authenticate headless scheduled runs as the automation owner.
|
||
"""
|
||
scope = {
|
||
'type': 'http',
|
||
'asgi': {'version': '3.0', 'spec_version': '2.0'},
|
||
'method': 'POST',
|
||
'path': '/api/v1/automations/internal',
|
||
'query_string': b'',
|
||
'headers': Headers({}).raw,
|
||
'client': ('', 0),
|
||
'server': ('127.0.0.1', 80),
|
||
'scheme': 'http',
|
||
'app': app,
|
||
}
|
||
request = Request(scope)
|
||
# Ensure request.state is initialized with required attributes
|
||
request.state.token = HTTPAuthorizationCredentials(scheme='Bearer', credentials=token) if token else None
|
||
request.state.enable_api_keys = False
|
||
return request
|
||
|
||
|
||
async def _resolve_model_defaults(app, model_id: str) -> dict:
|
||
models = getattr(app.state, 'MODELS', {})
|
||
model = models.get(model_id, {})
|
||
meta = model.get('info', {}).get('meta', {})
|
||
|
||
defaults = {
|
||
'tool_ids': list(meta.get('toolIds') or []),
|
||
'filter_ids': list(meta.get('defaultFilterIds') or []),
|
||
'terminal_id': meta.get('terminalId'),
|
||
}
|
||
defaults = {key: value for key, value in defaults.items() if value}
|
||
default_feature_ids = meta.get('defaultFeatureIds', [])
|
||
if not default_feature_ids:
|
||
return defaults
|
||
|
||
capabilities = meta.get('capabilities') or {}
|
||
features = {}
|
||
|
||
# code_interpreter is excluded: it requires the frontend event emitter
|
||
# and does not work in headless backend execution.
|
||
feature_checks = {
|
||
'web_search': await Config.get('web.search.enable'),
|
||
'image_generation': await Config.get('image_generation.enable'),
|
||
}
|
||
|
||
for feature_id in default_feature_ids:
|
||
if feature_id in feature_checks:
|
||
# Feature must be: in defaultFeatureIds + capability enabled + admin enabled
|
||
if capabilities.get(feature_id) and feature_checks[feature_id]:
|
||
features[feature_id] = True
|
||
|
||
if features:
|
||
defaults['features'] = features
|
||
return defaults
|
||
|
||
|
||
async def validate_automation_terminal(request, user, data: dict, automation_id: str | None = None):
|
||
terminal = data.get('terminal') or {}
|
||
model_id = data['model_id']
|
||
is_agent = model_id.startswith('agent:')
|
||
if not terminal.get('server_id'):
|
||
if is_agent:
|
||
raise HTTPException(400, 'Select a terminal for the agent.')
|
||
return
|
||
if terminal['server_id'].startswith('template:'):
|
||
from open_webui.utils.terminals import template_connection, require_model_access
|
||
|
||
_, template = await template_connection(user, terminal['server_id'])
|
||
scope = (template.get('data') or {}).get('scope')
|
||
if (data.get('target') or {}).get('type') == 'channel' and (is_agent or scope == 'chat'):
|
||
raise HTTPException(400, 'This terminal requires a chat destination.')
|
||
if scope == 'model':
|
||
await require_model_access(user, model_id, terminal['server_id'], request)
|
||
return
|
||
metadata = {'terminal_id': terminal['server_id'], 'model_id': model_id}
|
||
if automation_id:
|
||
metadata['automation_id'] = automation_id
|
||
if not await resolve_terminal_connection(request, user, metadata):
|
||
raise HTTPException(400, 'The configured terminal is unavailable for this automation.')
|
||
if is_agent:
|
||
if (data.get('target') or {}).get('type') == 'channel':
|
||
raise HTTPException(400, 'Agent automations require a chat destination.')
|
||
available = await get_agent_models(request, user, metadata, cwd=terminal.get('cwd'))
|
||
if not any(model['id'] == model_id for model in available):
|
||
raise HTTPException(400, 'Agent model is unavailable in the configured terminal.')
|
||
|
||
|
||
async def _execute_channel_automation(
|
||
app,
|
||
automation: AutomationModel,
|
||
user,
|
||
prompt: str,
|
||
model_id: str,
|
||
token: str,
|
||
) -> None:
|
||
target = automation.data.get('target') or {}
|
||
channel_id = target.get('channel_id')
|
||
if not channel_id or not await Config.get('channels.enable'):
|
||
raise ValueError('Channel not found')
|
||
|
||
from open_webui.utils.access_control import has_permission
|
||
|
||
if user.role != 'admin' and not await has_permission(
|
||
user.id, 'features.channels', await Config.get('user.permissions')
|
||
):
|
||
raise ValueError('Owner no longer permitted to use channels')
|
||
|
||
model = getattr(app.state, 'MODELS', {}).get(model_id, {})
|
||
request = _build_request(app, token=token)
|
||
|
||
from open_webui.routers.channels import new_message_handler
|
||
|
||
async with get_async_db() as db:
|
||
user_message, channel = await new_message_handler(
|
||
request,
|
||
channel_id,
|
||
MessageForm(
|
||
content=prompt,
|
||
data={},
|
||
meta={'automation_id': automation.id},
|
||
),
|
||
user,
|
||
db,
|
||
)
|
||
response_parent_id = (
|
||
user_message.parent_id
|
||
if user_message.parent_id
|
||
else (user_message.id if await Config.get('channels.model_response_mode', 'thread') == 'thread' else None)
|
||
)
|
||
assistant_message, channel = await new_message_handler(
|
||
request,
|
||
channel.id,
|
||
MessageForm(
|
||
parent_id=response_parent_id,
|
||
content='',
|
||
data={},
|
||
meta={
|
||
'automation_id': automation.id,
|
||
'model_id': model_id,
|
||
'model_name': model.get('name', model_id),
|
||
},
|
||
),
|
||
user,
|
||
db,
|
||
)
|
||
|
||
form_data = {
|
||
**await _resolve_model_defaults(app, model_id),
|
||
'model': model_id,
|
||
'messages': [
|
||
{
|
||
'role': 'system',
|
||
'content': f'You are {model.get("name", model_id)}, participating in a channel conversation. Be concise and conversational.',
|
||
},
|
||
{'role': 'user', 'content': f'{user.name if user else "User"}: {prompt}'},
|
||
],
|
||
'stream': True,
|
||
'chat_id': f'channel:{channel.id}',
|
||
'id': assistant_message.id,
|
||
'session_id': f'channel:{channel.id}',
|
||
'automation_id': automation.id,
|
||
'background_tasks': {},
|
||
}
|
||
terminal = automation.data.get('terminal') or {}
|
||
if terminal.get('server_id'):
|
||
form_data['terminal_id'] = terminal['server_id']
|
||
await initialize_terminal_directory(
|
||
request,
|
||
user,
|
||
{'chat_id': form_data['chat_id'], 'automation_id': automation.id, 'model_id': model_id},
|
||
terminal['server_id'],
|
||
terminal.get('cwd'),
|
||
)
|
||
await app.state.CHAT_COMPLETION_HANDLER(request, form_data, user=user)
|
||
|
||
from open_webui.socket.main import sio
|
||
|
||
await sio.emit(
|
||
'automation:result',
|
||
{
|
||
'automation_id': automation.id,
|
||
'name': automation.name,
|
||
'chat_id': f'channel:{channel.id}',
|
||
'message_id': assistant_message.id,
|
||
'status': 'success',
|
||
},
|
||
room=f'user:{automation.user_id}',
|
||
)
|
||
|
||
await _record_run(request, automation.id, 'success', chat_id=f'channel:{channel.id}')
|
||
await publish_event(
|
||
request,
|
||
EVENTS.AUTOMATION_RUN_COMPLETED,
|
||
actor=user,
|
||
subject_id=automation.id,
|
||
data={'name': automation.name, 'channel_id': channel.id, 'message_id': assistant_message.id},
|
||
)
|
||
|
||
|
||
async def execute_automation(app, automation: AutomationModel) -> None:
|
||
"""Execute an automation through the full chat completion pipeline.
|
||
|
||
Creates a real chat or channel message, then calls chat_completion exactly like the frontend:
|
||
session_id + chat_id + message_id → async task → pipeline handles everything
|
||
(filters, model params, knowledge/RAG, tools, DB saves, webhooks).
|
||
"""
|
||
chat = None
|
||
request = _build_request(app)
|
||
try:
|
||
user = await Users.get_user_by_id(automation.user_id)
|
||
if not user:
|
||
await _record_run(request, automation.id, 'error', error='User not found')
|
||
return
|
||
|
||
# Re-gate the rehydrated owner: a demoted/deactivated or de-permissioned owner must not run.
|
||
from open_webui.utils.access_control import has_permission
|
||
|
||
if user.role not in ('user', 'admin') or (
|
||
user.role != 'admin'
|
||
and not await has_permission(user.id, 'features.automations', await Config.get('user.permissions'))
|
||
):
|
||
error = 'Owner no longer permitted to run automations'
|
||
await _record_run(request, automation.id, 'error', error=error, actor=user)
|
||
return
|
||
|
||
prompt = await prompt_template(automation.data['prompt'], user)
|
||
model_id = automation.data['model_id']
|
||
try:
|
||
expires_delta = parse_duration(str(await Config.get('automations.auth_token_expires_in', '1h')))
|
||
except ValueError:
|
||
expires_delta = None
|
||
from open_webui.models.auths import Auths
|
||
|
||
auth = await Auths.get_auth_by_id(user.id)
|
||
if auth is None or not auth.active:
|
||
raise ValueError('Automation owner is no longer active')
|
||
token = create_token(
|
||
data={'id': user.id, 'typ': 'automation', 'session_stamp': auth.session_stamp},
|
||
expires_delta=expires_delta or timedelta(hours=1),
|
||
)
|
||
|
||
if not app.state.MODELS:
|
||
await get_all_models(_build_request(app, token=token), user=user)
|
||
|
||
request = _build_request(app, token=token)
|
||
await validate_automation_terminal(request, user, automation.data, automation.id)
|
||
|
||
target = automation.data.get('target') or {}
|
||
if target.get('type') == 'channel':
|
||
await _execute_channel_automation(app, automation, user, prompt, model_id, token)
|
||
return
|
||
|
||
# Runs land in the automation's folder while its creator can still write there.
|
||
folder_id = automation.folder_id
|
||
if folder_id and not await has_folder_write_access(automation.user_id, folder_id):
|
||
await Automations.clear_folder_ids(automation.user_id, [folder_id])
|
||
folder_id = None
|
||
|
||
# Generate proper UUIDs for messages (same as frontend)
|
||
user_msg_id = str(uuid4())
|
||
assistant_msg_id = str(uuid4())
|
||
|
||
chat_id = str(uuid4())
|
||
chat = await Chats.insert_new_chat(
|
||
chat_id,
|
||
automation.user_id,
|
||
ChatForm(
|
||
folder_id=folder_id,
|
||
chat={
|
||
'title': automation.name,
|
||
'models': [model_id],
|
||
'history': {
|
||
'currentId': assistant_msg_id,
|
||
'messages': {
|
||
user_msg_id: {
|
||
'id': user_msg_id,
|
||
'parentId': None,
|
||
'role': 'user',
|
||
'content': prompt,
|
||
'childrenIds': [assistant_msg_id],
|
||
'timestamp': int(time.time()),
|
||
'models': [model_id],
|
||
},
|
||
assistant_msg_id: {
|
||
'id': assistant_msg_id,
|
||
'parentId': user_msg_id,
|
||
'role': 'assistant',
|
||
'content': '',
|
||
'done': False,
|
||
'model': model_id,
|
||
'childrenIds': [],
|
||
'timestamp': int(time.time()),
|
||
},
|
||
},
|
||
},
|
||
'messages': [
|
||
{'role': 'user', 'content': prompt},
|
||
],
|
||
'meta': {'automation_id': automation.id},
|
||
},
|
||
),
|
||
internal_meta={'automation_id': automation.id},
|
||
)
|
||
|
||
if not chat:
|
||
error = 'Failed to create chat'
|
||
await _record_run(request, automation.id, 'error', error=error, actor=user)
|
||
return
|
||
|
||
# Notify frontend to refresh chat list
|
||
from open_webui.socket.main import sio
|
||
|
||
await sio.emit(
|
||
'events',
|
||
{
|
||
'chat_id': chat.id,
|
||
'message_id': user_msg_id,
|
||
'data': {'type': 'chat:list'},
|
||
},
|
||
room=f'user:{automation.user_id}',
|
||
)
|
||
|
||
# Build the same payload the frontend sends to /api/chat/completions
|
||
form_data = {
|
||
**await _resolve_model_defaults(app, model_id),
|
||
'model': model_id,
|
||
'messages': [{'role': 'user', 'content': prompt}],
|
||
'stream': True,
|
||
'chat_id': chat.id,
|
||
'id': assistant_msg_id,
|
||
'parent_id': None, # Root message (chat already created above)
|
||
'user_message': {
|
||
'id': user_msg_id,
|
||
'parentId': None,
|
||
'role': 'user',
|
||
'content': prompt,
|
||
},
|
||
'session_id': f'automation:{automation.id}',
|
||
'automation_id': automation.id,
|
||
'background_tasks': {},
|
||
}
|
||
# Match the chat UI's memory preference, including inherited defaults.
|
||
ui_settings = (user.settings.ui or {}) if user.settings else {}
|
||
memory = ui_settings.get('memory')
|
||
if memory is None:
|
||
defaults = await Config.get('ui.default_interface_settings') or {}
|
||
memory = defaults.get('memory')
|
||
if memory is None:
|
||
memory = await Config.get('memories.enable')
|
||
if memory:
|
||
form_data.setdefault('features', {})['memory'] = True
|
||
|
||
# Call the full chat completion pipeline (same as POST /api/chat/completions).
|
||
# The handler reference is stored on app.state to avoid circular imports.
|
||
request = _build_request(app, token=token)
|
||
terminal = automation.data.get('terminal') or {}
|
||
if terminal.get('server_id'):
|
||
form_data['terminal_id'] = terminal['server_id']
|
||
await initialize_terminal_directory(
|
||
request,
|
||
user,
|
||
{'chat_id': chat.id, 'automation_id': automation.id, 'model_id': model_id},
|
||
terminal['server_id'],
|
||
terminal.get('cwd'),
|
||
)
|
||
await app.state.CHAT_COMPLETION_HANDLER(request, form_data, user=user)
|
||
|
||
reply = await Chats.get_message_by_id_and_message_id(chat.id, assistant_msg_id)
|
||
if reply and reply.get('error'):
|
||
raise ValueError(str(reply['error']['content']))
|
||
if not reply or not reply.get('done'):
|
||
raise ValueError('Automation reply did not complete')
|
||
|
||
# Notify user
|
||
from open_webui.socket.main import sio
|
||
|
||
await sio.emit(
|
||
'automation:result',
|
||
{
|
||
'automation_id': automation.id,
|
||
'name': automation.name,
|
||
'chat_id': chat.id,
|
||
'status': 'success',
|
||
},
|
||
room=f'user:{automation.user_id}',
|
||
)
|
||
|
||
await _record_run(request, automation.id, 'success', chat_id=chat.id)
|
||
await publish_event(
|
||
request,
|
||
EVENTS.AUTOMATION_RUN_COMPLETED,
|
||
actor=user,
|
||
subject_id=automation.id,
|
||
data={'name': automation.name, 'chat_id': chat.id},
|
||
)
|
||
|
||
except (Exception, asyncio.CancelledError) as e:
|
||
log.exception(f'Automation {automation.id} failed')
|
||
error = 'Automation reply was cancelled' if isinstance(e, asyncio.CancelledError) else str(e)[:4000]
|
||
await _record_run(request, automation.id, 'error', chat_id=chat.id if chat else None, error=error)
|
||
if isinstance(e, asyncio.CancelledError):
|
||
raise
|
||
|
||
|
||
# Internals
|
||
|
||
|
||
async def _check_calendar_alerts(app) -> None:
|
||
"""Check for upcoming calendar events and send alert notifications.
|
||
|
||
De-duplication is DB-backed via meta.alerted_at — survives restarts
|
||
and works across multiple instances.
|
||
"""
|
||
from open_webui.models.calendar import CalendarEvents
|
||
from open_webui.socket.main import sio
|
||
|
||
now_ns = int(time.time_ns())
|
||
default_lookahead_ns = CALENDAR_ALERT_LOOKAHEAD_MINUTES * 60 * 1_000_000_000
|
||
# Grace window covers one poll cycle + jitter so "At time of event"
|
||
# alerts (alert_minutes=0) are not missed.
|
||
grace_ns = (SCHEDULER_POLL_INTERVAL + 5) * 1_000_000_000
|
||
|
||
async with get_async_db() as db:
|
||
upcoming = await CalendarEvents.get_upcoming_events(now_ns, default_lookahead_ns, grace_ns=grace_ns, db=db)
|
||
|
||
if not upcoming:
|
||
return
|
||
|
||
request = _build_request(app)
|
||
for event, user_tz in upcoming:
|
||
# Skip if already alerted for this start time
|
||
if event.meta and event.meta.get('alerted_at') == event.start_at:
|
||
continue
|
||
|
||
# Compute minutes until event starts
|
||
minutes_until = max(0, int((event.start_at - now_ns) / (60 * 1_000_000_000)))
|
||
|
||
alert_data = {
|
||
'event_id': event.id,
|
||
'title': event.title,
|
||
'description': event.description or '',
|
||
'start_at': event.start_at,
|
||
'minutes_until': minutes_until,
|
||
'calendar_id': event.calendar_id,
|
||
'location': event.location or '',
|
||
}
|
||
|
||
from open_webui.models.calendar import CalendarEvent
|
||
from sqlalchemy import select
|
||
|
||
async with get_async_db() as db:
|
||
current = await db.scalar(select(CalendarEvent).where(CalendarEvent.id == event.id).with_for_update())
|
||
if (
|
||
not current
|
||
or current.is_cancelled
|
||
or current.start_at != event.start_at
|
||
or (current.meta or {}).get('alerted_at') == event.start_at
|
||
):
|
||
continue
|
||
time_str = f'in {minutes_until} min' if minutes_until > 0 else 'now'
|
||
from open_webui.utils.inbox import consume_event
|
||
from open_webui.models.notifications import emit_changes
|
||
|
||
alert_event = build_event(
|
||
request,
|
||
EVENTS.CALENDAR_ALERT,
|
||
subject_id=event.id,
|
||
subject_type='calendar.event',
|
||
source='scheduler',
|
||
message=event.title,
|
||
event_id=f'calendar:{event.id}:{event.start_at}',
|
||
data={
|
||
**alert_data,
|
||
'user_id': event.user_id,
|
||
'starts_in': time_str,
|
||
'start_at': event.start_at,
|
||
'message': f'{event.title}: starting {time_str}',
|
||
},
|
||
)
|
||
current.meta = {**(current.meta or {}), 'alerted_at': event.start_at}
|
||
current.updated_at = time.time_ns()
|
||
recipients = await consume_event(alert_event, db=db)
|
||
await db.commit()
|
||
await emit_changes(recipients)
|
||
await publish_event(request, alert_event)
|
||
|
||
await sio.emit(
|
||
'events',
|
||
{
|
||
'data': {
|
||
'type': 'calendar:alert',
|
||
'data': alert_data,
|
||
},
|
||
},
|
||
room=f'user:{event.user_id}',
|
||
)
|
||
|
||
|
||
async def _record_run(
|
||
request: Request,
|
||
automation_id: str,
|
||
status: str,
|
||
chat_id: str = None,
|
||
error: str = None,
|
||
actor=None,
|
||
):
|
||
"""Insert a run record into automation_run."""
|
||
async with get_async_db() as db:
|
||
await AutomationRuns.insert(
|
||
automation_id, status, chat_id=chat_id, error=error, request=request, event_actor=actor, db=db
|
||
)
|