(self, element: "Element")
| 375 | ###### Elements ###### |
| 376 | @queue_until_user_message() |
| 377 | async def create_element(self, element: "Element"): |
| 378 | if self.show_logger: |
| 379 | logger.info(f"SQLAlchemy: create_element, element_id = {element.id}") |
| 380 | if not getattr(context.session.user, "id", None): |
| 381 | raise ValueError("No authenticated user in context") |
| 382 | if not self.storage_provider: |
| 383 | logger.warn( |
| 384 | "SQLAlchemy: create_element error. No blob_storage_client is configured!" |
| 385 | ) |
| 386 | return |
| 387 | if not element.for_id: |
| 388 | return |
| 389 | |
| 390 | content: Optional[Union[bytes, str]] = None |
| 391 | |
| 392 | if element.path: |
| 393 | async with aiofiles.open(element.path, "rb") as f: |
| 394 | content = await f.read() |
| 395 | elif element.url: |
| 396 | async with aiohttp.ClientSession() as session: |
| 397 | async with session.get(element.url) as response: |
| 398 | if response.status == 200: |
| 399 | content = await response.read() |
| 400 | else: |
| 401 | content = None |
| 402 | elif element.content: |
| 403 | content = element.content |
| 404 | else: |
| 405 | raise ValueError("Element url, path or content must be provided") |
| 406 | if content is None: |
| 407 | raise ValueError("Content is None, cannot upload file") |
| 408 | |
| 409 | context_user = context.session.user |
| 410 | |
| 411 | user_folder = getattr(context_user, "id", "unknown") |
| 412 | file_object_key = f"{user_folder}/{element.id}" + ( |
| 413 | f"/{element.name}" if element.name else "" |
| 414 | ) |
| 415 | |
| 416 | if not element.mime: |
| 417 | element.mime = "application/octet-stream" |
| 418 | |
| 419 | uploaded_file = await self.storage_provider.upload_file( |
| 420 | object_key=file_object_key, data=content, mime=element.mime, overwrite=True |
| 421 | ) |
| 422 | if not uploaded_file: |
| 423 | raise ValueError( |
| 424 | "SQLAlchemy Error: create_element, Failed to persist data in storage_provider" |
| 425 | ) |
| 426 | |
| 427 | element_dict: ElementDict = element.to_dict() |
| 428 | |
| 429 | element_dict["url"] = uploaded_file.get("url") |
| 430 | element_dict["objectKey"] = uploaded_file.get("object_key") |
| 431 | element_dict_cleaned = {k: v for k, v in element_dict.items() if v is not None} |
| 432 | |
| 433 | columns = ", ".join(f'"{column}"' for column in element_dict_cleaned.keys()) |
| 434 | placeholders = ", ".join(f":{column}" for column in element_dict_cleaned.keys()) |
nothing calls this directly
no test coverage detected