Coverage for pepdbagent/modules/schema.py: 82%

1import logging 

2 

3from sqlalchemy import Select, and_, func, or_, select 

4from sqlalchemy.orm import Session 

5from sqlalchemy.orm.attributes import flag_modified 

6 

7from pepdbagent.const import DEFAULT_TAG_VERSION, LATEST_SCHEMA_VERSION, PKG_NAME 

8from pepdbagent.db_utils import ( 

9 BaseEngine, 

10 SchemaRecords, 

11 SchemaTags, 

12 SchemaVersions, 

13 User, 

14) 

15from pepdbagent.exceptions import ( 

16 SchemaAlreadyExistsError, 

17 SchemaDoesNotExistError, 

18 SchemaTagAlreadyExistsError, 

19 SchemaTagDoesNotExistError, 

20 SchemaVersionAlreadyExistsError, 

21 SchemaVersionDoesNotExistError, 

22) 

23from pepdbagent.models import ( 

24 PaginationResult, 

25 SchemaRecordAnnotation, 

26 SchemaSearchResult, 

27 SchemaVersionAnnotation, 

28 SchemaVersionSearchResult, 

29 UpdateSchemaRecordFields, 

30 UpdateSchemaVersionFields, 

31) 

32 

33_LOGGER = logging.getLogger(PKG_NAME) 

34 

35 

36class PEPDatabaseSchema: 

37 """ 

38 Class that represents SchemaRecords in Database. 

39 

40 While using this class, user can create, retrieve, delete, and update schemas from database 

41 """ 

42 

43 def __init__(self, pep_db_engine: BaseEngine): 

44 """ 

45 Args: 

46 pep_db_engine: PEPDatabaseAgent engine object. 

47 """ 

48 self._sa_engine = pep_db_engine.engine 

49 self._pep_db_engine = pep_db_engine 

50 

51 def get(self, namespace: str, name: str, version: str) -> dict: 

52 """Get a schema value from the database. 

53 

54 Args: 

55 namespace: User namespace. 

56 name: Schema name. 

57 version: Schema version (use "latest" for the most recent). 

58 

59 Returns: 

60 Schema value dict. 

61 

62 Raises: 

63 SchemaVersionDoesNotExistError: If the schema or version does not exist. 

64 """ 

65 

66 with Session(self._sa_engine) as session: 

67 if version == LATEST_SCHEMA_VERSION: 

68 schema_obj = session.scalar( 

69 select(SchemaVersions) 

70 .join(SchemaRecords, SchemaRecords.id == SchemaVersions.schema_id) 

71 .where( 

72 and_( 

73 SchemaRecords.namespace == namespace, 

74 SchemaRecords.name == name, 

75 ) 

76 ) 

77 .order_by(SchemaVersions.version.desc()) 

78 ) 

79 

80 else: 

81 schema_obj = session.scalar( 

82 select(SchemaVersions) 

83 .join(SchemaRecords, SchemaRecords.id == SchemaVersions.schema_id) 

84 .where( 

85 and_( 

86 SchemaRecords.namespace == namespace, 

87 SchemaRecords.name == name, 

88 SchemaVersions.version == version, 

89 ) 

90 ) 

91 ) 

92 

93 if not schema_obj: 

94 raise SchemaVersionDoesNotExistError( 

95 f"Schema '{name}' does not exist in the database" 

96 ) 

97 

98 return schema_obj.schema_value 

99 

100 def create( 

101 self, 

102 namespace: str, 

103 name: str, 

104 schema_value: dict, 

105 version: str = DEFAULT_TAG_VERSION, 

106 description: str = "", 

107 lifecycle_stage: str = "", 

108 maintainers: str = "", 

109 contributors: str = "", 

110 release_notes: str = "", 

111 tags: list[str] | str | dict[str, str] | list[dict[str, str]] | None = None, 

112 private: bool = False, # TODO: for simplicity was not implemented yet 

113 ) -> None: 

114 """Create a new schema record in the database. 

115 

116 Args: 

117 namespace: User namespace. 

118 name: Schema name. 

119 schema_value: Schema content as a dict. 

120 version: Initial version string. 

121 description: Schema description. 

122 lifecycle_stage: Lifecycle stage, e.g., "stable" or "deprecated". 

123 maintainers: Comma-separated maintainer names. 

124 contributors: Comma-separated contributor names. 

125 release_notes: Release notes for this version. 

126 tags: Tags to associate with the schema version. 

127 private: Mark schema as private if True. 

128 

129 Raises: 

130 SchemaAlreadyExistsError: If a schema with the same name exists in the namespace. 

131 """ 

