Coverage for open_webui/models/files.py: 59%

1"""File upload 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.utils.misc import sanitize_metadata

11from pydantic import BaseModel, ConfigDict, model_validator

12from sqlalchemy import JSON, BigInteger, Column, String, Text, delete, func, select

13from sqlalchemy.ext.asyncio import AsyncSession

15log = logging.getLogger(__name__)

18class File(Base): # uploaded file record

19 __tablename__ = 'file'

20 id = Column(String, primary_key=True, unique=True)

21 user_id = Column(String, index=True) # owner user id

22 hash = Column(Text, nullable=True)

24 filename = Column(Text) # original upload filename

25 path = Column(Text, nullable=True)

27 data = Column(JSON, nullable=True)

28 meta = Column(JSON, nullable=True)

30 created_at = Column(BigInteger, index=True) # upload timestamp

31 updated_at = Column(BigInteger)

34class FileModel(BaseModel):

35 model_config = ConfigDict(from_attributes=True)

37 id: str

38 user_id: str

39 hash: str | None = None

41 filename: str

42 path: str | None = None

44 data: dict | None = None

45 meta: dict | None = None

47 created_at: int | None # timestamp in epoch

48 updated_at: int | None # timestamp in epoch

51# --- metadata structures ---

52class FileMeta(BaseModel):

53 name: str | None = None

54 content_type: str | None = None

55 size: int | None = None

57 model_config = ConfigDict(extra='allow')

59 @model_validator(mode='before')

60 @classmethod

61 def sanitize_meta(cls, data):

62 """Sanitize metadata fields to handle malformed legacy data."""

63 if not isinstance(data, dict): 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true

64 return data

66 # Handle content_type that may be a list like ['application/pdf', None]

67 content_type = data.get('content_type')

68 if isinstance(content_type, list): 68 ↛ 70line 68 didn't jump to line 70 because the condition on line 68 was never true

69 # Extract first non-None string value

70 data['content_type'] = next((item for item in content_type if isinstance(item, str)), None)

71 elif content_type is not None and not isinstance(content_type, str): 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true

72 data['content_type'] = None

74 return data

77class FileModelResponse(BaseModel):

78 id: str

79 user_id: str

80 hash: str | None = None

82 filename: str

83 data: dict | None = None

84 meta: FileMeta | None = None

86 created_at: int # timestamp in epoch

87 updated_at: int | None = None # timestamp in epoch, optional for legacy files

89 model_config = ConfigDict(extra='allow')

92class FileMetadataResponse(BaseModel):

93 id: str

94 hash: str | None = None

95 meta: dict | None = None

96 created_at: int # timestamp in epoch

97 updated_at: int # timestamp in epoch

100class FileListResponse(BaseModel):

101 items: list[FileModelResponse]

102 total: int

103

104

105class FileForm(BaseModel):

106 id: str

107 hash: str | None = None

108 filename: str

109 path: str

110 data: dict = {}

111 meta: dict = {}

112

113

114class FileUpdateForm(BaseModel):

115 hash: str | None = None

116 data: dict | None = None

117 meta: dict | None = None

118

119

120class FilesTable:

121 async def insert_new_file(

122 self, user_id: str, form_data: FileForm, db: AsyncSession | None = None

123 ) -> FileModel | None:

124 async with get_async_db_context(db) as db:

125 file_data = form_data.model_dump()

126

127 # Sanitize meta to remove non-JSON-serializable objects

128 # (e.g. callable tool functions, MCP client instances from middleware)

129 if file_data.get('meta'): 129 ↛ 132line 129 didn't jump to line 132 because the condition on line 129 was always true

130 file_data['meta'] = sanitize_metadata(file_data['meta'])

131

132 file = FileModel(

133 **{

134 **file_data,

135 'user_id': user_id,

136 'created_at': int(time.time()),

137 'updated_at': int(time.time()),

138 }

139 )

140

141 try:

142 result = File(**file.model_dump())

143 db.add(result)

144 await db.commit()

145 if result:

146 return FileModel.model_validate(result)

147 else:

148 return None

149 except Exception as e:

150 log.exception(f'Error inserting a new file: {e}')

151 return None # insertion failed

152

153 async def get_file_by_id(

154 self,

155 id: str,

156 db: AsyncSession | None = None,

157 ) -> FileModel | None:

158 """Look up a file by its primary key."""

159 try:

160 async with get_async_db_context(db) as db:

161 file = await db.get(File, id)

162 if not file:

163 return None

164 return FileModel.model_validate(file)

165 except Exception:

166 return None

167

168 async def get_file_by_id_and_user_id(

169 self, id: str, user_id: str, db: AsyncSession | None = None

170 ) -> FileModel | None:

171 async with get_async_db_context(db) as db:

172 try:

173 result = await db.execute(select(File).filter_by(id=id, user_id=user_id))

174 file = result.scalars().first()

175 if file:

176 return FileModel.model_validate(file)

177 else:

178 return None

179 except Exception:

180 return None

181

182 async def get_file_metadata_by_id(self, id: str, db: AsyncSession | None = None) -> FileMetadataResponse | None:

183 async with get_async_db_context(db) as db:

184 try:

185 file = await db.get(File, id)

186 if not file:

187 return None

188 return FileMetadataResponse(

189 id=file.id,

190 hash=file.hash,

191 meta=file.meta,

192 created_at=file.created_at,

193 updated_at=file.updated_at,

194 )

195 except Exception:

196 return None

197

198 async def get_files(self, db: AsyncSession | None = None) -> list[FileModel]:

199 async with get_async_db_context(db) as db:

200 result = await db.execute(select(File))

201 return [FileModel.model_validate(file) for file in result.scalars().all()]

202

203 async def count_files_by_user_id(

204 self,

205 user_id: str | None = None,

206 db: AsyncSession | None = None,

207 ) -> int:

208 async with get_async_db_context(db) as db:

209 stmt = select(func.count(File.id))

210 if user_id: 210 ↛ 211line 210 didn't jump to line 211 because the condition on line 210 was never true

211 stmt = stmt.filter_by(user_id=user_id)

212 result = await db.execute(stmt)

213 return result.scalar() or 0

214

215 async def check_access_by_user_id(self, id, user_id, permission='write', db: AsyncSession | None = None) -> bool:

216 file = await self.get_file_by_id(id, db=db)

217 if not file:

218 return False

219 if file.user_id == user_id:

220 return True

221 # Implement additional access control logic here as needed

222 return False

223

224 async def get_files_by_ids(self, ids: list[str], db: AsyncSession | None = None) -> list[FileModel]:

225 async with get_async_db_context(db) as db:

226 result = await db.execute(select(File).filter(File.id.in_(ids)).order_by(File.updated_at.desc()))

227 return [FileModel.model_validate(file) for file in result.scalars().all()]

228

229 async def get_file_metadatas_by_ids(

230 self, ids: list[str], db: AsyncSession | None = None

231 ) -> list[FileMetadataResponse]:

232 async with get_async_db_context(db) as db:

233 result = await db.execute(

234 select(File.id, File.hash, File.meta, File.created_at, File.updated_at)

235 .filter(File.id.in_(ids))

236 .order_by(File.updated_at.desc())

237 )

238 return [

239 FileMetadataResponse(

240 id=row.id,

241 hash=row.hash,

242 meta=row.meta,

243 created_at=row.created_at,

244 updated_at=row.updated_at,

245 )

246 for row in result.all()

247 ]

248

249 async def get_files_by_user_id(self, user_id: str, db: AsyncSession | None = None) -> list[FileModel]:

250 async with get_async_db_context(db) as db:

251 result = await db.execute(select(File).filter_by(user_id=user_id))

252 return [FileModel.model_validate(file) for file in result.scalars().all()]

253

254 async def get_file_list(

255 self,

256 user_id: str | None = None,

257 skip: int = 0,

258 limit: int = 50,

259 db: AsyncSession | None = None,

260 ) -> 'FileListResponse':

261 async with get_async_db_context(db) as db:

262 stmt = select(File)

263 if user_id: 263 ↛ 264line 263 didn't jump to line 264 because the condition on line 263 was never true

264 stmt = stmt.filter_by(user_id=user_id)

265

266 count_result = await db.execute(select(func.count()).select_from(stmt.subquery()))

267 total = count_result.scalar()

268

269 result = await db.execute(stmt.order_by(File.updated_at.desc(), File.id.desc()).offset(skip).limit(limit))

270 items = [FileModelResponse.model_validate(file, from_attributes=True) for file in result.scalars().all()]

271

272 return FileListResponse(items=items, total=total)

273

274 @staticmethod

275 def _glob_to_like_pattern(glob: str) -> str:

276 """

