phase4: entity_attachments table + DMS unified storage + attachment service rewritten + download via DMS
This commit is contained in:
Agent Zero
@@ -0,0 +1,62 @@
|
||||
"""Create entity_attachments table — references DMS files.
|
||||
|
||||
Instead of storing files in a separate attachment storage path,
|
||||
all files go through the DMS (files table) and entity_attachments
|
||||
just references the DMS file with entity_type/entity_id.
|
||||
|
||||
This unifies the storage layer: one upload path, one download path,
|
||||
one permission model, one deduplication (content_hash).
|
||||
|
||||
Revision ID: 0071
|
||||
Revises: 0070
|
||||
"""
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy.dialects.postgresql import UUID as PGUUID
|
||||
|
||||
revision = "0071"
|
||||
down_revision = "0070"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_table(
|
||||
"entity_attachments",
|
||||
sa.Column("id", PGUUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
|
||||
sa.Column("tenant_id", PGUUID(as_uuid=True), sa.ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False),
|
||||
sa.Column("entity_type", sa.String(50), nullable=False),
|
||||
sa.Column("entity_id", PGUUID(as_uuid=True), nullable=False),
|
||||
sa.Column("dms_file_id", PGUUID(as_uuid=True), sa.ForeignKey("files.id", ondelete="RESTRICT"), nullable=False),
|
||||
sa.Column("category", sa.String(50), nullable=True),
|
||||
sa.Column("display_name", sa.String(255), nullable=True),
|
||||
sa.Column("owner_id", PGUUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
|
||||
sa.Column("created_by", PGUUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
|
||||
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()"), nullable=False),
|
||||
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()"), nullable=False),
|
||||
sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
|
||||
)
|
||||
|
||||
op.create_index("ix_entity_attachments_entity", "entity_attachments", ["entity_type", "entity_id", "tenant_id"])
|
||||
op.create_index("ix_entity_attachments_tenant", "entity_attachments", ["tenant_id"])
|
||||
op.create_index("ix_entity_attachments_dms_file", "entity_attachments", ["dms_file_id"])
|
||||
op.create_index("ix_entity_attachments_owner", "entity_attachments", ["owner_id"])
|
||||
|
||||
# Enable RLS on entity_attachments (tenant isolation)
|
||||
op.execute("ALTER TABLE entity_attachments ENABLE ROW LEVEL SECURITY")
|
||||
op.execute(
|
||||
"CREATE POLICY entity_attachments_tenant_isolation ON entity_attachments "
|
||||
"FOR ALL "
|
||||
"USING (tenant_id = current_setting('app.current_tenant_id', true)::uuid) "
|
||||
"WITH CHECK (tenant_id = current_setting('app.current_tenant_id', true)::uuid)"
|
||||
)
|
||||
|
||||
# Grant to crm_api and crm_worker
|
||||
op.execute("GRANT SELECT, INSERT, UPDATE, DELETE ON entity_attachments TO crm_api, crm_worker")
|
||||
op.execute("GRANT USAGE ON SCHEMA public TO crm_api, crm_worker")
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.execute("DROP POLICY IF EXISTS entity_attachments_tenant_isolation ON entity_attachments")
|
||||
op.drop_table("entity_attachments")
|
||||
@@ -84,3 +84,4 @@ __all__ = [
|
||||
"WorkflowStepHistory",
|
||||
"SavedView",
|
||||
]
|
||||
from app.models.entity_attachment import EntityAttachment # noqa: F401
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
"""EntityAttachment model — references DMS files for any entity."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import DateTime, ForeignKey, Index, String
|
||||
from sqlalchemy.dialects.postgresql import UUID as PGUUID
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from app.core.db import Base, TenantMixin
|
||||
from app.models.owned_mixin import OwnedMixin
|
||||
|
||||
|
||||
class EntityAttachment(Base, TenantMixin, OwnedMixin):
|
||||
"""Links a DMS file to any entity (contact, address, invoice, etc.).
|
||||
|
||||
The actual file is stored in the DMS (files table).
|
||||
This table only holds the reference + category + display_name.
|
||||
"""
|
||||
|
||||
__tablename__ = "entity_attachments"
|
||||
__table_args__ = (
|
||||
Index("ix_entity_attachments_entity", "entity_type", "entity_id", "tenant_id"),
|
||||
Index("ix_entity_attachments_dms_file", "dms_file_id"),
|
||||
)
|
||||
|
||||
id: Mapped[uuid.UUID] = mapped_column(
|
||||
PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
|
||||
)
|
||||
entity_type: Mapped[str] = mapped_column(String(50), nullable=False)
|
||||
entity_id: Mapped[uuid.UUID] = mapped_column(PGUUID(as_uuid=True), nullable=False)
|
||||
dms_file_id: Mapped[uuid.UUID] = mapped_column(
|
||||
PGUUID(as_uuid=True),
|
||||
ForeignKey("files.id", ondelete="RESTRICT"),
|
||||
nullable=False,
|
||||
)
|
||||
category: Mapped[str | None] = mapped_column(String(50), nullable=True)
|
||||
display_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
|
||||
created_by: Mapped[uuid.UUID | None] = mapped_column(
|
||||
PGUUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True
|
||||
)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, server_default=__import__('sqlalchemy').func.now()
|
||||
)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, server_default=__import__('sqlalchemy').func.now(),
|
||||
onupdate=__import__('sqlalchemy').func.now(),
|
||||
)
|
||||
deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
|
||||
@@ -93,12 +93,17 @@ async def download_attachment(
|
||||
if data is None:
|
||||
raise HTTPException(404, detail={"detail": "Attachment not found", "code": "not_found"})
|
||||
|
||||
file_path = data["file_path"]
|
||||
# Use storage backend instead of os.path.isfile (P1.1 fix)
|
||||
# Get storage path from DMS file via new unified service
|
||||
storage_path = await attachment_service.get_attachment_download_path(
|
||||
db, tenant_id, aid, user_id=user_id, is_system_admin=is_admin
|
||||
)
|
||||
if storage_path is None:
|
||||
raise HTTPException(404, detail={"detail": "File not found in DMS", "code": "file_missing"})
|
||||
|
||||
from app.core.storage import get_storage_backend
|
||||
storage = get_storage_backend()
|
||||
try:
|
||||
file_bytes = await storage.load(file_path)
|
||||
file_bytes = await storage.load(storage_path)
|
||||
except Exception:
|
||||
raise HTTPException(404, detail={"detail": "File not found in storage", "code": "file_missing"})
|
||||
|
||||
|
||||
@@ -1,35 +1,32 @@
|
||||
"""Attachment service — upload, list, download, delete with file storage."""
|
||||
"""Attachment service — unified through DMS.
|
||||
|
||||
All files are stored in the DMS (files table).
|
||||
entity_attachments just references the DMS file with entity_type/entity_id.
|
||||
|
||||
This unifies: one upload path, one download path, one permission model,
|
||||
one deduplication (content_hash), one storage backend.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy import select, func
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.audit import log_audit
|
||||
from app.core.storage import get_storage_backend
|
||||
from app.models.attachment import Attachment
|
||||
from app.core.visibility import apply_visibility_filter, check_single_entity_access
|
||||
from app.models.entity_attachment import EntityAttachment
|
||||
from app.plugins.builtins.dms.models import File as DmsFile
|
||||
|
||||
|
||||
def _attachment_to_dict(a: Attachment) -> dict[str, Any]:
|
||||
"""Serialize an Attachment ORM object to dict."""
|
||||
return {
|
||||
"id": str(a.id),
|
||||
"entity_type": a.entity_type,
|
||||
"entity_id": str(a.entity_id),
|
||||
"filename": a.filename,
|
||||
"file_path": a.file_path,
|
||||
"mime_type": a.mime_type,
|
||||
"file_size": a.file_size,
|
||||
"uploaded_by": str(a.uploaded_by) if a.uploaded_by else None,
|
||||
"created_at": a.created_at.isoformat() if a.created_at else None,
|
||||
"updated_at": a.updated_at.isoformat() if a.updated_at else None,
|
||||
}
|
||||
# File size limit: 50MB
|
||||
MAX_FILE_SIZE = 50 * 1024 * 1024
|
||||
|
||||
|
||||
def _generate_unique_filename(original_filename: str) -> str:
|
||||
@@ -38,6 +35,27 @@ def _generate_unique_filename(original_filename: str) -> str:
|
||||
return f"{uuid.uuid4().hex}{ext}"
|
||||
|
||||
|
||||
def _entity_attachment_to_dict(ea: EntityAttachment, dms_file: DmsFile | None = None) -> dict[str, Any]:
|
||||
"""Serialize an EntityAttachment + DMS File to dict."""
|
||||
return {
|
||||
"id": str(ea.id),
|
||||
"entity_type": ea.entity_type,
|
||||
"entity_id": str(ea.entity_id),
|
||||
"dms_file_id": str(ea.dms_file_id),
|
||||
"category": ea.category,
|
||||
"display_name": ea.display_name,
|
||||
"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,
|
||||
"updated_at": ea.updated_at.isoformat() if ea.updated_at else None,
|
||||
}
|
||||
|
||||
|
||||
async def save_attachment(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
@@ -49,51 +67,80 @@ async def save_attachment(
|
||||
mime_type: str,
|
||||
is_system_admin: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Save a file to storage and create an Attachment record."""
|
||||
# File size limit: 50MB
|
||||
MAX_FILE_SIZE = 50 * 1024 * 1024 # 50 MB
|
||||
"""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})")
|
||||
|
||||
# Check access on parent entity
|
||||
if not is_system_admin:
|
||||
from app.core.visibility import check_single_entity_access
|
||||
has_access = await check_single_entity_access(
|
||||
db, entity_type, entity_id, user_id, tenant_id, "write", is_system_admin
|
||||
)
|
||||
if not has_access:
|
||||
raise PermissionError(f"No write access to {entity_type} {entity_id}")
|
||||
|
||||
# Generate unique filename and relative storage path
|
||||
# Generate unique filename and storage path
|
||||
unique_filename = _generate_unique_filename(filename)
|
||||
file_path = f"{entity_type}/{entity_id}/{unique_filename}"
|
||||
storage_path = f"attachments/{entity_type}/{entity_id}/{unique_filename}"
|
||||
|
||||
# Save file via storage backend
|
||||
storage = get_storage_backend()
|
||||
await storage.save(file_path, file_content)
|
||||
# Calculate content hash for deduplication (tenant-local)
|
||||
content_hash = hashlib.sha256(file_content).hexdigest()
|
||||
|
||||
file_size = len(file_content)
|
||||
# Check for existing DMS file with same hash in same tenant (deduplication)
|
||||
existing_file = await db.execute(
|
||||
select(DmsFile).where(
|
||||
DmsFile.tenant_id == tenant_id,
|
||||
DmsFile.content_hash == content_hash,
|
||||
DmsFile.deleted_at.is_(None),
|
||||
).limit(1)
|
||||
)
|
||||
existing_dms_file = existing_file.scalar_one_or_none()
|
||||
|
||||
# Create DB record
|
||||
attachment = Attachment(
|
||||
if existing_dms_file:
|
||||
# Deduplicate: reuse existing DMS file, just create new reference
|
||||
dms_file = existing_dms_file
|
||||
else:
|
||||
# Save file via storage backend
|
||||
storage = get_storage_backend()
|
||||
await storage.save(storage_path, file_content)
|
||||
|
||||
# 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),
|
||||
storage_path=storage_path,
|
||||
content_hash=content_hash,
|
||||
owner_id=user_id,
|
||||
)
|
||||
db.add(dms_file)
|
||||
await db.flush()
|
||||
await db.refresh(dms_file)
|
||||
|
||||
# Create entity_attachments reference
|
||||
entity_attachment = EntityAttachment(
|
||||
tenant_id=tenant_id,
|
||||
entity_type=entity_type,
|
||||
entity_id=entity_id,
|
||||
filename=filename,
|
||||
file_path=file_path,
|
||||
mime_type=mime_type,
|
||||
file_size=file_size,
|
||||
uploaded_by=user_id,
|
||||
dms_file_id=dms_file.id,
|
||||
category=None,
|
||||
display_name=filename,
|
||||
owner_id=user_id,
|
||||
created_by=user_id,
|
||||
)
|
||||
db.add(attachment)
|
||||
db.add(entity_attachment)
|
||||
await db.flush()
|
||||
await db.refresh(attachment)
|
||||
await db.refresh(entity_attachment)
|
||||
|
||||
await log_audit(
|
||||
db, tenant_id, user_id, "upload", "attachment", attachment.id,
|
||||
changes={"filename": filename, "entity_type": entity_type, "entity_id": str(entity_id)},
|
||||
db, tenant_id, user_id, "upload", "attachment", entity_attachment.id,
|
||||
changes={"filename": filename, "entity_type": entity_type, "entity_id": str(entity_id), "dms_file_id": str(dms_file.id)},
|
||||
)
|
||||
return _attachment_to_dict(attachment)
|
||||
return _entity_attachment_to_dict(entity_attachment, dms_file)
|
||||
|
||||
|
||||
async def list_attachments(
|
||||
@@ -104,22 +151,28 @@ async def list_attachments(
|
||||
user_id: uuid.UUID | None = None,
|
||||
is_system_admin: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""List attachments for a specific entity."""
|
||||
q = select(Attachment).where(
|
||||
Attachment.tenant_id == tenant_id,
|
||||
Attachment.entity_type == entity_type,
|
||||
Attachment.entity_id == entity_id,
|
||||
Attachment.deleted_at.is_(None),
|
||||
).order_by(Attachment.created_at.desc())
|
||||
"""List attachments for a specific entity (via DMS files)."""
|
||||
q = (
|
||||
select(EntityAttachment, DmsFile)
|
||||
.join(DmsFile, EntityAttachment.dms_file_id == DmsFile.id)
|
||||
.where(
|
||||
EntityAttachment.tenant_id == tenant_id,
|
||||
EntityAttachment.entity_type == entity_type,
|
||||
EntityAttachment.entity_id == entity_id,
|
||||
EntityAttachment.deleted_at.is_(None),
|
||||
DmsFile.deleted_at.is_(None),
|
||||
)
|
||||
.order_by(EntityAttachment.created_at.desc())
|
||||
)
|
||||
if user_id and not is_system_admin:
|
||||
q = await apply_visibility_filter(
|
||||
db, q, "attachment", Attachment, user_id, tenant_id, is_system_admin
|
||||
db, q, "entity_attachment", EntityAttachment, user_id, tenant_id, is_system_admin
|
||||
)
|
||||
result = await db.execute(q)
|
||||
attachments = result.scalars().all()
|
||||
rows = result.all()
|
||||
return {
|
||||
"items": [_attachment_to_dict(a) for a in attachments],
|
||||
"total": len(attachments),
|
||||
"items": [_entity_attachment_to_dict(ea, df) for ea, df in rows],
|
||||
"total": len(rows),
|
||||
}
|
||||
|
||||
|
||||
@@ -130,23 +183,60 @@ async def get_attachment(
|
||||
user_id: uuid.UUID | None = None,
|
||||
is_system_admin: bool = False,
|
||||
) -> dict[str, Any] | None:
|
||||
"""Get a single attachment by ID."""
|
||||
q = select(Attachment).where(
|
||||
Attachment.id == attachment_id,
|
||||
Attachment.tenant_id == tenant_id,
|
||||
Attachment.deleted_at.is_(None),
|
||||
"""Get a single attachment by ID (with DMS file info)."""
|
||||
q = (
|
||||
select(EntityAttachment, DmsFile)
|
||||
.join(DmsFile, EntityAttachment.dms_file_id == DmsFile.id)
|
||||
.where(
|
||||
EntityAttachment.id == attachment_id,
|
||||
EntityAttachment.tenant_id == tenant_id,
|
||||
EntityAttachment.deleted_at.is_(None),
|
||||
)
|
||||
)
|
||||
result = await db.execute(q)
|
||||
attachment = result.scalar_one_or_none()
|
||||
if attachment is None:
|
||||
row = result.first()
|
||||
if row is None:
|
||||
return None
|
||||
ea, dms_file = row
|
||||
if user_id and not is_system_admin:
|
||||
has_access = await check_single_entity_access(
|
||||
db, "attachment", attachment.id, user_id, tenant_id, "read", is_system_admin
|
||||
db, "entity_attachment", ea.id, user_id, tenant_id, "read", is_system_admin
|
||||
)
|
||||
if not has_access:
|
||||
raise PermissionError("No access")
|
||||
return _attachment_to_dict(attachment)
|
||||
return _entity_attachment_to_dict(ea, dms_file)
|
||||
|
||||
|
||||
async def get_attachment_download_path(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
attachment_id: uuid.UUID,
|
||||
user_id: uuid.UUID | None = None,
|
||||
is_system_admin: bool = False,
|
||||
) -> str | None:
|
||||
"""Get the storage path for downloading an attachment's DMS file."""
|
||||
q = (
|
||||
select(EntityAttachment, DmsFile)
|
||||
.join(DmsFile, EntityAttachment.dms_file_id == DmsFile.id)
|
||||
.where(
|
||||
EntityAttachment.id == attachment_id,
|
||||
EntityAttachment.tenant_id == tenant_id,
|
||||
EntityAttachment.deleted_at.is_(None),
|
||||
DmsFile.deleted_at.is_(None),
|
||||
)
|
||||
)
|
||||
result = await db.execute(q)
|
||||
row = result.first()
|
||||
if row is None:
|
||||
return None
|
||||
ea, dms_file = row
|
||||
if user_id and not is_system_admin:
|
||||
has_access = await check_single_entity_access(
|
||||
db, "entity_attachment", ea.id, user_id, tenant_id, "read", is_system_admin
|
||||
)
|
||||
if not has_access:
|
||||
raise PermissionError("No access")
|
||||
return dms_file.storage_path
|
||||
|
||||
|
||||
async def delete_attachment(
|
||||
@@ -156,35 +246,32 @@ async def delete_attachment(
|
||||
attachment_id: uuid.UUID,
|
||||
is_system_admin: bool = False,
|
||||
) -> bool:
|
||||
"""Soft-delete an attachment and remove physical file from storage."""
|
||||
q = select(Attachment).where(
|
||||
Attachment.id == attachment_id,
|
||||
Attachment.tenant_id == tenant_id,
|
||||
Attachment.deleted_at.is_(None),
|
||||
"""Soft-delete an entity_attachments reference.
|
||||
|
||||
The DMS file is NOT deleted because other entities may reference it.
|
||||
DMS file cleanup happens via DMS's own deletion workflow.
|
||||
"""
|
||||
q = select(EntityAttachment).where(
|
||||
EntityAttachment.id == attachment_id,
|
||||
EntityAttachment.tenant_id == tenant_id,
|
||||
EntityAttachment.deleted_at.is_(None),
|
||||
)
|
||||
result = await db.execute(q)
|
||||
attachment = result.scalar_one_or_none()
|
||||
if attachment is None:
|
||||
ea = result.scalar_one_or_none()
|
||||
if ea is None:
|
||||
return False
|
||||
|
||||
if not is_system_admin:
|
||||
has_access = await check_single_entity_access(
|
||||
db, "attachment", attachment.id, user_id, tenant_id, "admin", is_system_admin
|
||||
db, "entity_attachment", ea.id, user_id, tenant_id, "admin", is_system_admin
|
||||
)
|
||||
if not has_access:
|
||||
raise PermissionError("No access")
|
||||
|
||||
# Delete physical file from storage (P1.1 fix)
|
||||
try:
|
||||
storage = get_storage_backend()
|
||||
await storage.delete(attachment.file_path)
|
||||
except Exception as exc:
|
||||
logger.warning("Failed to delete physical file %s: %s", attachment.file_path, exc)
|
||||
|
||||
attachment.deleted_at = datetime.now(UTC)
|
||||
ea.deleted_at = datetime.now(UTC)
|
||||
await db.flush()
|
||||
await log_audit(
|
||||
db, tenant_id, user_id, "delete", "attachment", attachment_id,
|
||||
changes={"filename": attachment.filename},
|
||||
changes={"display_name": ea.display_name, "dms_file_id": str(ea.dms_file_id)},
|
||||
)
|
||||
return True
|
||||
|
||||
Reference in New Issue
Block a user