132 

133 tags = self._unify_tags(tags) 

134 

135 with Session(self._sa_engine) as session: 

136 schema_obj = session.scalar( 

137 select(SchemaRecords).where( 

138 and_( 

139 SchemaRecords.namespace == namespace, SchemaRecords.name == name 

140 ) 

141 ) 

142 ) 

143 

144 if schema_obj: 

145 raise SchemaAlreadyExistsError( 

146 f"Schema '{name}' already exists in the database" 

147 ) 

148 

149 user = session.scalar(select(User).where(User.namespace == namespace)) 

150 

151 if not user: 

152 user = User(namespace=namespace) 

153 session.add(user) 

154 session.commit() 

155 

156 user.number_of_schemas += 1 

157 

158 schema_obj = SchemaRecords( 

159 namespace=namespace, 

160 name=name, 

161 description=description, 

162 maintainers=maintainers, 

163 lifecycle_stage=lifecycle_stage, 

164 private=private, 

165 ) 

166 

167 session.add(schema_obj) 

168 

169 schema_version_obj = SchemaVersions( 

170 schema_mapping=schema_obj, 

171 version=version, 

172 schema_value=schema_value, 

173 release_notes=release_notes, 

174 contributors=contributors, 

175 ) 

176 

177 for tag_name, tag_value in tags.items(): 

178 tag_obj = session.scalar( 

179 select(SchemaTags).where(SchemaTags.tag_name == tag_name) 

180 ) 

181 if not tag_obj: 

182 tag_obj = SchemaTags( 

183 tag_name=tag_name, 

184 tag_value=tag_value, 

185 schema_mapping=schema_version_obj, 

186 ) 

187 session.add(tag_obj) 

188 

189 session.add(schema_version_obj) 

190 session.commit() 

191 

192 return None 

193 

194 def add_version( 

195 self, 

196 namespace: str, 

197 name: str, 

198 version: str, 

199 schema_value: dict, 

200 release_notes: str = "", 

201 contributors: str = "", 

202 overwrite: bool = False, 

203 tags: list[str] | str | dict[str, str] | list[dict[str, str]] | None = None, 

204 ) -> None: 

205 

206 tags = self._unify_tags(tags) 

207 

208 with Session(self._sa_engine) as session: 

209 schema_obj = session.scalar( 

210 select(SchemaRecords).where( 

211 and_( 

212 SchemaRecords.namespace == namespace, SchemaRecords.name == name 

213 ) 

214 ) 

215 ) 

216 if not schema_obj: 

217 raise SchemaDoesNotExistError( 

218 f"Schema '{name}' does not exist in the database. Unable to add version." 

219 ) 

220 

221 version_obj = session.scalar( 

222 select(SchemaVersions) 

223 .join(SchemaRecords, SchemaRecords.id == SchemaVersions.schema_id) 

224 .where( 

225 and_( 

226 SchemaRecords.namespace == namespace, 

227 SchemaRecords.name == name, 

228 SchemaVersions.version == version, 

229 ) 

230 ) 

231 ) 

232 

233 if version_obj: 

234 if not overwrite: 

235 raise SchemaVersionAlreadyExistsError( 

236 f"Schema '{name}' with version '{version}' already exists in the database" 

237 ) 

238 

239 return self.update_schema_version( 

240 namespace, 

241 name, 

242 version, 

243 update_fields=UpdateSchemaVersionFields( 

244 schema_value=schema_value, 

245 release_notes=release_notes, 

246 contributors=contributors, 

247 ), 

248 ) 

249 

250 schema_obj.last_update_date = func.now() 

251 

252 schema_version_obj = SchemaVersions( 

253 schema_id=schema_obj.id, 

254 version=version, 

255 schema_value=schema_value, 

256 release_notes=release_notes, 

257 contributors=contributors, 

258 ) 

259 

260 for tag_name, tag_value in tags.items(): 

261 tag_obj = SchemaTags( 

262 tag_name=tag_name, 

263 tag_value=tag_value, 

264 schema_mapping=schema_version_obj, 

265 ) 

266 session.add(tag_obj) 

267 

268 session.add(schema_version_obj) 

269 session.commit() 

270 

271 return None 

272 