277 Convert a glob/fnmatch pattern to a SQL LIKE pattern.

278

279 Escapes SQL special characters and converts glob wildcards:

280 - `*` becomes `%` (match any sequence of characters)

281 - `?` becomes `_` (match exactly one character)

282

283 Args:

284 glob: A glob pattern (e.g., "*.txt", "file?.doc")

285

286 Returns:

287 A SQL LIKE compatible pattern with proper escaping.

288 """

289 # Escape SQL special characters first, then convert glob wildcards

290 pattern = glob.replace('\\', '\\\\')

291 pattern = pattern.replace('%', '\\%')

292 pattern = pattern.replace('_', '\\_')

293 pattern = pattern.replace('*', '%')

294 pattern = pattern.replace('?', '_')

295 return pattern

296

297 async def search_files(

298 self,

299 user_id: str | None = None,

300 filename: str = '*',

301 skip: int = 0,

302 limit: int = 100,

303 db: AsyncSession | None = None,

304 ) -> list[FileModel]:

305 """

306 Search files with glob pattern matching, optional user filter, and pagination.

307

308 Args:

309 user_id: Filter by user ID. If None, returns files for all users.

310 filename: Glob pattern to match filenames (e.g., "*.txt"). Default "*" matches all.

311 skip: Number of results to skip for pagination.

