Coverage for app/routers/nodes/label.py: 91%

1from app.config.config import NEO4J_HOST, NEO4J_PORT, NEO4J_USER, NEO4J_PASSWORD, RESERVED_NODE_LABEL_NAMES

2from fastapi import APIRouter

3from fastapi import Depends, Path, Query

4from fastapi.security import OAuth2PasswordBearer

5from neo4j import GraphDatabase, basic_auth

6from loguru import logger

8from app.utils.check_token import check_constellation_access, constellation_check_error

9from app.utils.send_sse_notification import send_constellation_notification

10from app.utils.response_format import generate_response, generate_error_response

11from app.utils.decode_ydoc import decode_attributes_inplace, DecodeType

12from app.utils.typing import Optional, List, JSONValue, serialize_value

14router = APIRouter(tags=["Node Label"])

16# Define the OAuth2 schema to get the JWT

17oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

19# Define the Neo4j driver

20driver = GraphDatabase.driver(

21 f"neo4j://{NEO4J_HOST}:{NEO4J_PORT}",

22 auth=basic_auth(NEO4J_USER, NEO4J_PASSWORD))

24# Get all labels

25@router.get("/constellation/{constellation_uuid}/labels",

26 summary="Get all labels in the constellation",

27 description="Get all labels in the constellation specified by its UUID",

28 response_description="The list of labels",

29 responses={

30 200: {

31 "description": "The list of labels",

32 "content": {

33 "application/json": {

34 "example": {

35 "success": True,

36 "data": [

37 "Node",

38 "Person",

39 "Movie",

40 "Book"

41 ],

42 "message": "Labels successfully retrieved",

43 "error": None

44 }

45 }

46 }

47 },

48 **constellation_check_error

49 }

50)

51async def read_labels(constellation_uuid: str = Path(..., title="The UUID of the constellation"),

52 token: str = Depends(oauth2_scheme)):

53 test = check_constellation_access(token, constellation_uuid, "READ")

54 if test is not None:

55 return test

57 with driver.session() as session:

58 result = session.run("""

59 MATCH (n {constellation_uuid: $constellation_uuid})

60 WHERE ALL(label IN $RESERVED_NODE_LABEL_NAMES WHERE NOT label IN labels(n))

61 UNWIND labels(n) AS label

62 RETURN DISTINCT label

63 """,

64 constellation_uuid=constellation_uuid,

65 RESERVED_NODE_LABEL_NAMES=RESERVED_NODE_LABEL_NAMES

66 )

67 final_result: JSONValue = [record["label"] for record in result]

68 driver.close()

70 return generate_response(

71 status_code=200,

72 data=final_result,

73 message="Labels successfully retrieved"

74 )

76# Get all label of a specific node

77@router.get("/constellation/{constellation_uuid}/node/{node_uuid}/labels",

78 summary="Get all labels of a node",

79 description="Get all labels of the node specified by its UUID",

80 response_description="The list of labels",

81 responses={

82 200: {

83 "description": "The list of labels",

84 "content": {

85 "application/json": {

86 "example": {

87 "success": True,

88 "data": [

89 [

90 "Node",

91 "Person"

92 ]

93 ],

94 "message": "Labels successfully retrieved",

95 "error": None

96 }

97 }

98 }

99 },

100 **constellation_check_error

101 }

102)

103async def read_node_labels(constellation_uuid: str = Path(..., title="The UUID of the constellation"),

104 node_uuid: str = Path(..., title="The UUID of the node"),

105 token: str = Depends(oauth2_scheme)):

106 test = check_constellation_access(token, constellation_uuid, "READ")

107 if test is not None:

108 return test

109

110 with driver.session() as session:

111 # check if the node exists

112 result = session.run("""

113 MATCH (n {constellation_uuid: $constellation_uuid, node_uuid: $node_uuid})

114 WHERE ALL(label IN $RESERVED_NODE_LABEL_NAMES WHERE NOT label IN labels(n))

115 RETURN n

116 """,

117 constellation_uuid=constellation_uuid,

118 node_uuid=node_uuid,

119 RESERVED_NODE_LABEL_NAMES=RESERVED_NODE_LABEL_NAMES

120 )

121 if len(result.data()) == 0:

122 return generate_error_response(

123 status_code=404,

124 error_code="NODE_NOT_FOUND",

125 error_message=f"Node {node_uuid} not found",

126 message=f"Node {node_uuid} not found"

127 )

128

