62 lines
1.9 KiB
Python
62 lines
1.9 KiB
Python
from datetime import datetime, timezone
|
|
from uuid import uuid4, UUID
|
|
from fastapi import HTTPException
|
|
|
|
from fastapi_demo.app.domain.status import AssetStatus
|
|
from fastapi_demo.app.domain.transitions import ensure_transition_allowed
|
|
from fastapi_demo.app.infrastructure.repositories.assets_repo import AssetsRepo
|
|
from fastapi_demo.app.schemas.asset import (
|
|
AssetCreate,
|
|
AssetOut,
|
|
AssetTransitionIn,
|
|
AssetEventOut,
|
|
)
|
|
|
|
|
|
class AssetsService:
|
|
def __init__(self, repo: AssetsRepo) -> None:
|
|
self.repo = repo
|
|
|
|
def create_asset(self, data: AssetCreate) -> AssetOut:
|
|
now = datetime.now(timezone.utc)
|
|
asset = AssetOut(
|
|
id=uuid4(),
|
|
name=data.name,
|
|
serial=data.serial,
|
|
status=AssetStatus.WARENEINGANG,
|
|
revision=0, # <-- neu
|
|
updated_at=now,
|
|
)
|
|
self.repo.create(asset)
|
|
return asset
|
|
|
|
def get_asset(self, asset_id: UUID) -> AssetOut:
|
|
asset = self.repo.get(asset_id)
|
|
if not asset:
|
|
raise HTTPException(status_code=404, detail="Asset nicht gefunden")
|
|
return asset
|
|
|
|
def transition(
|
|
self, asset_id: UUID, data: AssetTransitionIn
|
|
) -> tuple[AssetOut, AssetEventOut]:
|
|
asset = self.get_asset(asset_id)
|
|
|
|
# Domain-Regel: Transition erlaubt?
|
|
ensure_transition_allowed(asset.status, data.to_status)
|
|
|
|
now = datetime.now(timezone.utc)
|
|
|
|
# Repo macht atomar: revision check + status update + event insert
|
|
updated_asset, event = self.repo.transition_with_revision(
|
|
asset_id=asset_id,
|
|
expected_revision=data.expected_revision,
|
|
to_status=data.to_status,
|
|
at=now,
|
|
note=data.note,
|
|
)
|
|
return updated_asset, event
|
|
|
|
def list_events(self, asset_id: UUID) -> list[AssetEventOut]:
|
|
self.get_asset(asset_id) # 404 falls nicht existiert
|
|
return self.repo.list_events(asset_id)
|