312 limit: Maximum number of results to return.

313 db: Optional database session.

314

315 Returns:

316 List of matching FileModel objects, ordered by created_at descending.

317 """

318 async with get_async_db_context(db) as db:

319 stmt = select(File)

320

321 if user_id: 321 ↛ 322line 321 didn't jump to line 322 because the condition on line 321 was never true

322 stmt = stmt.filter_by(user_id=user_id)

323

324 pattern = self._glob_to_like_pattern(filename)

325 if pattern != '%': 325 ↛ 328line 325 didn't jump to line 328 because the condition on line 325 was always true

326 stmt = stmt.filter(File.filename.ilike(pattern, escape='\\'))

327

328 result = await db.execute(stmt.order_by(File.created_at.desc(), File.id.desc()).offset(skip).limit(limit))

329 return [FileModel.model_validate(file) for file in result.scalars().all()]

330

331 async def update_file_by_id(

332 self, id: str, form_data: FileUpdateForm, db: AsyncSession | None = None

333 ) -> FileModel | None:

334 async with get_async_db_context(db) as db:

335 try:

336 result = await db.execute(select(File).filter_by(id=id))

337 file = result.scalars().first()

338

339 if form_data.hash is not None:

340 file.hash = form_data.hash

341

342 if form_data.data is not None:

343 file.data = {**(file.data if file.data else {}), **form_data.data}

344

345 if form_data.meta is not None:

346 file.meta = {**(file.meta if file.meta else {}), **form_data.meta}

347

348 file.updated_at = int(time.time())

349 await db.commit()

350 return FileModel.model_validate(file)

351 except Exception as e:

352 log.exception(f'Error updating file completely by id: {e}')

353 return None

354

355 async def update_file_hash_by_id(

356 self, id: str, hash: str | None, db: AsyncSession | None = None

357 ) -> FileModel | None:

358 async with get_async_db_context(db) as db:

359 try:

360 result = await db.execute(select(File).filter_by(id=id))

361 file = result.scalars().first()

362 file.hash = hash

363 file.updated_at = int(time.time())

364 await db.commit()

365

366 return FileModel.model_validate(file)

367 except Exception:

368 return None

369

370 async def update_file_data_by_id(self, id: str, data: dict, db: AsyncSession | None = None) -> FileModel | None:

371 async with get_async_db_context(db) as db:

372 try:

373 result = await db.execute(select(File).filter_by(id=id))

374 file = result.scalars().first()

375 file.data = {**(file.data if file.data else {}), **data}

376 file.updated_at = int(time.time())

377 await db.commit()

378 return FileModel.model_validate(file)

379 except Exception as e:

380 return None

381

382 async def update_file_metadata_by_id(self, id: str, meta: dict, db: AsyncSession | None = None) -> FileModel | None:

383 async with get_async_db_context(db) as db:

384 try:

385 result = await db.execute(select(File).filter_by(id=id))

386 file = result.scalars().first()

387 file.meta = {**(file.meta if file.meta else {}), **meta}

388 file.updated_at = int(time.time())

389 await db.commit()

390 return FileModel.model_validate(file)

391 except Exception:

392 return None

393

394 async def update_file_name_by_id(self, id: str, name: str, db: AsyncSession | None = None) -> FileModel | None:

395 async with get_async_db_context(db) as db:

396 try:

397 result = await db.execute(select(File).filter_by(id=id))

398 file = result.scalars().first()

399 file.filename = name

400 file.meta = {**(file.meta if file.meta else {}), 'name': name}

401 file.updated_at = int(time.time())

402 await db.commit()

403 return FileModel.model_validate(file)

404 except Exception:

405 return None

406

407 async def get_pending_files_for_knowledge(

408 self, knowledge_id: str, db: AsyncSession | None = None

409 ) -> list[FileModelResponse]:

410 """Return files still being processed for this knowledge base.