129 result = session.run("""

130 MATCH (n {constellation_uuid: $constellation_uuid, node_uuid: $node_uuid})

131 WHERE ALL(label IN $RESERVED_NODE_LABEL_NAMES WHERE NOT label IN labels(n))

132 RETURN labels(n)

133 """,

134 constellation_uuid=constellation_uuid,

135 node_uuid=node_uuid,

136 RESERVED_NODE_LABEL_NAMES=RESERVED_NODE_LABEL_NAMES

137 )

138 final_result: JSONValue = [record["labels(n)"] for record in result]

139 driver.close()

140

141 return generate_response(

142 status_code=200,

143 data=final_result,

144 message="Labels successfully retrieved"

145 )

146

147# Get all nodes with a specific label

148@router.get("/constellation/{constellation_uuid}/nodes/label/{label}",

149 summary="Get all nodes with a label",

150 description="Get all nodes with the label specified in parameter",

151 response_description="The list of nodes",

152 responses={

153 200: {

154 "description": "The list of nodes",

155 "content": {

156 "application/json": {

157 "example": {

158 "success": True,

159 "data": [

160 {

161 "attributes": {

162 "node_uuid": 1,

163 "title": "example_title",

164 "content": "example_content"

165 },

166 "labels": [

167 "Node"

168 ]

169 }

170 ],

171 "message": "Nodes successfully retrieved",

172 "error": None

173 }

174 }

175 }

176 },

177 **constellation_check_error

178 }

179)

180async def read_nodes_with_label(constellation_uuid: str = Path(..., title="The UUID of the constellation"),

181 label: str = Path(..., title="The label to search for"),

182 decode: Optional[DecodeType] = Query(default=None, description="Whether to decode the attributes or not. Options: 'xml', 'plain_text'"),

183 in_filter: Optional[List[str]] = Query(default=None, description="Filter to include only specific attributes in the final results"),

184 out_filter: Optional[List[str]] = Query(default=None, description="Filter to exclude specific attributes from the final results"),

185 token: str = Depends(oauth2_scheme)):

186 test = check_constellation_access(token, constellation_uuid, "READ")

187 if test is not None:

188 return test

189

190 if in_filter is None:

191 in_filter = []

192 if out_filter is None:

193 out_filter = []

194

195 if label in RESERVED_NODE_LABEL_NAMES:

196 return generate_error_response(

197 status_code=400,

198 error_code="RESERVED_LABEL_NAME",

199 error_message=f"Label {label} is a reserved label name",

200 message=f"Label {label} is a reserved label name"

201 )

202

203 with driver.session() as session:

204 result = session.run("""

205 MATCH (n:$($label) {constellation_uuid: $constellation_uuid})

206 WHERE ALL(label IN $RESERVED_NODE_LABEL_NAMES WHERE NOT label IN labels(n))

207 RETURN properties(n) AS attributes, labels(n) as labels

208 """,

209 label=label,

210 constellation_uuid=constellation_uuid,

211 RESERVED_NODE_LABEL_NAMES=RESERVED_NODE_LABEL_NAMES

212 )

213 final_result: JSONValue = [{"attributes": {

214 key: serialize_value(value)

215 for key, value in record["attributes"].items()

216 if (len(in_filter) == 0 or key in in_filter) and (len(out_filter) == 0 or key not in out_filter)

217 }, "labels": record["labels"]} for record in result]

218 driver.close()

219

220 final_result = decode_attributes_inplace(final_result, decode)

221

222 return generate_response(

223 status_code=200,

224 data=final_result,

225 message="Nodes successfully retrieved"

226 )

227

228# Add a label to a node

229@router.post("/constellation/{constellation_uuid}/node/{node_uuid}/label/{label}",

230 summary="Add a label to a node",

231 description="Add the label specified in parameter to the node specified by its UUID",

232 response_description="The node with the new label",

233 responses={

234 200: {

235 "description": "The node with the new label",

236 "content": {

237 "application/json": {

238 "example": {

239 "success": True,

240 "data": [

241 {

242 "attributes": {

243 "node_uuid": 1,

244 "title": "example_title",

245 "content": "example_content"

246 },

247 "labels": [

248 "Node"

249 ]

250 }

251 ],

252 "message": "Label successfully added",

253 "error": None

254 }

255 }

256 }

257 },

258 **constellation_check_error

259 }

260)

261async def add_label(constellation_uuid: str = Path(..., title="The UUID of the constellation"),

262 node_uuid: str = Path(..., title="The UUID of the node"),

263 label: str = Path(..., title="The label to add"),

264 decode: Optional[DecodeType] = Query(default=None, description="Whether to decode the attributes or not. Options: 'xml', 'plain_text'"),

265 in_filter: Optional[List[str]] = Query(default=None, description="Filter to include only specific attributes in the final results"),

266 out_filter: Optional[List[str]] = Query(default=None, description="Filter to exclude specific attributes from the final results"),

267 token: str = Depends(oauth2_scheme)):

