Phase 6: DMS & Attachments — Streaming, Deduplikation, API-Bereinigung

6.4 Upload streamen:
- attachment_service.save_attachment: Streamt in 1MB Chunks statt await file.read()
- routes/attachments.py: Uebergibt UploadFile direkt statt bytes

6.5 Download streamen:
- DMS preview_file: FileResponse fuer LocalStorage (automatisches Streaming)
- Kein storage.read() mehr fuer LocalStorage

6.6 Tenantlokale Deduplikation:
- DMS Upload: Prueft content_hash vor Erstellung, wiederverwendet existierendes File
- attachment_service: Dedup bereits vorhanden, jetzt mit Streaming kompatibel
- Migration 0098: Partial Unique Index (tenant_id, content_hash) WHERE content_hash IS NOT NULL AND deleted_at IS NULL

6.7 API-Ausgabe bereinigt:
- attachment_service: storage_path und content_hash aus API-Ausgaben entfernt
- DMS routes: content_hash aus 4 API-Endpunkten entfernt

Tests: 54/54 bestanden (17 Workspace + 13 API Token + 24 Command)

app/plugins/builtins/dms/routes.py

@@ -519,19 +519,34 @@ async def upload_file(
 
     mime_type = upload_data["mime_type"]
 
-    dms_file = DmsFile(
-        id=file_id,
-        tenant_id=tenant_id,
-        name=upload_data["filename"],
-        folder_id=fid,
-        uploaded_by=user_id,
-        mime_type=mime_type,
-        size_bytes=file_size,
-        storage_path=storage_path,
-        content_hash=content_hash,
+    # Tenant-local deduplication: check for existing file with same content_hash
+    existing_q = await db.execute(
+        select(DmsFile).where(
+            DmsFile.tenant_id == tenant_id,
+            DmsFile.content_hash == content_hash,
+            DmsFile.deleted_at.is_(None),
+        ).limit(1)
     )
-    db.add(dms_file)
-    await db.flush()
+    existing_file = existing_q.scalar_one_or_none()
+
+    if existing_file:
+        # Deduplicate: reuse existing file, remove the duplicate we just saved
+        await storage.delete(storage_path)
+        dms_file = existing_file
+    else:
+        dms_file = DmsFile(
+            id=file_id,
+            tenant_id=tenant_id,
+            name=upload_data["filename"],
+            folder_id=fid,
+            uploaded_by=user_id,
+            mime_type=mime_type,
+            size_bytes=file_size,
+            storage_path=storage_path,
+            content_hash=content_hash,
+        )
+        db.add(dms_file)
+        await db.flush()
 
     return {
         "id": str(dms_file.id),
@@ -540,7 +555,6 @@ async def upload_file(
         "uploaded_by": str(dms_file.uploaded_by),
         "mime_type": dms_file.mime_type,
         "size_bytes": dms_file.size_bytes,
-        "content_hash": dms_file.content_hash,
         "deleted_at": None,
         "created_at": dms_file.created_at.isoformat() if dms_file.created_at else None,
         "updated_at": dms_file.updated_at.isoformat() if dms_file.updated_at else None,
@@ -580,7 +594,6 @@ async def get_file(
         "uploaded_by": str(dms_file.uploaded_by),
         "mime_type": dms_file.mime_type,
         "size_bytes": dms_file.size_bytes,
-        "content_hash": dms_file.content_hash,
         "deleted_at": None,
         "created_at": dms_file.created_at.isoformat() if dms_file.created_at else None,
         "updated_at": dms_file.updated_at.isoformat() if dms_file.updated_at else None,
@@ -729,7 +742,6 @@ async def update_file(
         "uploaded_by": str(dms_file.uploaded_by),
         "mime_type": dms_file.mime_type,
         "size_bytes": dms_file.size_bytes,
-        "content_hash": dms_file.content_hash,
         "deleted_at": None,
         "created_at": dms_file.created_at.isoformat() if dms_file.created_at else None,
         "updated_at": dms_file.updated_at.isoformat() if dms_file.updated_at else None,
@@ -806,7 +818,6 @@ async def restore_file(
         "uploaded_by": str(dms_file.uploaded_by),
         "mime_type": dms_file.mime_type,
         "size_bytes": dms_file.size_bytes,
-        "content_hash": dms_file.content_hash,
         "deleted_at": None,
         "created_at": dms_file.created_at.isoformat() if dms_file.created_at else None,
         "updated_at": dms_file.updated_at.isoformat() if dms_file.updated_at else None,
@@ -853,16 +864,28 @@ async def preview_file(
             404, detail={"detail": "File not found on disk", "code": "file_missing"}
         )
 
-    content = await storage.read(dms_file.storage_path)
+    # Stream file directly from storage without loading into RAM
+    from fastapi.responses import FileResponse as FastApiFileResponse
+    import os as _os
 
-    def _stream():
-        yield content
-
-    return StreamingResponse(
-        _stream(),
-        media_type="application/pdf",
-        headers={"Content-Disposition": f'inline; filename="{dms_file.name}"'},
-    )
+    if isinstance(storage, LocalStorage):
+        # LocalStorage: use FileResponse for automatic streaming
+        full_path = storage._full_path(dms_file.storage_path)
+        return FastApiFileResponse(
+            path=full_path,
+            media_type="application/pdf",
+            headers={"Content-Disposition": f'inline; filename="{dms_file.name}"'},
+        )
+    else:
+        # S3 or other: fall back to read (TODO: implement S3 streaming)
+        content = await storage.read(dms_file.storage_path)
+        def _stream():
+            yield content
+        return StreamingResponse(
+            _stream(),
+            media_type="application/pdf",
+            headers={"Content-Disposition": f'inline; filename="{dms_file.name}"'},
+        )
 
 
 @router.post("/files/{file_id}/edit-session", dependencies=[Depends(require_permission("dms:write"))])

app/routes/attachments.py

@@ -34,13 +34,12 @@ async def upload_attachment(
     except ValueError:
         raise HTTPException(400, detail={"detail": "Invalid entity_id", "code": "invalid_id"}) from None
 
-    file_content = await file.read()
     mime_type = file.content_type or "application/octet-stream"
 
     try:
         return await attachment_service.save_attachment(
             db, tenant_id, user_id, entity_type, eid,
-            file.filename or "unknown", file_content, mime_type,
+            file.filename or "unknown", file, mime_type,
             is_system_admin=is_admin,
         )
     except PermissionError as e:

app/services/attachment_service.py

@@ -47,8 +47,6 @@ def _entity_attachment_to_dict(ea: EntityAttachment, dms_file: DmsFile | None =
         "filename": dms_file.name if dms_file else (ea.display_name or "unknown"),
         "mime_type": dms_file.mime_type if dms_file else "application/octet-stream",
         "file_size": dms_file.size_bytes if dms_file else 0,
-        "storage_path": dms_file.storage_path if dms_file else None,
-        "content_hash": dms_file.content_hash if dms_file else None,
         "uploaded_by": str(ea.created_by) if ea.created_by else None,
         "owner_id": str(ea.owner_id) if ea.owner_id else None,
         "created_at": ea.created_at.isoformat() if ea.created_at else None,
@@ -63,20 +61,60 @@ async def save_attachment(
     entity_type: str,
     entity_id: uuid.UUID,
     filename: str,
-    file_content: bytes,
+    file: Any,  # UploadFile or async iterator of chunks
     mime_type: str,
     is_system_admin: bool = False,
 ) -> dict[str, Any]:
-    """Save a file to DMS and create an entity_attachments reference."""
-    # File size limit
-    if len(file_content) > MAX_FILE_SIZE:
-        raise ValueError(f"File too large: {len(file_content)} bytes (max {MAX_FILE_SIZE})")
+    """Save a file to DMS and create an entity_attachments reference.
+
+    Streams the file in chunks to avoid loading entire file into RAM.
+    """
+    import hashlib
+    from app.core.storage import get_storage_backend
+
+    # Generate unique filename and storage path
+    unique_filename = _generate_unique_filename(filename)
+    storage_path = f"attachments/{entity_type}/{entity_id}/{unique_filename}"
+
+    # Stream file to storage — compute hash and size during streaming
+    sha256 = hashlib.sha256()
+    file_size = 0
+    CHUNK_SIZE = 1024 * 1024  # 1MB chunks
+
+    async def chunk_stream():
+        nonlocal file_size
+        if hasattr(file, 'read'):
+            # UploadFile object
+            while True:
+                chunk = await file.read(CHUNK_SIZE)
+                if not chunk:
+                    break
+                file_size += len(chunk)
+                sha256.update(chunk)
+                yield chunk
+        else:
+            # Already bytes (backward compat)
+            nonlocal_bytes = file if isinstance(file, bytes) else b''.join([c async for c in file])
+            file_size = len(nonlocal_bytes)
+            sha256.update(nonlocal_bytes)
+            yield nonlocal_bytes
+
+    # Check file size limit during streaming
+    # (we check after streaming — for true streaming we'd need a wrapper)
+    # For now, stream and check size after
+    storage = get_storage_backend()
+    await storage.save_stream(storage_path, chunk_stream())
+
+    if file_size > MAX_FILE_SIZE:
+        await storage.delete(storage_path)
+        raise ValueError(f"File too large: {file_size} bytes (max {MAX_FILE_SIZE})")
 
     # Check for blocked file types
     import os as _os
     _BLOCKED = {".exe", ".bat", ".cmd", ".sh", ".jar", ".com", ".scr", ".msi", ".dll", ".vbs", ".ps1", ".app", ".bin", ".reg", ".inf"}
     _ext = _os.path.splitext(filename)[1].lower()
     if _ext in _BLOCKED:
+        await storage.delete(storage_path)
         raise ValueError(f"File type not allowed: {_ext}")
 
     # Check access on parent entity
@@ -85,14 +123,10 @@ async def save_attachment(
             db, entity_type, entity_id, user_id, tenant_id, "write", is_system_admin
         )
         if not has_access:
+            await storage.delete(storage_path)
             raise PermissionError(f"No write access to {entity_type} {entity_id}")
 
-    # Generate unique filename and storage path
-    unique_filename = _generate_unique_filename(filename)
-    storage_path = f"attachments/{entity_type}/{entity_id}/{unique_filename}"
-
-    # Calculate content hash for deduplication (tenant-local)
-    content_hash = hashlib.sha256(file_content).hexdigest()
+    content_hash = sha256.hexdigest()
 
     # Check for existing DMS file with same hash in same tenant (deduplication)
     existing_file = await db.execute(
@@ -107,19 +141,16 @@ async def save_attachment(
     if existing_dms_file:
         # Deduplicate: reuse existing DMS file, just create new reference
         dms_file = existing_dms_file
+        await storage.delete(storage_path)  # Remove the duplicate we just saved
     else:
-        # Save file via storage backend
-        storage = get_storage_backend()
-        await storage.save(storage_path, file_content)
-
-        # Create DMS File record
+        # File already streamed to storage — create DMS File record
         dms_file = DmsFile(
             tenant_id=tenant_id,
             name=filename,
             folder_id=None,  # Attachments don't go in DMS folders
             uploaded_by=user_id,
             mime_type=mime_type,
-            size_bytes=len(file_content),
+            size_bytes=file_size,
             storage_path=storage_path,
             content_hash=content_hash,
             owner_id=user_id,