"""HTTP routes for the asset inventory (盘点) feature. Two surfaces, one router: * **Admin** (``require_asset_admin``) — full lifecycle: create / list / preview / start / detail / list items / send reminder / resolve issue / close. * **Employee** (``get_current_user``) — list pending items routed to the caller, respond to one. Authorisation is per-item (the holder must match) and is enforced inside ``AssetInventoryService``. """ from __future__ import annotations from typing import Any from uuid import UUID from fastapi import APIRouter, Body, Depends, Query, Request from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.background import spawn as _spawn_background from app.core.database import get_db from app.core.dependencies import ( get_current_user, get_real_ip, require_asset_admin, require_asset_read, ) from app.core.rbac import can_manage_assets from app.core.response import paginated_response, success_response from app.domains.asset.inventory_schemas import ( CycleCreateRequest, CycleRemindRequest, CycleScopeFilter, CycleUpdateRequest, ItemIssueResolveRequest, ItemRespondRequest, ) from app.domains.asset.inventory_service import ( AssetInventoryService, run_start_notifications_in_background, ) from app.domains.asset.models import ( AssetInventoryCycle, AssetInventoryItem, AssetItem, AssetOfficeAddress, ) from app.identity.person_models import IdentityPerson router = APIRouter(prefix="/api/v1/asset-inventory", tags=["asset-inventory"]) # ── Hydration helpers ───────────────────────────────────────────────── async def _hydrate_cycle( db: AsyncSession, cycle: AssetInventoryCycle, ) -> dict[str, Any]: created_by_name: str | None = None if cycle.created_by: person = await db.get(IdentityPerson, cycle.created_by) if person: created_by_name = person.display_name or person.work_email return { "id": cycle.id, "name": cycle.name, "description": cycle.description, "status": cycle.status, "scope_filter": cycle.scope_filter or {}, "started_at": cycle.started_at, "due_at": cycle.due_at, "closed_at": cycle.closed_at, "reminder_sent_at": cycle.reminder_sent_at, "created_by": cycle.created_by, "created_by_name": created_by_name, "closed_by": cycle.closed_by, "total_items": cycle.total_items, "pending_count": cycle.pending_count, "confirmed_count": cycle.confirmed_count, "issue_count": cycle.issue_count, "created_at": cycle.created_at, "updated_at": cycle.updated_at, } async def _hydrate_items( db: AsyncSession, items: list[AssetInventoryItem], *, admin_view: bool = True, ) -> list[dict[str, Any]]: """Bulk-resolve display names + current asset state for a batch of items. One query each — beats N+1 and keeps the route handler fast for the common 50-row page. ``admin_view`` (default True for back-compat) gates the ``new_location`` field on each row. The field carries the holder's raw free-text response to a ``moved_to`` inventory prompt — typically an HRIS-derived city / country string (``Santa Clara`` / ``北京市`` / ``Würzburg``) that the write-time resolver canonicalises onto ``asset.location`` but preserves verbatim on the inventory item row for forensic traceability. The same admin-only-audit policy that gates ``operation.detail.location_raw`` and the structured logger output applies here — non-admin viewers (the management-tier audit roles on ``GET /inventory/items``) see the canonical ``new_location`` redacted out. The holder's own ``list_my_pending`` view DOES see this value (they typed it) — caller decides via ``admin_view``. Round-6 review high. """ if not items: return [] person_ids: set[UUID] = set() asset_ids: set[UUID] = set() office_ids: set[UUID] = set() for it in items: if it.holder_person_id: person_ids.add(it.holder_person_id) if it.transferred_to_id: person_ids.add(it.transferred_to_id) if it.responded_by: person_ids.add(it.responded_by) asset_ids.add(it.asset_id) if it.office_id: office_ids.add(it.office_id) person_lookup: dict[UUID, IdentityPerson] = {} if person_ids: rows = ( await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(person_ids)) ) ).scalars().all() person_lookup = {p.id: p for p in rows} asset_lookup: dict[UUID, AssetItem] = {} if asset_ids: rows = ( await db.execute( select(AssetItem).where(AssetItem.id.in_(asset_ids)) ) ).scalars().all() asset_lookup = {a.id: a for a in rows} office_lookup: dict[UUID, AssetOfficeAddress] = {} if office_ids: rows = ( await db.execute( select(AssetOfficeAddress).where( AssetOfficeAddress.id.in_(office_ids), ) ) ).scalars().all() office_lookup = {o.id: o for o in rows} # Names for "current_assigned_to" — the asset's *current* assignee, # which may differ from the snapshot if a transfer happened mid-cycle. current_assignee_ids = { a.assigned_to for a in asset_lookup.values() if a.assigned_to } if current_assignee_ids: rows = ( await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(current_assignee_ids), ) ) ).scalars().all() for p in rows: person_lookup.setdefault(p.id, p) out: list[dict[str, Any]] = [] for it in items: holder = person_lookup.get(it.holder_person_id) if it.holder_person_id else None responder = person_lookup.get(it.responded_by) if it.responded_by else None transferred = person_lookup.get(it.transferred_to_id) if it.transferred_to_id else None office = office_lookup.get(it.office_id) if it.office_id else None asset = asset_lookup.get(it.asset_id) current_assignee = ( person_lookup.get(asset.assigned_to) if asset and asset.assigned_to else None ) out.append({ "id": it.id, "cycle_id": it.cycle_id, "asset_id": it.asset_id, "holder_kind": it.holder_kind, "holder_person_id": it.holder_person_id, "holder_name": holder.display_name if holder else None, "holder_employee_no": holder.employee_no if holder else None, "holder_employment_status": ( holder.employment_status if holder else None ), "office_id": it.office_id, "office_city": office.city if office else None, "asset_tag_snapshot": it.asset_tag_snapshot, "asset_name_snapshot": it.asset_name_snapshot, "asset_category_snapshot": it.asset_category_snapshot, "asset_location_snapshot": it.asset_location_snapshot, "asset_status_snapshot": it.asset_status_snapshot, "current_status": asset.status if asset else None, "current_assigned_to_name": ( current_assignee.display_name if current_assignee else None ), "current_brand": asset.brand if asset else None, "current_model": asset.model if asset else None, "current_serial_number": asset.serial_number if asset else None, "status": it.status, "response_type": it.response_type, "response_notes": it.response_notes, "transferred_to_id": it.transferred_to_id, "transferred_to_name": ( transferred.display_name if transferred else None ), # ``new_location`` is the holder's raw free-text response # to a moved_to prompt — admin-only audit signal. See # ``_hydrate_items`` docstring for the rationale. "new_location": it.new_location if admin_view else None, "responded_at": it.responded_at, "responded_by": it.responded_by, "responded_by_name": responder.display_name if responder else None, # True when IT recorded the response *for* the routed holder # ("IT 帮忙确认"). Derived, not stored — a proxy response is # exactly ``responded_by != holder_person_id`` on a row that # HAS a holder. ``it_fallback`` rows have no holder at all, # so an admin answering that bucket is the normal path, not # a proxy. Both the admin table and the holder's own # /me/inventory view render this so a response the employee # never typed is never presented as if they did. "responded_on_behalf": bool( it.responded_by and it.holder_person_id and it.responded_by != it.holder_person_id ), "reminded_at": it.reminded_at, "issue_status": it.issue_status, "issue_resolved_at": it.issue_resolved_at, "issue_resolved_by": it.issue_resolved_by, "issue_resolution_notes": it.issue_resolution_notes, "evidence_photos": it.evidence_photos or None, "created_at": it.created_at, "updated_at": it.updated_at, }) return out # ── Admin endpoints ─────────────────────────────────────────────────── @router.get("/cycles") async def list_cycles( request: Request, status: str | None = Query(None), page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200), current_user: AuthUser = Depends(require_asset_read()), db: AsyncSession = Depends(get_db), ): service = AssetInventoryService(db) rows, total = await service.list_cycles( status=status, page=page, page_size=page_size, ) data = [await _hydrate_cycle(db, c) for c in rows] return paginated_response( data=data, total=total, page=page, page_size=page_size, request=request, ) @router.post("/cycles") async def create_cycle( request: Request, body: CycleCreateRequest, current_user: AuthUser = Depends(require_asset_admin()), db: AsyncSession = Depends(get_db), ): service = AssetInventoryService(db) cycle = await service.create_cycle( name=body.name, description=body.description, scope_filter=body.scope_filter, due_at=body.due_at, created_by=current_user.id, ) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="asset_inventory.cycle.create", resource_type="asset_inventory_cycle", resource_id=str(cycle.id), status="success", detail={ "name": cycle.name, "scope_filter": cycle.scope_filter, "auto_start": body.auto_start, }, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) started = False if body.auto_start: cycle = await service.start_cycle( cycle.id, actor_user_id=current_user.id, trace_id=request.state.trace_id, ) started = True await audit.log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="asset_inventory.cycle.start", resource_type="asset_inventory_cycle", resource_id=str(cycle.id), status="success", detail={"total_items": cycle.total_items}, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() if started: # Fan out the launch cards off the request — see the start endpoint. _spawn_background( run_start_notifications_in_background( cycle.id, actor_user_id=current_user.id, trace_id=request.state.trace_id, ), name="asset_inventory.start_notify", ) return success_response( data=await _hydrate_cycle(db, cycle), request=request, ) @router.get("/cycles/{cycle_id}") async def get_cycle( request: Request, cycle_id: UUID, current_user: AuthUser = Depends(require_asset_read()), db: AsyncSession = Depends(get_db), ): service = AssetInventoryService(db) cycle = await service.get_cycle(cycle_id) return success_response( data=await _hydrate_cycle(db, cycle), request=request, ) @router.patch("/cycles/{cycle_id}") async def update_cycle( request: Request, cycle_id: UUID, body: CycleUpdateRequest, current_user: AuthUser = Depends(require_asset_admin()), db: AsyncSession = Depends(get_db), ): service = AssetInventoryService(db) # ``exclude_unset`` is what separates "field omitted" from "field # explicitly set to null" — the edit modal clears a deadline by # sending ``due_at: null``, and renaming a cycle must not wipe it. provided = body.model_fields_set cycle = await service.update_cycle( cycle_id, name=body.name, description=body.description, due_at=body.due_at, clear_due_at="due_at" in provided and body.due_at is None, ) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="asset_inventory.cycle.update", resource_type="asset_inventory_cycle", resource_id=str(cycle.id), status="success", # ``mode="json"`` coerces datetime → ISO string. Without it the # raw ``due_at`` datetime reached ``audit_logs.detail`` (a JSON # column whose serializer is plain ``json.dumps`` — no # ``default=``) and every PATCH carrying a deadline 500'd at the # audit flush, rolling the edit back. Same failure mode as the # ``person_ids: list[UUID]`` bug on create. # ``exclude_unset`` (not ``exclude_none``) keeps an explicit # ``due_at: null`` clear visible in the audit trail. detail=body.model_dump(exclude_unset=True, mode="json"), ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() return success_response( data=await _hydrate_cycle(db, cycle), request=request, ) @router.delete("/cycles/{cycle_id}") async def delete_draft_cycle( request: Request, cycle_id: UUID, current_user: AuthUser = Depends(require_asset_admin()), db: AsyncSession = Depends(get_db), ): service = AssetInventoryService(db) await service.delete_draft(cycle_id) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="asset_inventory.cycle.delete", resource_type="asset_inventory_cycle", resource_id=str(cycle_id), status="success", ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() return success_response(data={"id": str(cycle_id)}, request=request) @router.get("/cycles/{cycle_id}/preview") async def preview_cycle( request: Request, cycle_id: UUID, current_user: AuthUser = Depends(require_asset_read()), db: AsyncSession = Depends(get_db), ): service = AssetInventoryService(db) preview = await service.preview_cycle(cycle_id) return success_response(data=preview, request=request) @router.post("/preview-filter") async def preview_filter( request: Request, body: CycleScopeFilter | None = Body(default=None), current_user: AuthUser = Depends(require_asset_admin()), db: AsyncSession = Depends(get_db), ): """Stateless dry-run for a candidate scope filter. Lets the create-cycle form show "this filter matches N assets, M distinct recipients" BEFORE the admin submits — the critical affordance for "I only want to test on my own laptop first". No cycle is created and no notifications are dispatched. A null body is treated as the company-wide filter (same semantics as posting an empty object). """ service = AssetInventoryService(db) preview = await service.preview_filter(body) return success_response(data=preview, request=request) @router.get("/lookups/departments") async def lookup_inventory_departments( request: Request, current_user: AuthUser = Depends(require_asset_read()), db: AsyncSession = Depends(get_db), ): """Department names eligible for inventory scope narrowing. Returns the same shape ``/api/v1/identity-sync/lookups/departments`` exposes (source_id + name + parent + path) but gated by ``require_asset_admin`` so asset admins — who don't hold ``it_admin`` — can populate the picker on the create-cycle form. Asset-admin already implies authority over per-person asset assignments, so surfacing department names here adds no new sensitive disclosure. """ from sqlalchemy import select as sa_select from app.identity.person_models import IdentityOrgNode rows = (await db.execute( sa_select(IdentityOrgNode).where( IdentityOrgNode.org_node_type == "department", IdentityOrgNode.deleted_at.is_(None), ).order_by(IdentityOrgNode.org_path) )).scalars().all() return success_response( data=[ { "source_id": r.org_node_source_id, "name": r.org_node_name, "parent_source_id": r.parent_org_node_source_id, "org_path": r.org_path, "level": r.org_level, } for r in rows ], request=request, ) @router.post("/cycles/{cycle_id}/start") async def start_cycle( request: Request, cycle_id: UUID, current_user: AuthUser = Depends(require_asset_admin()), db: AsyncSession = Depends(get_db), ): service = AssetInventoryService(db) cycle = await service.start_cycle( cycle_id, actor_user_id=current_user.id, trace_id=request.state.trace_id, ) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="asset_inventory.cycle.start", resource_type="asset_inventory_cycle", resource_id=str(cycle.id), status="success", detail={"total_items": cycle.total_items}, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() # The launch fan-out (one Feishu card per holder — hundreds of external # calls for a company-wide cycle) runs AFTER commit, off the request, so # the state transition never holds the cycle lock across the slow sends. # Rows are enqueued durably first, so the notification.outbox_retry # worker backstops any card this drain doesn't reach. _spawn_background( run_start_notifications_in_background( cycle.id, actor_user_id=current_user.id, trace_id=request.state.trace_id, ), name="asset_inventory.start_notify", ) return success_response( data=await _hydrate_cycle(db, cycle), request=request, ) @router.post("/cycles/{cycle_id}/close") async def close_cycle( request: Request, cycle_id: UUID, current_user: AuthUser = Depends(require_asset_admin()), db: AsyncSession = Depends(get_db), ): service = AssetInventoryService(db) cycle = await service.close_cycle( cycle_id, actor_person_id=current_user.id, ) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="asset_inventory.cycle.close", resource_type="asset_inventory_cycle", resource_id=str(cycle.id), status="success", # Closing a cycle freezes whatever ``pending`` items haven't # responded — the cycle is the system of record for "how many # confirmations / issues / unresponsive items did this round # produce." Stamp the per-bucket counts so an auditor can # answer "what was the participation rate of cycle X" without # re-aggregating from items. detail={ "total_items": cycle.total_items, "confirmed_count": cycle.confirmed_count, "issue_count": cycle.issue_count, "pending_count": cycle.pending_count, }, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() return success_response( data=await _hydrate_cycle(db, cycle), request=request, ) @router.post("/cycles/{cycle_id}/remind") async def send_reminder( request: Request, cycle_id: UUID, payload: CycleRemindRequest = Body(default_factory=CycleRemindRequest), current_user: AuthUser = Depends(require_asset_admin()), db: AsyncSession = Depends(get_db), ): service = AssetInventoryService(db) outcome = await service.send_cycle_reminder( cycle_id, actor_user_id=current_user.id, trace_id=request.state.trace_id, force=payload.force, notify_manager=payload.notify_manager, notify_hrbp=payload.notify_hrbp, ) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="asset_inventory.cycle.remind", resource_type="asset_inventory_cycle", resource_id=str(cycle_id), status="success", detail={ "recipients_notified": outcome.recipients_notified, "managers_notified": outcome.managers_notified, "hrbps_notified": outcome.hrbps_notified, "force": payload.force, "notify_manager": payload.notify_manager, "notify_hrbp": payload.notify_hrbp, }, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() return success_response( data={ "recipients_notified": outcome.recipients_notified, "managers_notified": outcome.managers_notified, "hrbps_notified": outcome.hrbps_notified, }, request=request, ) @router.get("/cycles/{cycle_id}/items") async def list_items( request: Request, cycle_id: UUID, holder_kind: str | None = Query(None), status: str | None = Query(None), holder_person_id: UUID | None = Query(None), page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200), current_user: AuthUser = Depends(require_asset_read()), db: AsyncSession = Depends(get_db), ): service = AssetInventoryService(db) rows, total = await service.list_items( cycle_id, holder_kind=holder_kind, status=status, holder_person_id=holder_person_id, page=page, page_size=page_size, ) # ``require_asset_read`` admits management / auditor + asset # admin tiers; only the manage-asset tier (global / asset_admin) # should see the raw holder-submitted ``new_location`` value. # Management + auditor are read-only oversight roles that don't # need the operator's free-text input. Round-6 review high. data = await _hydrate_items( db, rows, admin_view=can_manage_assets(current_user.role_names), ) return paginated_response( data=data, total=total, page=page, page_size=page_size, request=request, ) @router.post("/items/{item_id}/respond") async def respond_to_item( request: Request, item_id: UUID, body: ItemRespondRequest, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Record a recipient's response to an inventory item. Authorisation: holder of the item, OR an asset admin (admins handle the it_fallback bucket and may also respond on behalf of an absent assignee). Enforced inside ``respond_to_item``. """ service = AssetInventoryService(db) is_admin = can_manage_assets(current_user.role_names) # #1778: region-scope the ADMIN-override path. When an asset_admin # responds on someone else's item, gate the asset's own region so a # region-scoped admin can't act on an out-of-region item. Holder-self # responses are unaffected (a holder responding to their own item is # authorised regardless of region). Note: inventory responses no # longer move ownership — user-to-user transfers are asset-admin-only # via ``POST /assets/{id}/transfer`` / ``change-assignee`` — so there # is no transfer-target region check here anymore. if is_admin: from app.domains.asset.models import AssetInventoryItem from app.domains.asset.router import ( # lazy: avoid router import cycle _enforce_asset_region, ) inv_item = (await db.execute( select(AssetInventoryItem).where(AssetInventoryItem.id == item_id) )).scalar_one_or_none() if inv_item is not None and inv_item.holder_person_id != current_user.id: await _enforce_asset_region(db, current_user, inv_item.asset_id) item = await service.respond_to_item( item_id, actor_person_id=current_user.id, is_admin=is_admin, response_type=body.response_type, response_notes=body.response_notes, new_location=body.new_location, evidence_photos=body.evidence_photos, ) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="asset_inventory.item.respond", resource_type="asset_inventory_item", resource_id=str(item.id), status="success", detail={ "response_type": body.response_type, "is_admin_override": ( is_admin and item.holder_person_id != current_user.id ), # Narrower than ``is_admin_override``, which also fires for # the holder-less ``it_fallback`` bucket: this one is true # only for "IT 帮忙确认" — an admin answering an item that # WAS routed to a person. Matches the ``responded_on_behalf`` # flag the UI tags the row with, so the audit log and the # screen agree on what happened. "responded_on_behalf": ( item.holder_person_id is not None and item.holder_person_id != current_user.id ), }, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() # The responder either typed ``new_location`` themselves (holder # path) or is an asset admin (admin override path). Either way # they're authorised to see what they just submitted on the # callback, so ``admin_view=True`` is correct. hydrated = await _hydrate_items(db, [item], admin_view=True) return success_response(data=hydrated[0], request=request) @router.post("/items/{item_id}/resolve") async def resolve_issue( request: Request, item_id: UUID, body: ItemIssueResolveRequest, current_user: AuthUser = Depends(require_asset_admin()), db: AsyncSession = Depends(get_db), ): service = AssetInventoryService(db) # #1778: region-scope issue resolution — asset_admin is otherwise # global across regions here. Only the managing region (or global # asset scope) may resolve an item's issue. from app.domains.asset.models import AssetInventoryItem from app.domains.asset.router import ( # lazy: avoid router import cycle _enforce_asset_region, ) inv_item = (await db.execute( select(AssetInventoryItem).where(AssetInventoryItem.id == item_id) )).scalar_one_or_none() if inv_item is not None: await _enforce_asset_region(db, current_user, inv_item.asset_id) item = await service.resolve_issue( item_id, actor_person_id=current_user.id, notes=body.issue_resolution_notes, ) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="asset_inventory.item.resolve", resource_type="asset_inventory_item", resource_id=str(item.id), status="success", detail={"notes": body.issue_resolution_notes}, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() # Endpoint is gated to asset_admin only — admin_view stays True. hydrated = await _hydrate_items(db, [item], admin_view=True) return success_response(data=hydrated[0], request=request) # ── Employee endpoints ──────────────────────────────────────────────── @router.get("/me/pending") async def list_my_pending( request: Request, cycle_id: UUID | None = Query(None), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Items routed to the caller across active cycles (or one specific cycle when ``cycle_id`` is given). Includes already-responded items when scoped to a single cycle so the employee can see what they already submitted within the same cycle. """ service = AssetInventoryService(db) if cycle_id is not None: items, _ = await service.list_items( cycle_id, holder_person_id=current_user.id, page=1, page_size=200, ) else: items = await service.list_pending_for_person(current_user.id) # /me/pending only returns items where the caller is the holder # (``holder_person_id == current_user.id`` filter above), so the # ``new_location`` they see is THEIR OWN prior submission — not a # leak. ``admin_view=True`` preserves the existing edit-response # flow (the holder can re-open the RespondModal and see what they # said last time). data = await _hydrate_items(db, items, admin_view=True) return success_response(data=data, request=request) @router.get("/me/active-cycles") async def list_active_cycles_for_me( request: Request, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Active cycles in which the caller has at least one item. Used by the header notification badge / employee landing tile so we can show "you have 2 inventory rounds open" without loading every item. """ rows = ( await db.execute( select(AssetInventoryCycle).where( AssetInventoryCycle.status == "active", AssetInventoryCycle.id.in_( select(AssetInventoryItem.cycle_id).where( AssetInventoryItem.holder_person_id == current_user.id, ) ), ) ) ).scalars().all() data = [await _hydrate_cycle(db, c) for c in rows] return success_response(data=data, request=request)