273 def update_schema_version( 

274 self, 

275 namespace: str, 

276 name: str, 

277 version: str, 

278 update_fields: UpdateSchemaVersionFields | dict, 

279 ) -> None: 

280 """Update fields of an existing schema version. 

281 

282 Args: 

283 namespace: User namespace. 

284 name: Schema name. 

285 version: Schema version to update. 

286 update_fields: Fields to update — contributors, schema_value, and/or release_notes. 

287 

288 Raises: 

289 SchemaVersionDoesNotExistError: If the version does not exist. 

290 """ 

291 if isinstance(update_fields, dict): 

292 update_fields = UpdateSchemaVersionFields(**update_fields) 

293 update_fields = update_fields.model_dump( 

294 exclude_unset=True, exclude_defaults=True 

295 ) 

296 

297 with Session(self._sa_engine) as session: 

298 schema_obj = session.scalar( 

299 select(SchemaVersions) 

300 .join(SchemaRecords, SchemaRecords.id == SchemaVersions.schema_id) 

301 .where( 

302 and_( 

303 SchemaRecords.namespace == namespace, 

304 SchemaRecords.name == name, 

305 SchemaVersions.version == version, 

306 ) 

307 ) 

308 ) 

309 

310 if not schema_obj: 

311 raise SchemaVersionDoesNotExistError( 

312 f"Schema '{name}' with version '{version}' does not exist in the database. Unable to update version." 

313 ) 

314 schema_obj.last_update_date = func.now() 

315 

316 for field, value in update_fields.items(): 

317 setattr(schema_obj, field, value) 

318 if field == "schema_value": 

319 flag_modified(schema_obj, field) 

320 

321 session.commit() 

322 

323 def update_schema_record( 

324 self, 

325 namespace: str, 

326 name: str, 

327 update_fields: UpdateSchemaRecordFields | dict, 

328 ) -> None: 

329 """Update metadata fields of a schema record. 

330 

331 Args: 

332 namespace: User namespace. 

333 name: Schema name. 

334 update_fields: Fields to update — maintainers, lifecycle_stage, private, and/or name. 

335 

336 Raises: 

337 SchemaDoesNotExistError: If the schema does not exist. 

338 """ 

339 

340 if isinstance(update_fields, dict): 

341 update_fields = UpdateSchemaRecordFields(**update_fields) 

342 

343 update_fields = update_fields.model_dump( 

344 exclude_unset=True, exclude_defaults=True 

345 ) 

346 

347 with Session(self._sa_engine) as session: 

348 schema_obj = session.scalar( 

349 select(SchemaRecords).where( 

350 and_( 

351 SchemaRecords.namespace == namespace, SchemaRecords.name == name 

352 ) 

353 ) 

354 ) 

355 

356 if not schema_obj: 

357 raise SchemaDoesNotExistError( 

358 f"Schema '{name}' does not exist in the database" 

359 ) 

360 

361 for field, value in update_fields.items(): 

362 setattr(schema_obj, field, value) 

363 

364 session.commit() 

365 

366 def schema_exist(self, namespace: str, name: str) -> bool: 

367 """Check whether a schema exists in the database. 

368 

369 Args: 

370 namespace: User namespace. 

371 name: Schema name. 

372 

373 Returns: 

374 True if the schema exists. 

375 """ 

376 

377 with Session(self._sa_engine) as session: 

378 schema_obj = session.scalar( 

379 select(SchemaRecords).where( 

380 and_( 

381 SchemaRecords.namespace == namespace, SchemaRecords.name == name 

382 ) 

383 ) 

384 ) 

385 return True if schema_obj else False 

386 

387 def version_exist(self, namespace: str, name: str, version: str) -> bool: 

388 """Check whether a specific schema version exists. 

389 

390 Args: 

391 namespace: User namespace. 

392 name: Schema name. 

393 version: Schema version. 

394 

395 Returns: 

396 True if the version exists. 

397 """ 

398 

399 with Session(self._sa_engine) as session: 

400 schema_obj = session.scalar( 

401 select(SchemaVersions) 

402 .join(SchemaRecords, SchemaRecords.id == SchemaVersions.schema_id) 

403 .where( 

404 and_( 

405 SchemaRecords.namespace == namespace, 

406 SchemaRecords.name == name, 

407 SchemaVersions.version == version, 

408 ) 

409 ) 

410 ) 

411 return True if schema_obj else False 

