Update app/services/sequence_service.py

app/services/sequence_service.py

@@ -1,133 +1 @@
-"""Sequence service — get_next_number with SELECT FOR UPDATE locking, CRUD, create."""
+§§include(/a0/usr/chats/XttavUaL/messages/seq_svc_update.py)
-
-from __future__ import annotations
-
-import uuid
-from datetime import UTC, datetime
-from typing import Any
-
-from sqlalchemy import select, text
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from app.core.audit import log_audit
-from app.models.sequence import Sequence
-
-
-def _sequence_to_dict(s: Sequence) -> dict[str, Any]:
-    """Serialize a Sequence ORM object to dict."""
-    return {
-        "id": str(s.id),
-        "name": s.name,
-        "prefix": s.prefix,
-        "next_number": s.next_number,
-        "padding": s.padding,
-        "created_at": s.created_at.isoformat() if s.created_at else None,
-        "updated_at": s.updated_at.isoformat() if s.updated_at else None,
-    }
-
-
-async def get_next_number(
-    db: AsyncSession,
-    name: str,
-    tenant_id: uuid.UUID,
-) -> str:
-    """Get the next formatted number for a sequence.
-
-    Uses SELECT FOR UPDATE to lock the row, increment next_number,
-    and return the formatted string (prefix + zero-padded number).
-
-    Raises ValueError if the sequence does not exist.
-    """
-    # SELECT FOR UPDATE to lock the row
-    q = (
-        select(Sequence)
-        .where(
-            Sequence.name == name,
-            Sequence.tenant_id == tenant_id,
-            Sequence.deleted_at.is_(None),
-        )
-        .with_for_update()
-    )
-    result = await db.execute(q)
-    seq = result.scalar_one_or_none()
-    if seq is None:
-        raise ValueError(f"Sequence '{name}' not found for tenant {tenant_id}")
-
-    current_number = seq.next_number
-    formatted = f"{seq.prefix}{str(current_number).zfill(seq.padding)}"
-    seq.next_number = current_number + 1
-
-    await db.flush()
-    return formatted
-
-
-async def create_sequence(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    data: dict[str, Any],
-) -> dict[str, Any]:
-    """Create a new sequence."""
-    sequence = Sequence(
-        tenant_id=tenant_id,
-        name=data["name"],
-        prefix=data.get("prefix", ""),
-        next_number=1,
-        padding=data.get("padding", 4),
-    )
-    db.add(sequence)
-    await db.flush()
-    await db.refresh(sequence)
-    await log_audit(
-        db, tenant_id, user_id, "create", "sequence", sequence.id,
-        changes={"name": sequence.name, "prefix": sequence.prefix},
-    )
-    return _sequence_to_dict(sequence)
-
-
-async def list_sequences(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-) -> dict[str, Any]:
-    """List all sequences for a tenant."""
-    q = select(Sequence).where(
-        Sequence.tenant_id == tenant_id,
-        Sequence.deleted_at.is_(None),
-    ).order_by(Sequence.name.asc())
-    result = await db.execute(q)
-    sequences = result.scalars().all()
-    return {
-        "items": [_sequence_to_dict(s) for s in sequences],
-        "total": len(sequences),
-    }
-
-
-async def update_sequence(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    sequence_id: uuid.UUID,
-    data: dict[str, Any],
-) -> dict[str, Any] | None:
-    """Update a sequence (name, prefix, padding only — NOT next_number)."""
-    q = select(Sequence).where(
-        Sequence.id == sequence_id,
-        Sequence.tenant_id == tenant_id,
-        Sequence.deleted_at.is_(None),
-    )
-    result = await db.execute(q)
-    sequence = result.scalar_one_or_none()
-    if sequence is None:
-        return None
-
-    changes: dict[str, Any] = {}
-    for field in ("name", "prefix", "padding"):
-        if field in data and data[field] is not None:
-            old_val = getattr(sequence, field)
-            changes[field] = {"old": old_val, "new": data[field]}
-            setattr(sequence, field, data[field])
-
-    await db.flush()
-    await db.refresh(sequence)
-    await log_audit(db, tenant_id, user_id, "update", "sequence", sequence_id, changes=changes)
-    return _sequence_to_dict(sequence)