268 test = check_constellation_access(token, constellation_uuid, "WRITE")

269 if test is not None:

270 return test

271

272 if in_filter is None:

273 in_filter = []

274 if out_filter is None:

275 out_filter = []

276

277 if label in RESERVED_NODE_LABEL_NAMES:

278 return generate_error_response(

279 status_code=400,

280 error_code="RESERVED_LABEL_NAME",

281 error_message=f"Label {label} is a reserved label name",

282 message=f"Label {label} is a reserved label name"

283 )

284

285 with driver.session() as session:

286 # check if the node exists

287 result = session.run("""

288 MATCH (n {constellation_uuid: $constellation_uuid, node_uuid: $node_uuid})

289 WHERE ALL(label IN $RESERVED_NODE_LABEL_NAMES WHERE NOT label IN labels(n))

290 RETURN n

291 """,

292 constellation_uuid=constellation_uuid,

293 node_uuid=node_uuid,

294 RESERVED_NODE_LABEL_NAMES=RESERVED_NODE_LABEL_NAMES

295 )

296 if len(result.data()) == 0:

297 return generate_error_response(

298 status_code=404,

299 error_code="NODE_NOT_FOUND",

300 error_message=f"Node {node_uuid} not found",

301 message=f"Node {node_uuid} not found"

302 )

303

304 # check if the label already exists on the node

305 label_check = session.run("""

306 MATCH (n {constellation_uuid: $constellation_uuid, node_uuid: $node_uuid})

307 WHERE ALL(label IN $RESERVED_NODE_LABEL_NAMES WHERE NOT label IN labels(n))

308 RETURN labels(n) as label

309 """,

310 constellation_uuid=constellation_uuid,

311 node_uuid=node_uuid,

312 RESERVED_NODE_LABEL_NAMES=RESERVED_NODE_LABEL_NAMES

313 )

314 if label in [record["label"] for record in label_check][0]:

315 return generate_error_response(

316 status_code=400,

317 error_code="LABEL_ALREADY_EXISTS",

318 error_message=f"Label {label} already exists on node {node_uuid}",

319 message=f"Label {label} already exists on node {node_uuid}"

320 )

321

322 # add the label to the node

323 result = session.run("""

324 MATCH (n {constellation_uuid: $constellation_uuid, node_uuid: $node_uuid})

325 WHERE ALL(label IN $RESERVED_NODE_LABEL_NAMES WHERE NOT label IN labels(n))

326 SET n:$($label)

327 SET n.created_at = coalesce(n.created_at, datetime())

328 SET n.updated_at = datetime()

329 SET n.label_count = size(labels(n))

330 RETURN properties(n) AS attributes, labels(n) as labels

331 """,

332 constellation_uuid=constellation_uuid,

333 node_uuid=node_uuid,

334 RESERVED_NODE_LABEL_NAMES=RESERVED_NODE_LABEL_NAMES,

335 label=label

336 )

337 final_result: JSONValue = [{"attributes": {

338 key: serialize_value(value)

339 for key, value in record["attributes"].items()

340 if (len(in_filter) == 0 or key in in_filter) and (len(out_filter) == 0 or key not in out_filter)

341 }, "labels": record["labels"]} for record in result]

342 driver.close()

343

344 final_result = decode_attributes_inplace(final_result, decode)

345

346 # Send a notification to the SSE server

347 notification_result = send_constellation_notification(

348 constellation_uuid=constellation_uuid,

349 data=final_result,

350 message="Node label added",

351 data_type="NODE_LABEL_ADDED",

352 token=token

353 )

354 if not notification_result:

355 logger.warning("Add node label: Error sending notification to SSE server")

356

357 return generate_response(

358 status_code=200,

359 data=final_result,

360 message="Label successfully added"

361 )

362

363# Remove a label from a node

364@router.delete("/constellation/{constellation_uuid}/node/{node_uuid}/label/{label}",

365 summary="Remove a label from a node",

366 description="Remove the label specified in parameter from the node specified by its UUID",

367 response_description="The node without the label",

368 responses={

369 200: {

370 "description": "The node without the label",

371 "content": {

372 "application/json": {

373 "example": {

374 "success": True,

375 "data": [

376 {

377 "attributes": {

378 "node_uuid": 1,

379 "title": "example_title",

380 "content": "example_content"

381 },

382 "labels": [

383 "Node"

384 ]

385 }

386 ],

387 "message": "Label successfully removed",

388 "error": None

389 }

390 }

391 }

392 },

393 **constellation_check_error

394 }

395)

