Coverage for lode / api.py: 0%

1# lode/api.py 

2from fastapi import FastAPI, File, UploadFile, Form, Request 

3from fastapi.responses import HTMLResponse 

4from fastapi.templating import Jinja2Templates 

5from fastapi.staticfiles import StaticFiles 

6from enum import Enum 

7from typing import Optional 

8import tempfile 

9import os 

10import traceback 

11import logging 

12 

13# Configura logging 

14logging.basicConfig( 

15 level=logging.DEBUG, 

16 format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' 

17) 

18logger = logging.getLogger(__name__) 

19 

20# Internal modules 

21from lode.reader import Reader 

22 

23app = FastAPI(title="LODE 2.0 API", version="1.0.0") 

24templates = Jinja2Templates(directory="lode/templates") 

25app.mount("/static", StaticFiles(directory="lode/static"), name="static") 

26 

27class ReadAsFormat(str, Enum): 

28 owl = "owl" 

29 rdf = "rdf" 

30 skos = "skos" 

31 

32@app.get("/extract", response_class=HTMLResponse) 

33async def extract_get( 

34 request: Request, 

35 read_as: ReadAsFormat, 

36 url: str, 

37 resource: Optional[str] = None, 

38 lang: Optional[str] = None, 

39 imported: Optional[bool] = None, 

40 closure: Optional[bool] = None 

41): 

42 """Visualizza semantic artefact da URL.""" 

43 

44 try: 

45 logger.info(f"=== REQUEST START ===") 

46 logger.info(f"URL: {url}") 

47 logger.info(f"Format: {read_as.value}") 

48 logger.info(f"Resource: {resource}") 

49 

50 reader = Reader() 

51 reader.load_instances(url, read_as.value, imported=imported, closure=closure) 

52 

53 viewer = reader.get_viewer() 

54 data = viewer.get_view_data(resource_uri=resource, language=lang) 

55 

56 logger.info(f"=== REQUEST SUCCESS ===") 

57 return templates.TemplateResponse("viewer.html", { 

58 "request": request, 

59 **data 

60 }) 

61 except Exception as e: 

62 logger.error(f"=== ERROR ===") 

63 logger.error(f"Type: {type(e).__name__}") 

64 logger.error(f"Message: {str(e)}") 

65 logger.error(f"Traceback:\n{traceback.format_exc()}") 

66 logger.error(f"=============") 

67 

68 return templates.TemplateResponse("viewer.html", { 

69 "request": request, 

70 "error": f"{type(e).__name__}: {str(e)}\n\nFull traceback:\n{traceback.format_exc()}" 

71 }) 

72 

73@app.post("/extract", response_class=HTMLResponse) 

74async def extract_post( 

75 request: Request, 

76 read_as: ReadAsFormat = Form(...), 

77 file: UploadFile = File(...), 

78 resource: Optional[str] = Form(None), 

79 lang: Optional[str] = None, 

80 imported: Optional[str] = Form(None), 

81 closure: Optional[str] = Form(None) 

82): 

83 """Visualizza semantic artefact da file.""" 

84 temp_file_path = None 

85 

86 try: 

87 logger.info(f"=== FILE UPLOAD START ===") 

88 logger.info(f"Filename: {file.filename}") 

89 logger.info(f"Format: {read_as.value}") 

90 

91 with tempfile.NamedTemporaryFile(delete=False, suffix=".rdf") as tmp: 

92 content = await file.read() 

93 tmp.write(content) 

94 temp_file_path = tmp.name 

95 

96 reader = Reader() 

97 reader.load_instances(temp_file_path, read_as.value, imported=imported, closure=closure) 

98 

99 viewer = reader.get_viewer() 

100 data = viewer.get_view_data(resource_uri=resource, language=lang) 

101 

102 logger.info(f"=== UPLOAD SUCCESS ===") 

103 return templates.TemplateResponse("viewer.html", { 

104 "request": request, 

105 **data 

106 }) 

107 except Exception as e: 

108 logger.error(f"=== ERROR ===") 

109 logger.error(f"Type: {type(e).__name__}") 

110 logger.error(f"Message: {str(e)}") 

111 logger.error(f"Traceback:\n{traceback.format_exc()}") 

112 logger.error(f"=============") 

113 

114 return templates.TemplateResponse("viewer.html", { 

115 "request": request, 

116 "error": f"{type(e).__name__}: {str(e)}\n\nFull traceback:\n{traceback.format_exc()}" 

117 }) 

118 finally: 

119 if temp_file_path and os.path.exists(temp_file_path): 

120 os.unlink(temp_file_path) 

121 

122@app.get("/info") 

123async def root(): 

124 return { 

125 "message": "LODE 2.0 API", 

126 "version": "1.0.0", 

127 "endpoints": { 

128 "extract": "/extract [GET] - View resources as HTML" 

129 } 

130 } 

131 

132@app.get("/", response_class=HTMLResponse) 

133async def input_web_interface(request: Request): 

134 """Interfaccia web per l'API""" 

135 return templates.TemplateResponse("index.html", { 

136 "request": request, 

137 "formats": [format.value for format in ReadAsFormat] 

138 }) 

139 

140@app.get("/health") 

141async def health_check(): 

142 return {"status": "ok"} 

143 

144if __name__ == "__main__": 

145 import uvicorn 

146 uvicorn.run(app, host="0.0.0.0", port=8000)