Get an ExecutionToken by token_id.
(
token_id: str,
db: Annotated[AsyncSession, Depends(get_db)],
)
| 477 | description="Retrieve an ExecutionToken by token_id.", |
| 478 | ) |
| 479 | async def get_token( |
| 480 | token_id: str, |
| 481 | db: Annotated[AsyncSession, Depends(get_db)], |
| 482 | ) -> ExecutionTokenOut: |
| 483 | """Get an ExecutionToken by token_id.""" |
| 484 | svc = ExecutionTokenService(db) |
| 485 | token = await svc.get_by_id(token_id) |
| 486 | if token is None: |
| 487 | raise HTTPException(status_code=404, detail=f"ExecutionToken not found: {token_id}") |
| 488 | out = ExecutionTokenOut( |
| 489 | token_id=token.token_id, |
| 490 | agent_id=token.agent_id, |
| 491 | mission_id=token.mission_id, |
| 492 | execution_scope=token.execution_scope, |
| 493 | immutable_params=token.immutable_params, |
| 494 | bounded_params=token.bounded_params, |
| 495 | issued_at=token.issued_at, |
| 496 | expires_at=token.expires_at, |
| 497 | boundary_snapshot_id=token.boundary_snapshot_id, |
| 498 | token_signature=token.token_signature, |
| 499 | used=token.used, |
| 500 | is_valid=token.is_valid, |
| 501 | invalidated_at=token.invalidated_at, |
| 502 | invalidation_reason=token.invalidation_reason, |
| 503 | ) |
| 504 | return out |
| 505 | |
| 506 | |
| 507 | @_token.post( |
nothing calls this directly
no test coverage detected