Coverage for open_webui/models/functions.py: 51%
264 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 05:07 +0000
1"""Function (filter/action/pipe) models, forms, and database operations."""
3from __future__ import annotations
5import logging
6import time
8# local imports
9from open_webui.internal.db import Base, JSONField, get_async_db_context
10from open_webui.models.users import User, UserResponse, Users, UserSettings
11from open_webui.utils.valves import decrypt_valves, encrypt_valves
12from pydantic import BaseModel, ConfigDict
13from sqlalchemy import BigInteger, Boolean, Column, Index, String, Text, delete, select, update
14from sqlalchemy.ext.asyncio import AsyncSession
16log = logging.getLogger(__name__)
19class Function(Base): # database table mapping
20 __tablename__ = 'function'
22 id = Column(String, primary_key=True, unique=True)
23 user_id = Column(String, index=True) # creator user id
24 name = Column(Text, nullable=False) # function identifier
25 type = Column(Text, nullable=False) # function type (pipe, filter, etc.)
26 content = Column(Text, nullable=True) # Python source code
27 meta = Column(JSONField, nullable=True) # function metadata
28 valves = Column(JSONField, nullable=True) # function configuration valves
29 is_active = Column(Boolean, default=False) # function activation status
30 is_global = Column(Boolean) # if True, applied to every chat automatically
31 updated_at = Column(BigInteger) # epoch seconds
32 created_at = Column(BigInteger) # epoch seconds
34 __table_args__ = (Index('is_global_idx', 'is_global'),) # speed up global-function lookups
37class FunctionMeta(BaseModel):
38 description: str | None = None
39 manifest: dict | None = {}
40 model_config = ConfigDict(extra='allow')
43class FunctionModel(BaseModel):
44 id: str
45 user_id: str | None = None # may be null for legacy/malformed records
46 name: str
47 type: str
48 content: str
49 meta: FunctionMeta
50 is_active: bool = False
51 is_global: bool = False
52 updated_at: int # timestamp in epoch
53 created_at: int # timestamp in epoch
55 model_config = ConfigDict(from_attributes=True) # allows ORM model binding
58# --- form / schema definitions ---
59class FunctionWithValvesModel(BaseModel):
60 id: str
61 user_id: str | None = None # may be null for legacy/malformed records
62 name: str
63 type: str
64 content: str
65 meta: FunctionMeta
66 valves: dict | None = None
67 is_active: bool = False
68 is_global: bool = False
69 updated_at: int # timestamp in epoch
70 created_at: int # timestamp in epoch
72 model_config = ConfigDict(from_attributes=True)
75####################
76# Forms
77####################
80class FunctionResponse(BaseModel):
81 id: str
82 user_id: str | None = None # may be null for legacy/malformed records
83 type: str
84 name: str
85 meta: FunctionMeta
86 is_active: bool
87 is_global: bool
88 updated_at: int # timestamp in epoch
89 created_at: int # timestamp in epoch
91 model_config = ConfigDict(from_attributes=True)
94class FunctionUserResponse(FunctionResponse):
95 user: UserResponse | None = None
98class FunctionForm(BaseModel):
99 id: str
100 name: str
101 content: str
102 meta: FunctionMeta
105class FunctionValves(BaseModel):
106 valves: dict | None = None
109class FunctionsTable:
110 async def insert_new_function(
111 self,
112 user_id: str,
113 type: str,
114 form_data: FunctionForm,
115 db: AsyncSession | None = None,
116 ) -> FunctionModel | None:
117 function = FunctionModel(
118 **{
119 **form_data.model_dump(),
120 'user_id': user_id,
121 'type': type,
122 'updated_at': int(time.time()),
123 'created_at': int(time.time()),
124 }
125 )
127 try:
128 async with get_async_db_context(db) as db:
129 result = Function(**function.model_dump())
130 db.add(result)
131 await db.commit()
132 if result:
133 return FunctionModel.model_validate(result)
134 else:
135 return None
136 except Exception as e:
137 log.exception(f'Error creating a new function: {e}')
138 return None
140 async def sync_functions(
141 self,
142 user_id: str,
143 functions: list[FunctionWithValvesModel],
144 db: AsyncSession | None = None,
145 ) -> list[FunctionWithValvesModel]:
146 # Synchronize functions by updating existing ones, inserting new ones,
147 # and removing those that are no longer present.
148 try:
149 async with get_async_db_context(db) as db:
150 # Get existing functions
151 result = await db.execute(select(Function))
152 existing_functions = result.scalars().all()
153 existing_ids = {func.id for func in existing_functions}
155 # Prepare a set of new function IDs
156 new_function_ids = {func.id for func in functions}
158 # Update or insert functions
159 for func in functions:
160 func_data = func.model_dump()
161 func_data['valves'] = encrypt_valves(func_data['valves']) if func_data.get('valves') else None
162 func_data['user_id'] = user_id
163 func_data['updated_at'] = int(time.time())
165 if func.id in existing_ids:
166 await db.execute(update(Function).filter_by(id=func.id).values(**func_data))
167 else:
168 new_func = Function(**func_data)
169 db.add(new_func)
171 # Remove functions that are no longer present
172 for func in existing_functions:
173 if func.id not in new_function_ids: 173 ↛ anywhereline 173 didn't jump anywhere: it always raised an exception.
174 await db.delete(func)
176 await db.commit()
178 result = await db.execute(select(Function))
179 return [FunctionModel.model_validate(func) for func in result.scalars().all()]
180 except Exception as e:
181 log.exception(f'Error syncing functions for user {user_id}: {e}')
182 return []
184 async def get_function_by_id(self, id: str, db: AsyncSession | None = None) -> FunctionModel | None:
185 try:
186 async with get_async_db_context(db) as db:
187 function = await db.get(Function, id)
188 return FunctionModel.model_validate(function) if function else None
189 except Exception:
190 return None
192 async def get_functions_by_ids(self, ids: list[str], db: AsyncSession | None = None) -> list[FunctionModel]:
193 """
194 Batch fetch multiple functions by their IDs in a single query.
195 Returns functions in the same order as the input IDs (None entries filtered out).
196 """
197 if not ids:
198 return []
199 try:
200 async with get_async_db_context(db) as db:
201 result = await db.execute(select(Function).filter(Function.id.in_(ids)))
202 functions = result.scalars().all()
203 # Create a dict for O(1) lookup
204 func_dict = {f.id: FunctionModel.model_validate(f) for f in functions}
205 # Return in original order, filtering out any not found
206 return [func_dict[id] for id in ids if id in func_dict]
207 except Exception:
208 return []
210 async def get_functions(
211 self, active_only=False, include_valves=False, db: AsyncSession | None = None
212 ) -> list[FunctionModel | FunctionWithValvesModel]:
213 async with get_async_db_context(db) as db:
214 if active_only:
215 result = await db.execute(select(Function).filter_by(is_active=True))
216 else:
217 result = await db.execute(select(Function))
219 functions = result.scalars().all()
221 if include_valves:
222 return [
223 FunctionWithValvesModel.model_validate(
224 {
225 **FunctionModel.model_validate(function).model_dump(),
226 'valves': decrypt_valves(function.valves),
227 }
228 )
229 for function in functions
230 ]
231 else:
232 return [FunctionModel.model_validate(function) for function in functions]
234 async def get_function_list(self, db: AsyncSession | None = None) -> list[FunctionUserResponse]:
235 async with get_async_db_context(db) as db:
236 result = await db.execute(select(Function).order_by(Function.updated_at.desc()))
237 functions = result.scalars().all()
238 user_ids = list(set(func.user_id for func in functions))
240 users = await Users.get_users_by_user_ids(user_ids, db=db) if user_ids else []
241 users_dict = {user.id: user for user in users}
243 return [
244 FunctionUserResponse.model_validate(
245 {
246 **FunctionResponse.model_validate(func).model_dump(),
247 'user': (
248 UserResponse(
249 id=users_dict[func.user_id].id,
250 name=users_dict[func.user_id].name,
251 role=users_dict[func.user_id].role,
252 email=users_dict[func.user_id].email,
253 ).model_dump()
254 if func.user_id in users_dict
255 else None
256 ),
257 }
258 )
259 for func in functions
260 ]
262 async def get_functions_by_type(
263 self, type: str, active_only=False, db: AsyncSession | None = None
264 ) -> list[FunctionModel]:
265 async with get_async_db_context(db) as db:
266 if active_only: 266 ↛ 269line 266 didn't jump to line 269 because the condition on line 266 was always true
267 result = await db.execute(select(Function).filter_by(type=type, is_active=True))
268 else:
269 result = await db.execute(select(Function).filter_by(type=type))
270 return [FunctionModel.model_validate(function) for function in result.scalars().all()]
272 async def get_global_filter_functions(self, db: AsyncSession | None = None) -> list[FunctionModel]:
273 async with get_async_db_context(db) as db:
274 result = await db.execute(select(Function).filter_by(type='filter', is_active=True, is_global=True))
275 return [FunctionModel.model_validate(function) for function in result.scalars().all()]
277 async def get_active_function_ids_by_type(
278 self, type: str, db: AsyncSession | None = None
279 ) -> list[tuple[str, bool]]:
280 """Return (id, is_global) for active functions without fetching plugin source."""
281 async with get_async_db_context(db) as db:
282 result = await db.execute(select(Function.id, Function.is_global).filter_by(type=type, is_active=True))
283 return [(id, bool(is_global)) for id, is_global in result.all()]
285 async def get_active_filter_ids(self, db: AsyncSession | None = None) -> list[tuple[str, bool]]:
286 """Return (id, is_global) for active filters without fetching plugin source."""
287 return await self.get_active_function_ids_by_type('filter', db=db)
289 async def get_global_action_functions(self, db: AsyncSession | None = None) -> list[FunctionModel]:
290 async with get_async_db_context(db) as db:
291 result = await db.execute(select(Function).filter_by(type='action', is_active=True, is_global=True))
292 return [FunctionModel.model_validate(function) for function in result.scalars().all()]
294 async def get_function_valves_by_id(self, id: str, db: AsyncSession | None = None) -> dict | None:
295 async with get_async_db_context(db) as db:
296 try:
297 result = await db.execute(select(Function.valves).filter_by(id=id))
298 return decrypt_valves(result.scalar_one_or_none())
299 except Exception as e:
300 log.exception(f'Error getting function valves by id {id}: {e}')
301 return None
303 async def get_function_valves_by_ids(self, ids: list[str], db: AsyncSession | None = None) -> dict[str, dict]:
304 """
305 Batch fetch valves for multiple functions in a single query.
306 Returns a dict mapping function_id -> valves dict.
307 Functions without valves are mapped to {}.
308 """
309 if not ids:
310 return {}
311 try:
312 async with get_async_db_context(db) as db:
313 result = await db.execute(select(Function.id, Function.valves).filter(Function.id.in_(ids)))
314 return {id: decrypt_valves(valves) for id, valves in result.all()}
315 except Exception as e:
316 log.exception(f'Error batch-fetching function valves: {e}')
317 return {}
319 async def update_function_valves_by_id(
320 self, id: str, valves: dict, db: AsyncSession | None = None
321 ) -> FunctionValves | None:
322 async with get_async_db_context(db) as db:
323 try:
324 function = await db.get(Function, id)
325 function.valves = encrypt_valves(valves)
326 function.updated_at = int(time.time())
327 await db.commit()
328 return FunctionModel.model_validate(function)
329 except Exception:
330 return None
332 async def update_function_metadata_by_id(
333 self, id: str, metadata: dict, db: AsyncSession | None = None
334 ) -> FunctionModel | None:
335 async with get_async_db_context(db) as db:
336 try:
337 function = await db.get(Function, id)
339 if function:
340 if function.meta:
341 function.meta = {**function.meta, **metadata}
342 else:
343 function.meta = metadata
345 function.updated_at = int(time.time())
346 await db.commit()
347 return FunctionModel.model_validate(function)
348 else:
349 return None
350 except Exception as e:
351 log.exception(f'Error updating function metadata by id {id}: {e}')
352 return None
354 async def get_user_valves_by_id_and_user_id(
355 self, id: str, user_id: str, db: AsyncSession | None = None
356 ) -> dict | None:
357 try:
358 async with get_async_db_context(db) as db:
359 result = await db.execute(select(User.settings).filter_by(id=user_id))
360 settings = result.scalar_one_or_none()
362 user_settings = UserSettings(**settings).model_dump() if settings else {}
364 # Check if user has "functions" and "valves" settings
365 if 'functions' not in user_settings:
366 user_settings['functions'] = {}
367 if 'valves' not in user_settings['functions']:
368 user_settings['functions']['valves'] = {}
370 return decrypt_valves(user_settings['functions']['valves'].get(id))
371 except Exception:
372 log.exception(f'Error getting user values by id {id} and user id {user_id}')
373 return None
375 async def update_user_valves_by_id_and_user_id(
376 self, id: str, user_id: str, valves: dict, db: AsyncSession | None = None
377 ) -> dict | None:
378 try:
379 user = await Users.get_user_by_id(user_id, db=db)
380 user_settings = user.settings.model_dump() if user.settings else {}
382 # Check if user has "functions" and "valves" settings
383 if 'functions' not in user_settings: 383 ↛ 385line 383 didn't jump to line 385 because the condition on line 383 was always true
384 user_settings['functions'] = {}
385 if 'valves' not in user_settings['functions']: 385 ↛ 388line 385 didn't jump to line 388 because the condition on line 385 was always true
386 user_settings['functions']['valves'] = {}
388 user_settings['functions']['valves'][id] = encrypt_valves(valves)
390 # Update the user settings in the database
391 await Users.update_user_by_id(user_id, {'settings': user_settings}, db=db)
393 return valves
394 except Exception as e:
395 log.exception(f'Error updating user valves by id {id} and user_id {user_id}: {e}')
396 return None
398 async def update_function_by_id(
399 self, id: str, updated: dict, db: AsyncSession | None = None
400 ) -> FunctionModel | None:
401 async with get_async_db_context(db) as db:
402 try:
403 await db.execute(
404 update(Function)
405 .filter_by(id=id)
406 .values(
407 **updated,
408 updated_at=int(time.time()),
409 )
410 )
411 await db.commit()
412 function = await db.get(Function, id)
413 return FunctionModel.model_validate(function) if function else None
414 except Exception:
415 return None
417 async def deactivate_all_functions(self, db: AsyncSession | None = None) -> bool | None:
418 async with get_async_db_context(db) as db:
419 try:
420 await db.execute(
421 update(Function).values(
422 is_active=False,
423 updated_at=int(time.time()),
424 )
425 )
426 await db.commit()
427 return True
428 except Exception:
429 return None
431 async def delete_function_by_id(self, id: str, db: AsyncSession | None = None) -> bool:
432 async with get_async_db_context(db) as db:
433 try:
434 await db.execute(delete(Function).filter_by(id=id))
435 await db.commit()
437 return True
438 except Exception:
439 return False
442Functions = FunctionsTable() # singleton functions engine