411

412 These are files uploaded with ``meta.data.knowledge_id`` set, whose

413 ``data.status`` is still ``pending`` or ``processing``, and which

414 have not yet been added to the ``knowledge_file`` join table.

415

416 The JSON subscript syntax (``Column['key']['subkey'].as_string()``)

417 is supported by both SQLite (``json_extract``) and PostgreSQL

418 (``->>``/``->``).

419 """

420 async with get_async_db_context(db) as db:

421 try:

422 # Lazy import to avoid circular dependency

423 from open_webui.models.knowledge import KnowledgeFile

424

425 # Subquery: file IDs already linked to this knowledge base

426 linked_ids = (

427 select(KnowledgeFile.file_id).filter(KnowledgeFile.knowledge_id == knowledge_id).correlate(None)

428 )

429

430 stmt = (

431 select(File)

432 .filter(

433 File.meta['data']['knowledge_id'].as_string() == knowledge_id,

434 File.data['status'].as_string().in_(['pending', 'processing']),

435 File.id.notin_(linked_ids),

436 )

437 .order_by(File.created_at.desc())

438 )

439 result = await db.execute(stmt)

440 return [FileModelResponse.model_validate(f, from_attributes=True) for f in result.scalars().all()]

441 except Exception as e:

442 log.warning(f'Error fetching pending files for knowledge {knowledge_id}: {e}')

443 return []

444

445 async def delete_file_by_id(self, id: str, db: AsyncSession | None = None) -> bool:

446 async with get_async_db_context(db) as db:

447 try:

448 await db.execute(delete(File).filter_by(id=id))

449 await db.commit()

450

451 return True

452 except Exception:

453 return False

454

455 async def delete_all_files(self, db: AsyncSession | None = None) -> bool:

456 async with get_async_db_context(db) as db:

457 try:

458 await db.execute(delete(File))

459 await db.commit()

460

461 return True

462 except Exception:

463 return False

464

465

466Files = FilesTable() # singleton files repository