396async def remove_label(constellation_uuid: str = Path(..., title="The UUID of the constellation"),

397 node_uuid: str = Path(..., title="The UUID of the node"),

398 label: str = Path(..., title="The label to remove"),

399 decode: Optional[DecodeType] = Query(default=None, description="Whether to decode the attributes or not. Options: 'xml', 'plain_text'"),

400 in_filter: Optional[List[str]] = Query(default=None, description="Filter to include only specific attributes in the final results"),

401 out_filter: Optional[List[str]] = Query(default=None, description="Filter to exclude specific attributes from the final results"),

402 token: str = Depends(oauth2_scheme)):

403 test = check_constellation_access(token, constellation_uuid, "WRITE")

404 if test is not None:

405 return test

406

407 if in_filter is None:

408 in_filter = []

409 if out_filter is None:

410 out_filter = []

411

412 with driver.session() as session:

413 # check if the node exists

414 result = session.run("""

415 MATCH (n {constellation_uuid: $constellation_uuid, node_uuid: $node_uuid})

416 WHERE ALL(label IN $RESERVED_NODE_LABEL_NAMES WHERE NOT label IN labels(n))

417 RETURN n

418 """,

419 constellation_uuid=constellation_uuid,

420 node_uuid=node_uuid,

421 RESERVED_NODE_LABEL_NAMES=RESERVED_NODE_LABEL_NAMES

422 )

423 if len(result.data()) == 0:

424 return generate_error_response(

425 status_code=404,

426 error_code="NODE_NOT_FOUND",

427 error_message=f"Node {node_uuid} not found",

428 message=f"Node {node_uuid} not found"

429 )

430

431 # check if the label exists on the node

432 label_check = session.run("""

433 MATCH (n {constellation_uuid: $constellation_uuid, node_uuid: $node_uuid})

434 WHERE ALL(label IN $RESERVED_NODE_LABEL_NAMES WHERE NOT label IN labels(n))

435 RETURN labels(n) as label

436 """,

437 constellation_uuid=constellation_uuid,

438 node_uuid=node_uuid,

439 RESERVED_NODE_LABEL_NAMES=RESERVED_NODE_LABEL_NAMES

440 )

441 if label not in [record["label"] for record in label_check][0]:

442 return generate_error_response(

443 status_code=404,

444 error_code="LABEL_NOT_FOUND",

445 error_message=f"Label {label} not found on node {node_uuid}",

446 message=f"Label {label} not found on node {node_uuid}"

447 )

448

449 # remove the label from the node

450 result = session.run("""

451 MATCH (n {constellation_uuid: $constellation_uuid, node_uuid: $node_uuid})

452 WHERE ALL(label IN $RESERVED_NODE_LABEL_NAMES WHERE NOT label IN labels(n))

453 REMOVE n:$($label)

454 SET n.created_at = coalesce(n.created_at, datetime())

455 SET n.updated_at = datetime()

456 SET n.label_count = size(labels(n))

457 RETURN properties(n) AS attributes, labels(n) as labels

458 """,

459 constellation_uuid=constellation_uuid,

460 node_uuid=node_uuid,

461 RESERVED_NODE_LABEL_NAMES=RESERVED_NODE_LABEL_NAMES,

462 label=label

463 )

464 final_result: JSONValue = [{"attributes": {

465 key: serialize_value(value)

466 for key, value in record["attributes"].items()

467 if (len(in_filter) == 0 or key in in_filter) and (len(out_filter) == 0 or key not in out_filter)

468 }, "labels": record["labels"]} for record in result]

469 driver.close()

470

471 final_result = decode_attributes_inplace(final_result, decode)

472

473 # Send a notification to the SSE server

474 notification_result = send_constellation_notification(

475 constellation_uuid=constellation_uuid,

476 data=final_result,

477 message="Node label deleted",

478 data_type="NODE_LABEL_DELETED",

479 token=token

480 )

481 if not notification_result:

482 logger.warning("Remove node label: Error sending notification to SSE server")

483

484 return generate_response(

485 status_code=200,

486 data=final_result,

487 message="Label successfully deleted"

488 )

489

490# Remove all labels from a node