412 

413 def get_schema_info(self, namespace: str, name: str) -> SchemaRecordAnnotation: 

414 """Get metadata for a schema record. 

415 

416 Args: 

417 namespace: User namespace. 

418 name: Schema name. 

419 

420 Returns: 

421 SchemaRecordAnnotation with schema metadata. 

422 

423 Raises: 

424 SchemaDoesNotExistError: If the schema does not exist. 

425 """ 

426 

427 with Session(self._sa_engine) as session: 

428 schema_obj = session.scalar( 

429 select(SchemaRecords).where( 

430 and_( 

431 SchemaRecords.namespace == namespace, SchemaRecords.name == name 

432 ) 

433 ) 

434 ) 

435 

436 if not schema_obj: 

437 raise SchemaDoesNotExistError( 

438 f"Schema '{name}' does not exist in the database" 

439 ) 

440 

441 return SchemaRecordAnnotation( 

442 namespace=schema_obj.namespace, 

443 schema_name=schema_obj.name, 

444 description=schema_obj.description, 

445 latest_released_version=schema_obj.versions_mapping[0].version, 

446 maintainers=schema_obj.maintainers, 

447 private=schema_obj.private, 

448 last_update_date=schema_obj.last_update_date, 

449 lifecycle_stage=schema_obj.lifecycle_stage, 

450 ) 

451 

452 def get_version_info( 

453 self, namespace: str, name: str, version: str 

454 ) -> SchemaVersionAnnotation: 

455 """Get metadata for a specific schema version. 

456 

457 Args: 

458 namespace: User namespace. 

459 name: Schema name. 

460 version: Schema version (use "latest" for the most recent). 

461 

462 Returns: 

463 SchemaVersionAnnotation with version metadata and tags. 

464 

465 Raises: 

466 SchemaVersionDoesNotExistError: If the version does not exist. 

467 """ 

468 

469 with Session(self._sa_engine) as session: 

470 # if user provided "latest" version 

471 if version == LATEST_SCHEMA_VERSION: 

472 version_obj = session.scalar( 

473 select(SchemaVersions) 

474 .join(SchemaRecords, SchemaRecords.id == SchemaVersions.schema_id) 

475 .where( 

476 and_( 

477 SchemaRecords.namespace == namespace, 

478 SchemaRecords.name == name, 

479 ) 

480 ) 

481 .order_by(SchemaVersions.version.desc()) 

482 .limit(1) 

483 ) 

484 else: 

485 version_obj = session.scalar( 

486 select(SchemaVersions) 

487 .join(SchemaRecords, SchemaRecords.id == SchemaVersions.schema_id) 

488 .where( 

489 and_( 

490 SchemaRecords.namespace == namespace, 

491 SchemaRecords.name == name, 

492 SchemaVersions.version == version, 

493 ) 

494 ) 

495 ) 

496 

497 if not version_obj: 

498 raise SchemaVersionDoesNotExistError( 

499 f"Schema '{name}' with version '{version}' does not exist in the database" 

500 ) 

501 

502 return SchemaVersionAnnotation( 

503 namespace=version_obj.schema_mapping.namespace, 

504 schema_name=version_obj.schema_mapping.name, 

505 version=version_obj.version, 

506 contributors=version_obj.contributors, 

507 release_notes=version_obj.release_notes, 

508 tags={tag.tag_name: tag.tag_value for tag in version_obj.tags_mapping}, 

509 release_date=version_obj.release_date, 

510 last_update_date=version_obj.last_update_date, 

511 ) 

512 

513 def fetch_schemas( 

514 self, 

515 namespace: str = None, 

516 name: str = None, 

517 maintainer: str = None, 

518 lifecycle_stage: str = None, 

519 latest_version: str = None, 

520 page: int = 0, 

521 page_size: int = 10, 

522 order_by: str = "update_date", 

523 order_desc: bool = False, 

524 ) -> SchemaSearchResult: 

525 """Get schemas matching the given filters. 

526 

527 Returns all schemas if no filters are provided. 

528 

529 Args: 

530 namespace: Restrict to this namespace (default: all namespaces). 

531 name: Partial name match. 

532 maintainer: Partial maintainer match. 

533 lifecycle_stage: Partial lifecycle stage match. 

534 latest_version: Partial latest version match. 

535 page: Page number (zero-based). 

536 page_size: Number of results per page. 

537 order_by: Sort field — "name" or "update_date". 

538 order_desc: Sort in descending order if True. 

539 

540 Returns: 

541 SchemaSearchResult with pagination and list of SchemaRecordAnnotation objects. 

542 """ 

