Coverage for volexport/api.py: 88%

1from logging import getLogger 

2from subprocess import SubprocessError 

3from fastapi import FastAPI, Request 

4from fastapi.responses import JSONResponse 

5from .api_export import router as export_router 

6from .api_volume import router as volume_router 

7from .api_mgmt import router as mgmt_router 

8from .exceptions import InvalidArgument 

9 

10_log = getLogger(__name__) 

11api = FastAPI() 

12api.include_router(export_router) 

13api.include_router(volume_router) 

14api.include_router(mgmt_router) 

15 

16 

17@api.exception_handler(FileNotFoundError) 

18def notfound(request: Request, exc: FileNotFoundError): 

19 """FileNotFoundError to 404 Not Found""" 

20 _log.info("not found: request=%s %s", request.method, request.url.path, exc_info=exc) 

21 return JSONResponse(status_code=404, content=dict(detail="\n".join(exc.args))) 

22 

23 

24@api.exception_handler(FileExistsError) 

25def inuse(request: Request, exc: FileExistsError): 

26 """FileExistsError to 400 Bad Request""" 

27 _log.info("file exists: request=%s %s", request.method, request.url.path, exc_info=exc) 

28 return JSONResponse(status_code=400, content=dict(detail="\n".join(exc.args))) 

29 

30 

31@api.exception_handler(NotImplementedError) 

32def notimplemented(request: Request, exc: NotImplementedError): 

33 """NotImplementedError to 501 Not Implemented""" 

34 _log.info("not implemented: request=%s %s", request.method, request.url.path, exc_info=exc) 

35 return JSONResponse(status_code=501, content=dict(detail=str(exc))) 

36 

37 

38@api.exception_handler(SubprocessError) 

39def commanderror(request: Request, exc: SubprocessError): 

40 """SubprocessError to 500 Internal Server Error""" 

41 _log.info("command error: request=%s %s", request.method, request.url.path, exc_info=exc) 

42 return JSONResponse(status_code=500, content=dict(detail="internal error")) 

43 

44 

45@api.exception_handler(InvalidArgument) 

46def badrequest(request: Request, exc: InvalidArgument): 

47 """InvalidArgument to 400 Bad Request""" 

48 _log.info("invalid argument: request=%s %s", request.method, request.url.path, exc_info=exc) 

49 return JSONResponse(status_code=400, content=dict(detail="\n".join(exc.args))) 

50 

51 

52@api.exception_handler(ValueError) 

53def valueerror(request: Request, exc: ValueError): 

54 """ValueError to 400 Bad Request""" 

55 _log.info("invalid argument: request=%s %s", request.method, request.url.path, exc_info=exc) 

56 return JSONResponse(status_code=400, content=dict(detail="\n".join(exc.args))) 

57 

58 

59@api.exception_handler(TypeError) 

60def typeerror(request: Request, exc: TypeError): 

61 """TypeError to 500 Internal Server Error""" 

62 _log.info("internal error: request=%s %s", request.method, request.url.path, exc_info=exc) 

63 return JSONResponse(status_code=500, content=dict(detail="internal error")) 

64 

65 

66@api.exception_handler(AssertionError) 

67def asserterror(request: Request, exc: AssertionError): 

68 """AssertionError to 500 Internal Server Error""" 

69 _log.info("internal error: request=%s %s", request.method, request.url.path, exc_info=exc) 

70 return JSONResponse(status_code=500, content=dict(detail="internal error")) 

71 

72 

73@api.get("/health", description="Health check endpoint") 

74def health(): 

75 return {"status": "OK"}