inecs
f8d3cbdd59
Add the FastAPI app, PostgreSQL migrations, Docker/Helm packaging, API contracts, docs, client examples, and the unit/integration/compatibility test suite for local client and tooling labs without a real vCenter.
81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
"""HttpNfcLease-compatible upload endpoints for OVF/VMDK lab transfers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, Request, Response
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from app.db.pool import Database
|
|
from app.dependencies import get_database
|
|
from app.vsphere.domain import platform_surface
|
|
from app.vsphere.errors import not_found
|
|
from app.vsphere.security.authz import require_privilege, require_read
|
|
from app.vsphere.security.session import SessionInfo
|
|
|
|
router = APIRouter(tags=["vSphere NFC"])
|
|
|
|
|
|
@router.get("/nfc/{lease_id}")
|
|
async def nfc_lease_status(
|
|
lease_id: str,
|
|
database: Database = Depends(get_database),
|
|
_: SessionInfo = Depends(require_read),
|
|
) -> dict[str, Any]:
|
|
lease = await platform_surface.get_nfc_lease(database, lease_id)
|
|
if lease is None:
|
|
raise not_found(f"NFC lease {lease_id} not found")
|
|
return lease
|
|
|
|
|
|
@router.post("/nfc/{lease_id}/complete")
|
|
async def nfc_lease_complete(
|
|
lease_id: str,
|
|
database: Database = Depends(get_database),
|
|
_: SessionInfo = Depends(require_privilege("VirtualMachine.Provisioning.DeployTemplate")),
|
|
) -> dict[str, Any]:
|
|
lease = await platform_surface.complete_nfc_lease(database, lease_id)
|
|
if lease is None:
|
|
raise not_found(f"NFC lease {lease_id} not found")
|
|
return lease
|
|
|
|
|
|
@router.put("/nfc/{lease_id}/files/{filename:path}")
|
|
@router.post("/nfc/{lease_id}/files/{filename:path}")
|
|
async def nfc_upload_file(
|
|
lease_id: str,
|
|
filename: str,
|
|
request: Request,
|
|
database: Database = Depends(get_database),
|
|
_: SessionInfo = Depends(require_privilege("VirtualMachine.Provisioning.DeployTemplate")),
|
|
) -> JSONResponse:
|
|
body = await request.body()
|
|
lease = await platform_surface.upload_nfc_file(database, lease_id, filename, len(body))
|
|
if lease is None:
|
|
raise not_found(f"NFC lease {lease_id} not found")
|
|
return JSONResponse(
|
|
content={
|
|
"lease": lease_id,
|
|
"filename": filename,
|
|
"bytes": len(body),
|
|
"state": lease.get("state"),
|
|
"transferProgress": lease.get("transferProgress"),
|
|
},
|
|
status_code=200,
|
|
)
|
|
|
|
|
|
@router.get("/nfc/{lease_id}/files/{filename:path}")
|
|
async def nfc_download_file(
|
|
lease_id: str,
|
|
filename: str,
|
|
database: Database = Depends(get_database),
|
|
_: SessionInfo = Depends(require_read),
|
|
) -> Response:
|
|
lease = await platform_surface.get_nfc_lease(database, lease_id)
|
|
if lease is None:
|
|
raise not_found(f"NFC lease {lease_id} not found")
|
|
content = f"# lab nfc placeholder for {filename} on {lease_id}\n".encode()
|
|
return Response(content=content, media_type="application/octet-stream")
|