543 

544 # filters = [ 

545 # SchemaRecords.namespace == namespace if namespace else None, 

546 # SchemaRecords.name == name if name else None, 

547 # SchemaRecords.maintainers == maintainer if maintainer else None, 

548 # SchemaRecords.lifecycle_stage == lifecycle_stage if lifecycle_stage else None, 

549 # ] 

550 filters = [ 

551 SchemaRecords.namespace.ilike(f"%{namespace}%") if namespace else None, 

552 SchemaRecords.name.ilike(f"%{name}%") if name else None, 

553 SchemaRecords.maintainers.ilike(f"%{maintainer}%") if maintainer else None, 

554 ( 

555 SchemaRecords.lifecycle_stage.ilike(f"%{lifecycle_stage}%") 

556 if lifecycle_stage 

557 else None 

558 ), 

559 ] 

560 

561 # Remove None values before applying and_ 

562 conditions = [f for f in filters if f is not None] 

563 

564 statement = ( 

565 select(SchemaRecords).where(and_(*conditions)) 

566 if conditions 

567 else select(SchemaRecords) 

568 ) 

569 statement_count = ( 

570 select(func.count(SchemaRecords.id)).where(and_(*conditions)) 

571 if conditions 

572 else select(func.count(SchemaRecords.id)) 

573 ) 

574 

575 with Session(self._sa_engine) as session: 

576 total = session.scalar(statement_count) 

577 

578 statement = self._add_order_by_schemas_keyword( 

579 statement, by=order_by, desc=order_desc 

580 ) 

581 

582 results_objects = session.scalars( 

583 statement.limit(page_size).offset(page * page_size) 

584 ) 

585 return SchemaSearchResult( 

586 pagination=PaginationResult( 

587 page=page, 

588 page_size=page_size, 

589 total=total, 

590 ), 

591 results=[ 

592 SchemaRecordAnnotation( 

593 namespace=result.namespace, 

594 schema_name=result.name, 

595 latest_released_version=result.versions_mapping[0].version, 

596 description=result.description, 

597 maintainers=result.maintainers, 

598 private=result.private, 

599 last_update_date=result.last_update_date, 

600 ) 

601 for result in results_objects 

602 ], 

603 ) 

604 

605 def query_schemas( 

606 self, 

607 namespace: str = None, 

608 search_str: str = "", 

609 page: int = 0, 

610 page_size: int = 10, 

611 order_by: str = "update_date", 

612 order_desc: bool = False, 

613 ) -> SchemaSearchResult: 

614 """Search schemas by name and description with pagination. 

615 

616 Args: 

617 namespace: Restrict to this namespace (default: all namespaces). 

618 search_str: Text to search in name and description (default: all schemas). 

619 page: Page number (zero-based). 

620 page_size: Number of results per page. 

621 order_by: Sort field — "name" or "update_date". 

622 order_desc: Sort in descending order if True. 

623 

624 Returns: 

625 SchemaSearchResult with pagination and list of SchemaRecordAnnotation objects. 

626 """ 

627 

628 search_str = search_str.lower() if search_str else "" 

629 

630 where_statement = or_( 

631 SchemaRecords.name.ilike(f"%{search_str}%"), 

632 SchemaRecords.description.ilike(f"%{search_str}%"), 

633 ) 

634 if namespace: 

635 where_statement = and_( 

636 where_statement, SchemaRecords.namespace == namespace 

637 ) 

638 

639 with Session(self._sa_engine) as session: 

640 total = session.scalar( 

641 select(func.count(SchemaRecords.id)).where(where_statement) 

642 ) 

643 statement = ( 

644 select(SchemaRecords) 

645 .where(where_statement) 

646 .limit(page_size) 

647 .offset(page * page_size) 

648 ) 

649 statement = self._add_order_by_schemas_keyword( 

650 statement, by=order_by, desc=order_desc 

651 ) 

652 results_objects = session.scalars(statement) 

653 

654 return SchemaSearchResult( 

655 pagination=PaginationResult( 

656 page=page, 

657 page_size=page_size, 

658 total=total, 

659 ), 

660 results=[ 

661 SchemaRecordAnnotation( 

662 namespace=result.namespace, 

663 schema_name=result.name, 

664 latest_released_version=result.versions_mapping[0].version, 

665 description=result.description, 

666 maintainers=result.maintainers, 

667 private=result.private, 

668 last_update_date=result.last_update_date, 

669 ) 

670 for result in results_objects 

671 ], 

672 ) 

