Marius Mutu
21c12ddb0f
New application for entering fiscal receipts (bonuri fiscale) with: Backend (FastAPI + SQLModel + Alembic): - Receipt, ReceiptAttachment, AccountingEntry models - CRUD operations with async SQLite database - Workflow: DRAFT → PENDING_REVIEW → APPROVED/REJECTED - Auto-generation of accounting entries with VAT calculation - File upload support (images, PDFs) - Predefined expense types (Fuel, Materials, Office, etc.) - Nomenclature service for partners, accounts, cash registers Frontend (Vue.js 3 + PrimeVue + Pinia): - ReceiptsListView with filters and stats - ReceiptCreateView with image upload - ReceiptDetailView with accounting entries - ReceiptApprovalView for accountant approval Documentation: - REQUIREMENTS.md with functional specifications - ARCHITECTURE.md with technical decisions - CLAUDE.md for AI assistant guidance Phase 1 MVP uses SQLite, prepared for Oracle integration in Phase 2. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
141 lines
4.5 KiB
Python
141 lines
4.5 KiB
Python
"""CRUD operations for receipt attachments."""
|
|
|
|
import os
|
|
import uuid
|
|
import aiofiles
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Optional, List
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from fastapi import UploadFile
|
|
|
|
from app.db.models.receipt import ReceiptAttachment
|
|
from app.config import settings
|
|
|
|
|
|
class AttachmentCRUD:
|
|
"""CRUD operations for ReceiptAttachment model."""
|
|
|
|
@staticmethod
|
|
def _generate_stored_filename(original_filename: str) -> str:
|
|
"""Generate unique filename for storage."""
|
|
ext = Path(original_filename).suffix.lower()
|
|
return f"{uuid.uuid4()}{ext}"
|
|
|
|
@staticmethod
|
|
def _get_upload_path(stored_filename: str) -> Path:
|
|
"""Get full path for storing file, organized by year/month."""
|
|
now = datetime.utcnow()
|
|
relative_path = Path(str(now.year)) / f"{now.month:02d}"
|
|
full_path = settings.upload_path_resolved / relative_path
|
|
|
|
# Ensure directory exists
|
|
full_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
return relative_path / stored_filename
|
|
|
|
@staticmethod
|
|
async def create(
|
|
session: AsyncSession,
|
|
receipt_id: int,
|
|
file: UploadFile,
|
|
) -> ReceiptAttachment:
|
|
"""Create attachment by saving file and creating DB record."""
|
|
# Generate stored filename
|
|
stored_filename = AttachmentCRUD._generate_stored_filename(file.filename or "upload")
|
|
|
|
# Get relative path
|
|
relative_path = AttachmentCRUD._get_upload_path(stored_filename)
|
|
|
|
# Full path for saving
|
|
full_path = settings.upload_path_resolved / relative_path
|
|
|
|
# Read file content
|
|
content = await file.read()
|
|
file_size = len(content)
|
|
|
|
# Validate file size
|
|
if file_size > settings.max_upload_size_bytes:
|
|
raise ValueError(f"File too large. Maximum size is {settings.max_upload_size_mb}MB")
|
|
|
|
# Validate MIME type
|
|
mime_type = file.content_type or "application/octet-stream"
|
|
if mime_type not in settings.allowed_mime_types:
|
|
raise ValueError(f"File type not allowed: {mime_type}")
|
|
|
|
# Save file
|
|
async with aiofiles.open(full_path, "wb") as f:
|
|
await f.write(content)
|
|
|
|
# Create DB record
|
|
attachment = ReceiptAttachment(
|
|
receipt_id=receipt_id,
|
|
filename=file.filename or "upload",
|
|
stored_filename=stored_filename,
|
|
file_path=str(relative_path),
|
|
file_size=file_size,
|
|
mime_type=mime_type,
|
|
)
|
|
|
|
session.add(attachment)
|
|
await session.commit()
|
|
await session.refresh(attachment)
|
|
|
|
return attachment
|
|
|
|
@staticmethod
|
|
async def get_by_id(
|
|
session: AsyncSession,
|
|
attachment_id: int,
|
|
) -> Optional[ReceiptAttachment]:
|
|
"""Get attachment by ID."""
|
|
query = select(ReceiptAttachment).where(ReceiptAttachment.id == attachment_id)
|
|
result = await session.execute(query)
|
|
return result.scalar_one_or_none()
|
|
|
|
@staticmethod
|
|
async def get_by_receipt_id(
|
|
session: AsyncSession,
|
|
receipt_id: int,
|
|
) -> List[ReceiptAttachment]:
|
|
"""Get all attachments for a receipt."""
|
|
query = select(ReceiptAttachment).where(
|
|
ReceiptAttachment.receipt_id == receipt_id
|
|
).order_by(ReceiptAttachment.uploaded_at.asc())
|
|
|
|
result = await session.execute(query)
|
|
return list(result.scalars().all())
|
|
|
|
@staticmethod
|
|
def get_file_path(attachment: ReceiptAttachment) -> Path:
|
|
"""Get full file path for an attachment."""
|
|
return settings.upload_path_resolved / attachment.file_path
|
|
|
|
@staticmethod
|
|
async def delete(session: AsyncSession, attachment: ReceiptAttachment) -> bool:
|
|
"""Delete attachment (file and DB record)."""
|
|
# Delete file
|
|
file_path = AttachmentCRUD.get_file_path(attachment)
|
|
if file_path.exists():
|
|
os.remove(file_path)
|
|
|
|
# Delete DB record
|
|
await session.delete(attachment)
|
|
await session.commit()
|
|
|
|
return True
|
|
|
|
@staticmethod
|
|
async def delete_all_for_receipt(session: AsyncSession, receipt_id: int) -> int:
|
|
"""Delete all attachments for a receipt."""
|
|
attachments = await AttachmentCRUD.get_by_receipt_id(session, receipt_id)
|
|
count = 0
|
|
|
|
for attachment in attachments:
|
|
await AttachmentCRUD.delete(session, attachment)
|
|
count += 1
|
|
|
|
return count
|