491@router.delete("/constellation/{constellation_uuid}/node/{node_uuid}/labels",

492 summary="Remove all labels from a node",

493 description="Remove all labels from the node specified by its UUID",

494 response_description="The node without any label",

495 responses={

496 200: {

497 "description": "The node without any label",

498 "content": {

499 "application/json": {

500 "example": {

501 "success": True,

502 "data": [

503 {

504 "attributes": {

505 "node_uuid": 1,

506 "title": "example_title",

507 "content": "example_content"

508 },

509 "labels": [

510 "Node"

511 ]

512 }

513 ],

514 "message": "All labels successfully removed",

515 "error": None

516 }

517 }

518 }

519 },

520 **constellation_check_error

521 }

522)

523async def remove_all_labels(constellation_uuid: str = Path(..., title="The UUID of the constellation"),

524 node_uuid: str = Path(..., title="The UUID of the node"),

525 decode: Optional[DecodeType] = Query(default=None, description="Whether to decode the attributes or not. Options: 'xml', 'plain_text'"),

526 in_filter: Optional[List[str]] = Query(default=None, description="Filter to include only specific attributes in the final results"),

527 out_filter: Optional[List[str]] = Query(default=None, description="Filter to exclude specific attributes from the final results"),

528 token: str = Depends(oauth2_scheme)):

529 test = check_constellation_access(token, constellation_uuid, "WRITE")

530 if test is not None:

531 return test

532

533 if in_filter is None:

534 in_filter = []

535 if out_filter is None:

536 out_filter = []

537

538 with driver.session() as session:

539 # check if the node exists

540 result = session.run("""

541 MATCH (n {constellation_uuid: $constellation_uuid, node_uuid: $node_uuid})

542 WHERE ALL(label IN $RESERVED_NODE_LABEL_NAMES WHERE NOT label IN labels(n))

543 RETURN n

544 """,

545 constellation_uuid=constellation_uuid,

546 node_uuid=node_uuid,

547 RESERVED_NODE_LABEL_NAMES=RESERVED_NODE_LABEL_NAMES

548 )

549 if len(result.data()) == 0:

550 return generate_error_response(

551 status_code=404,

552 error_code="NODE_NOT_FOUND",

553 error_message=f"Node {node_uuid} not found",

554 message=f"Node {node_uuid} not found"

555 )

556

557 # Get all labels of the node

558 current_labels = session.run("""

559 MATCH (n {constellation_uuid: $constellation_uuid, node_uuid: $node_uuid})

560 WHERE ALL(label IN $RESERVED_NODE_LABEL_NAMES WHERE NOT label IN labels(n))

561 RETURN labels(n)

562 """,

563 constellation_uuid=constellation_uuid,

564 node_uuid=node_uuid,

565 RESERVED_NODE_LABEL_NAMES=RESERVED_NODE_LABEL_NAMES

566 )

567 current_labels = [record["labels(n)"] for record in current_labels][0]

568

569 if len(current_labels) == 0:

570 return generate_error_response(

571 status_code=400,

572 error_code="NO_LABELS_TO_REMOVE",

573 error_message="No labels to remove from the node",

574 message="No labels to remove from the node"

575 )

576

577 result = session.run("""

578 MATCH (n {constellation_uuid: $constellation_uuid, node_uuid: $node_uuid})

579 WHERE ALL(label IN $RESERVED_NODE_LABEL_NAMES WHERE NOT label IN labels(n))

580 REMOVE n:$($labels)

581 SET n.created_at = coalesce(n.created_at, datetime())

582 SET n.updated_at = datetime()

583 SET n.label_count = size(labels(n))

584 RETURN properties(n) AS attributes, labels(n) as labels

585 """,

586 constellation_uuid=constellation_uuid,

587 node_uuid=node_uuid,

588 RESERVED_NODE_LABEL_NAMES=RESERVED_NODE_LABEL_NAMES,

589 labels=current_labels

590 )

591 final_result: JSONValue = [{"attributes": {

592 key: serialize_value(value)

593 for key, value in record["attributes"].items()

594 if (len(in_filter) == 0 or key in in_filter) and (len(out_filter) == 0 or key not in out_filter)

595 }, "labels": record["labels"]} for record in result]

596 driver.close()

597

598 final_result = decode_attributes_inplace(final_result, decode)

599

600 # Send a notification to the SSE server

601 notification_result = send_constellation_notification(

602 constellation_uuid=constellation_uuid,

603 data=final_result,

604 message="Node labels deleted",

605 data_type="NODE_LABELS_DELETED",

606 token=token

607 )

608 if not notification_result:

609 logger.warning("Remove all node labels: Error sending notification to SSE server")

610

611 return generate_response(

612 status_code=200,

613 data=final_result,

614 message="All labels successfully deleted"

615 )