673 

674 def query_schema_version( 

675 self, 

676 namespace: str, 

677 name: str, 

678 tag: str = None, 

679 search_str: str = "", 

680 page: int = 0, 

681 page_size: int = 10, 

682 ) -> SchemaVersionSearchResult: 

683 """Search versions of a schema with pagination. 

684 

685 Args: 

686 namespace: User namespace. 

687 name: Schema name. 

688 tag: Filter by tag name (default: all tags). 

689 search_str: Text to search in version and release_notes. 

690 page: Page number (zero-based). 

691 page_size: Number of results per page. 

692 

693 Returns: 

694 SchemaVersionSearchResult with pagination and list of SchemaVersionAnnotation objects. 

695 

696 Raises: 

697 SchemaDoesNotExistError: If the schema does not exist. 

698 """ 

699 

700 search_str = search_str.lower() if search_str else "" 

701 

702 with Session(self._sa_engine) as session: 

703 schema_obj = session.scalar( 

704 select(SchemaRecords).where( 

705 and_( 

706 SchemaRecords.namespace == namespace, SchemaRecords.name == name 

707 ) 

708 ) 

709 ) 

710 

711 if not schema_obj: 

712 raise SchemaDoesNotExistError( 

713 f"Schema '{name}' does not exist in the database" 

714 ) 

715 

716 where_statement = and_( 

717 SchemaRecords.namespace == namespace, 

718 SchemaRecords.name == name, 

719 or_( 

720 SchemaVersions.version.ilike(f"%{search_str}%"), 

721 SchemaVersions.release_notes.ilike(f"%{search_str}%"), 

722 ), 

723 ) 

724 

725 if tag: 

726 where_statement = and_(where_statement, SchemaTags.tag_name == tag) 

727 total_statement = ( 

728 select(func.count(SchemaVersions.id)) 

729 .join(SchemaRecords) 

730 .join(SchemaTags) 

731 .where(where_statement) 

732 ) 

733 find_statement = ( 

734 select(SchemaVersions) 

735 .join(SchemaRecords) 

736 .join(SchemaTags) 

737 .where(where_statement) 

738 ) 

739 

740 else: 

741 total_statement = ( 

742 select(func.count(SchemaVersions.id)) 

743 .join(SchemaRecords) 

744 .where(where_statement) 

745 ) 

746 find_statement = ( 

747 select(SchemaVersions).join(SchemaRecords).where(where_statement) 

748 ) 

749 

750 total = session.scalar(total_statement) 

751 

752 results_objects = session.scalars( 

753 find_statement.order_by(SchemaVersions.version.desc()) 

754 .limit(page_size) 

755 .offset(page * page_size) 

756 ).unique() 

757 

758 return SchemaVersionSearchResult( 

759 pagination=PaginationResult( 

760 page=page, 

761 page_size=page_size, 

762 total=total, 

763 ), 

764 results=[ 

765 SchemaVersionAnnotation( 

766 namespace=result.schema_mapping.namespace, 

767 schema_name=result.schema_mapping.name, 

768 version=result.version, 

769 contributors=result.contributors, 

770 release_notes=result.release_notes, 

771 tags={ 

772 tag.tag_name: tag.tag_value for tag in result.tags_mapping 

773 }, 

774 release_date=result.release_date, 

775 last_update_date=result.last_update_date, 

776 ) 

777 for result in results_objects 

778 ], 

779 ) 

780 

781 def delete_schema(self, namespace: str, name: str) -> None: 

782 """Delete a schema record and all its versions from the database. 

783 

784 Args: 

785 namespace: User namespace. 

786 name: Schema name. 

787 

788 Raises: 

789 SchemaDoesNotExistError: If the schema does not exist. 

790 """ 

791 

792 with Session(self._sa_engine) as session: 

793 schema_obj = session.scalar( 

794 select(SchemaRecords).where( 

795 and_( 

796 SchemaRecords.namespace == namespace, SchemaRecords.name == name 

797 ) 

798 ) 

799 ) 

800 

801 if not schema_obj: 

802 raise SchemaDoesNotExistError( 

803 f"Schema '{name}' does not exist in the database" 

804 ) 

