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>
254 lines
8.0 KiB
Python
254 lines
8.0 KiB
Python
"""CRUD operations for receipts."""
|
|
|
|
from datetime import datetime, date
|
|
from typing import Optional, List, Tuple
|
|
from sqlalchemy import select, func, or_
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.db.models.receipt import Receipt, ReceiptStatus
|
|
from app.schemas.receipt import ReceiptCreate, ReceiptUpdate, ReceiptFilter
|
|
|
|
|
|
class ReceiptCRUD:
|
|
"""CRUD operations for Receipt model."""
|
|
|
|
@staticmethod
|
|
async def create(
|
|
session: AsyncSession,
|
|
data: ReceiptCreate,
|
|
created_by: str,
|
|
) -> Receipt:
|
|
"""Create a new receipt."""
|
|
receipt = Receipt(
|
|
**data.model_dump(),
|
|
created_by=created_by,
|
|
status=ReceiptStatus.DRAFT,
|
|
)
|
|
session.add(receipt)
|
|
await session.commit()
|
|
await session.refresh(receipt)
|
|
return receipt
|
|
|
|
@staticmethod
|
|
async def get_by_id(
|
|
session: AsyncSession,
|
|
receipt_id: int,
|
|
include_relations: bool = True,
|
|
) -> Optional[Receipt]:
|
|
"""Get receipt by ID, optionally with relationships."""
|
|
query = select(Receipt).where(Receipt.id == receipt_id)
|
|
|
|
if include_relations:
|
|
query = query.options(
|
|
selectinload(Receipt.attachments),
|
|
selectinload(Receipt.entries),
|
|
)
|
|
|
|
result = await session.execute(query)
|
|
return result.scalar_one_or_none()
|
|
|
|
@staticmethod
|
|
async def get_list(
|
|
session: AsyncSession,
|
|
filters: ReceiptFilter,
|
|
) -> Tuple[List[Receipt], int]:
|
|
"""Get paginated list of receipts with filters."""
|
|
# Base query
|
|
query = select(Receipt).options(
|
|
selectinload(Receipt.attachments),
|
|
selectinload(Receipt.entries),
|
|
)
|
|
|
|
# Apply filters
|
|
if filters.status:
|
|
query = query.where(Receipt.status == filters.status)
|
|
|
|
if filters.company_id:
|
|
query = query.where(Receipt.company_id == filters.company_id)
|
|
|
|
if filters.created_by:
|
|
query = query.where(Receipt.created_by == filters.created_by)
|
|
|
|
if filters.date_from:
|
|
query = query.where(Receipt.receipt_date >= filters.date_from)
|
|
|
|
if filters.date_to:
|
|
query = query.where(Receipt.receipt_date <= filters.date_to)
|
|
|
|
if filters.search:
|
|
search_term = f"%{filters.search}%"
|
|
query = query.where(
|
|
or_(
|
|
Receipt.description.ilike(search_term),
|
|
Receipt.partner_name.ilike(search_term),
|
|
Receipt.receipt_number.ilike(search_term),
|
|
)
|
|
)
|
|
|
|
# Count total
|
|
count_query = select(func.count()).select_from(query.subquery())
|
|
total_result = await session.execute(count_query)
|
|
total = total_result.scalar() or 0
|
|
|
|
# Apply pagination and ordering
|
|
query = query.order_by(Receipt.created_at.desc())
|
|
offset = (filters.page - 1) * filters.page_size
|
|
query = query.offset(offset).limit(filters.page_size)
|
|
|
|
# Execute
|
|
result = await session.execute(query)
|
|
receipts = result.scalars().all()
|
|
|
|
return list(receipts), total
|
|
|
|
@staticmethod
|
|
async def get_pending_review(
|
|
session: AsyncSession,
|
|
company_id: Optional[int] = None,
|
|
) -> List[Receipt]:
|
|
"""Get all receipts pending review."""
|
|
query = select(Receipt).where(
|
|
Receipt.status == ReceiptStatus.PENDING_REVIEW
|
|
).options(
|
|
selectinload(Receipt.attachments),
|
|
selectinload(Receipt.entries),
|
|
)
|
|
|
|
if company_id:
|
|
query = query.where(Receipt.company_id == company_id)
|
|
|
|
query = query.order_by(Receipt.submitted_at.asc())
|
|
|
|
result = await session.execute(query)
|
|
return list(result.scalars().all())
|
|
|
|
@staticmethod
|
|
async def update(
|
|
session: AsyncSession,
|
|
receipt: Receipt,
|
|
data: ReceiptUpdate,
|
|
) -> Receipt:
|
|
"""Update receipt fields."""
|
|
update_data = data.model_dump(exclude_unset=True)
|
|
|
|
for field, value in update_data.items():
|
|
setattr(receipt, field, value)
|
|
|
|
receipt.updated_at = datetime.utcnow()
|
|
|
|
session.add(receipt)
|
|
await session.commit()
|
|
await session.refresh(receipt)
|
|
return receipt
|
|
|
|
@staticmethod
|
|
async def update_status(
|
|
session: AsyncSession,
|
|
receipt: Receipt,
|
|
new_status: ReceiptStatus,
|
|
reviewed_by: Optional[str] = None,
|
|
rejection_reason: Optional[str] = None,
|
|
) -> Receipt:
|
|
"""Update receipt workflow status."""
|
|
receipt.status = new_status
|
|
receipt.updated_at = datetime.utcnow()
|
|
|
|
if new_status == ReceiptStatus.PENDING_REVIEW:
|
|
receipt.submitted_at = datetime.utcnow()
|
|
|
|
if new_status in [ReceiptStatus.APPROVED, ReceiptStatus.REJECTED]:
|
|
receipt.reviewed_by = reviewed_by
|
|
receipt.reviewed_at = datetime.utcnow()
|
|
|
|
if new_status == ReceiptStatus.REJECTED:
|
|
receipt.rejection_reason = rejection_reason
|
|
|
|
if new_status == ReceiptStatus.DRAFT:
|
|
# Reset review fields when moving back to draft
|
|
receipt.rejection_reason = None
|
|
|
|
session.add(receipt)
|
|
await session.commit()
|
|
await session.refresh(receipt)
|
|
return receipt
|
|
|
|
@staticmethod
|
|
async def delete(session: AsyncSession, receipt: Receipt) -> bool:
|
|
"""Delete a receipt (cascade deletes attachments and entries)."""
|
|
await session.delete(receipt)
|
|
await session.commit()
|
|
return True
|
|
|
|
@staticmethod
|
|
async def can_edit(receipt: Receipt, username: str) -> bool:
|
|
"""Check if user can edit receipt."""
|
|
# Only DRAFT receipts can be edited
|
|
if receipt.status != ReceiptStatus.DRAFT:
|
|
return False
|
|
|
|
# Only creator can edit their own drafts
|
|
return receipt.created_by == username
|
|
|
|
@staticmethod
|
|
async def can_delete(receipt: Receipt, username: str) -> bool:
|
|
"""Check if user can delete receipt."""
|
|
# Only DRAFT receipts can be deleted
|
|
if receipt.status != ReceiptStatus.DRAFT:
|
|
return False
|
|
|
|
# Only creator can delete their own drafts
|
|
return receipt.created_by == username
|
|
|
|
@staticmethod
|
|
async def can_submit(receipt: Receipt, username: str) -> bool:
|
|
"""Check if user can submit receipt for review."""
|
|
# Only DRAFT or REJECTED receipts can be submitted
|
|
if receipt.status not in [ReceiptStatus.DRAFT, ReceiptStatus.REJECTED]:
|
|
return False
|
|
|
|
# Only creator can submit their own receipts
|
|
return receipt.created_by == username
|
|
|
|
@staticmethod
|
|
async def get_stats(
|
|
session: AsyncSession,
|
|
company_id: int,
|
|
created_by: Optional[str] = None,
|
|
) -> dict:
|
|
"""Get receipt statistics."""
|
|
base_query = select(
|
|
Receipt.status,
|
|
func.count(Receipt.id).label("count"),
|
|
func.sum(Receipt.amount).label("total_amount"),
|
|
).where(
|
|
Receipt.company_id == company_id
|
|
)
|
|
|
|
if created_by:
|
|
base_query = base_query.where(Receipt.created_by == created_by)
|
|
|
|
query = base_query.group_by(Receipt.status)
|
|
result = await session.execute(query)
|
|
rows = result.all()
|
|
|
|
stats = {
|
|
"draft": {"count": 0, "amount": 0},
|
|
"pending_review": {"count": 0, "amount": 0},
|
|
"approved": {"count": 0, "amount": 0},
|
|
"rejected": {"count": 0, "amount": 0},
|
|
"synced": {"count": 0, "amount": 0},
|
|
"total": {"count": 0, "amount": 0},
|
|
}
|
|
|
|
for row in rows:
|
|
status_key = row.status.value
|
|
stats[status_key] = {
|
|
"count": row.count,
|
|
"amount": float(row.total_amount or 0),
|
|
}
|
|
stats["total"]["count"] += row.count
|
|
stats["total"]["amount"] += float(row.total_amount or 0)
|
|
|
|
return stats
|