805 

806 statement = select(User).where(User.namespace == namespace) 

807 user = session.scalar(statement) 

808 if user: 

809 user.number_of_schemas -= 1 

810 session.commit() 

811 

812 session.delete(schema_obj) 

813 session.commit() 

814 

815 def delete_version(self, namespace: str, name: str, version: str) -> None: 

816 """Delete a specific schema version. 

817 

818 Args: 

819 namespace: Namespace of the schema. 

820 name: Name of the schema. 

821 version: Version to delete. 

822 

823 Raises: 

824 SchemaVersionDoesNotExistError: If the version does not exist. 

825 """ 

826 with Session(self._sa_engine) as session: 

827 schema_obj = session.scalar( 

828 select(SchemaVersions) 

829 .join(SchemaRecords, SchemaRecords.id == SchemaVersions.schema_id) 

830 .where( 

831 and_( 

832 SchemaRecords.namespace == namespace, 

833 SchemaRecords.name == name, 

834 SchemaVersions.version == version, 

835 ) 

836 ) 

837 ) 

838 if not schema_obj: 

839 raise SchemaVersionDoesNotExistError( 

840 f"Schema '{name}' with version '{version}' does not exist in the database. Unable to update version." 

841 ) 

842 

843 session.delete(schema_obj) 

844 session.commit() 

845 

846 def add_tag_to_schema( 

847 self, 

848 namespace: str, 

849 name: str, 

850 version: str, 

851 tag: list[str] | str | dict[str, str] | None, 

852 ) -> None: 

853 """Add a tag to a schema version. 

854 

855 Args: 

856 namespace: Namespace of the schema. 

857 name: Name of the schema. 

858 version: Schema version. 

859 tag: Tag to add — string, list of strings, or dict of name→value pairs. 

860 

861 Raises: 

862 SchemaVersionDoesNotExistError: If the version does not exist. 

863 SchemaTagAlreadyExistsError: If the tag already exists on the version. 

864 """ 

865 

866 tag = self._unify_tags(tag) 

867 

868 with Session(self._sa_engine) as session: 

869 schema_obj = session.scalar( 

870 select(SchemaVersions) 

871 .join(SchemaRecords, SchemaRecords.id == SchemaVersions.schema_id) 

872 .where( 

873 and_( 

874 SchemaRecords.namespace == namespace, 

875 SchemaRecords.name == name, 

876 SchemaVersions.version == version, 

877 ) 

878 ) 

879 ) 

880 if not schema_obj: 

881 raise SchemaVersionDoesNotExistError( 

882 f"Schema '{name}' with version '{version}' does not exist in the database. Unable to add tag." 

883 ) 

884 if isinstance(tag, str): 

885 tag = [tag] 

886 

887 for tag_name, tag_value in tag.items(): 

888 tag_obj = session.scalar( 

889 select(SchemaTags).where(SchemaTags.tag_name == tag_name) 

890 ) 

891 if not tag_obj: 

892 tag_obj = SchemaTags( 

893 tag_name=tag_name, 

894 tag_value=tag_value, 

895 schema_mapping=schema_obj, 

896 ) 

897 session.add(tag_obj) 

898 else: 

899 raise SchemaTagAlreadyExistsError( 

900 f"Tag '{tag_name}' already exists in the schema" 

901 ) 

902 

903 session.commit() 

904 

905 def remove_tag_from_schema( 

906 self, namespace: str, name: str, version: str, tag: str 

907 ) -> None: 

908 """Remove a tag from a schema version. 

909 

910 Args: 

911 namespace: Namespace of the schema. 

912 name: Name of the schema. 

913 version: Schema version. 

914 tag: Name of the tag to remove. 

915 

916 Raises: 

917 SchemaVersionDoesNotExistError: If the version does not exist. 

918 SchemaTagDoesNotExistError: If the tag does not exist on the version. 

919 """ 

920 with Session(self._sa_engine) as session: 

921 schema_obj = session.scalar( 

922 select(SchemaVersions) 

923 .join(SchemaRecords, SchemaRecords.id == SchemaVersions.schema_id) 

924 .where( 

925 and_( 

926 SchemaRecords.namespace == namespace, 

927 SchemaRecords.name == name, 

928 SchemaVersions.version == version, 

929 ) 

930 ) 

931 ) 

932 if not schema_obj: 

933 raise SchemaVersionDoesNotExistError(