Coverage for pepdbagent/modules/project.py: 90%

1import datetime

2import logging

4import numpy as np

5import peprs

6from peprs.const import (

7 CONFIG_KEY,

8 SAMPLE_RAW_DICT_KEY,

9 SUBSAMPLE_RAW_DICT_KEY,

10)

11from sqlalchemy import Select, and_, delete, select

12from sqlalchemy.exc import IntegrityError, NoResultFound

13from sqlalchemy.orm import Session

14from sqlalchemy.orm.attributes import flag_modified

16from pepdbagent.const import (

17 DEFAULT_TAG,

18 DESCRIPTION_KEY,

19 LATEST_SCHEMA_VERSION,

20 MAX_HISTORY_SAMPLES_NUMBER,

21 NAME_KEY,

22 PEPHUB_SAMPLE_ID_KEY,

23 PKG_NAME,

24 SAMPLE_NAME_ATTR,

25 SAMPLE_TABLE_INDEX_KEY,

26)

27from pepdbagent.db_utils import (

28 BaseEngine,

29 HistoryProjects,

30 HistorySamples,

31 Projects,

32 Samples,

33 SchemaRecords,

34 SchemaVersions,

35 Subsamples,

36 UpdateTypes,

37 User,

38)

39from pepdbagent.exceptions import (

40 HistoryNotFoundError,

41 PEPDatabaseAgentError,

42 ProjectDuplicatedSampleGUIDsError,

43 ProjectNotFoundError,

44 ProjectUniqueNameError,

45 SampleTableUpdateError,

46 SchemaDoesNotExistError,

47)

48from pepdbagent.models import (

49 HistoryAnnotationModel,

50 HistoryChangeModel,

51 ProjectDict,

52 UpdateItems,

53 UpdateModel,

54)

55from pepdbagent.utils import (

56 create_digest,

57 generate_guid,

58 order_samples,

59 registry_path_converter,

60 schema_path_converter,

61)

63_LOGGER = logging.getLogger(PKG_NAME)

66class PEPDatabaseProject:

67 """

68 Class that represents Project in Database.

70 While using this class, user can create, retrieve, delete, and update projects from database

71 """

73 def __init__(self, pep_db_engine: BaseEngine):

74 """

75 Args:

76 pep_db_engine: PEPDatabaseAgent engine object.

77 """

78 self._sa_engine = pep_db_engine.engine

79 self._pep_db_engine = pep_db_engine

81 def get(

82 self,

83 namespace: str,

84 name: str,

85 tag: str = DEFAULT_TAG,

86 raw: bool = True,

87 with_id: bool = False,

88 ) -> peprs.Project | dict | None:

89 """Retrieve a project from the database.

91 Args:

92 namespace: Namespace of the project.

93 name: Name of the project.

94 tag: Tag of the project.

95 raw: Return raw dict if True, peprs.Project object if False.

96 with_id: Include pephub_sample_id in each sample dict if True.

98 Returns:

99 Raw dict or peprs.Project object.

100

101 Raises:

102 ProjectNotFoundError: If the project does not exist.

103 """

104 # name = name.lower()

105 namespace = namespace.lower()

106 statement = self._create_select_statement(name, namespace, tag)

107

108 try:

109 with Session(self._sa_engine) as session:

110 found_prj = session.scalar(statement)

111

112 if found_prj:

113 _LOGGER.info(

114 f"Project has been found: {found_prj.namespace}, {found_prj.name}"

115 )

116 subsample_dict = {}

117 if found_prj.subsamples_mapping:

118 for subsample in found_prj.subsamples_mapping:

119 if subsample.subsample_number not in subsample_dict.keys():

120 subsample_dict[subsample.subsample_number] = []

121 subsample_dict[subsample.subsample_number].append(

122 subsample.subsample

123 )

124 subsample_list = list(subsample_dict.values())

125 else:

126 subsample_list = []

127

128 sample_list = self._get_samples(

129 session=session, prj_id=found_prj.id, with_id=with_id

130 )

131

132 project_value = {

133 CONFIG_KEY: found_prj.config,

134 SAMPLE_RAW_DICT_KEY: sample_list,

135 SUBSAMPLE_RAW_DICT_KEY: subsample_list,

136 }

137

138 if raw:

139 return project_value

140 else:

141 project_obj = peprs.Project.from_dict(project_value)

142 return project_obj

143

144 else:

145 raise ProjectNotFoundError(

146 f"No project found for supplied input: '{namespace}/{name}:{tag}'. "

147 f"Did you supply a valid namespace and project?"

148 )

149

150 except NoResultFound:

151 raise ProjectNotFoundError

152

153 def _get_samples(self, session: Session, prj_id: int, with_id: bool) -> list[dict]:

154 """Get ordered samples from a project using an open session.

155

156 Args:

157 session: Open SQLAlchemy session.

158 prj_id: Project id.

159 with_id: Include pephub_sample_id in each sample dict if True.

160

161 Returns:

162 Ordered list of sample dicts.

163 """

164 result_dict = self._get_samples_dict(prj_id, session, with_id)

165

166 result_dict = order_samples(result_dict)

167

168 ordered_samples_list = [sample["sample"] for sample in result_dict]

169 return ordered_samples_list

170

171 @staticmethod

172 def _get_samples_dict(prj_id: int, session: Session, with_id: bool) -> dict:

173 """Get unordered samples from a project keyed by guid.

174

175 Args:

176 prj_id: Project id.

177 session: Open SQLAlchemy session.

178 with_id: Include pephub_sample_id in each sample dict if True.

179

180 Returns:

181 Dict mapping guid → {sample, guid, parent_guid}.

182 """

183 samples_results = session.scalars(

184 select(Samples).where(Samples.project_id == prj_id)

185 )

186 result_dict = {}

187 for sample in samples_results:

188 sample_dict = sample.sample

189 if with_id:

190 sample_dict[PEPHUB_SAMPLE_ID_KEY] = sample.guid

191

192 result_dict[sample.guid] = {

193 "sample": sample_dict,

194 "guid": sample.guid,

195 "parent_guid": sample.parent_guid,

196 }

197

198 return result_dict

199

200 @staticmethod

201 def _create_select_statement(

202 name: str, namespace: str, tag: str = DEFAULT_TAG

203 ) -> Select:

204 """Create a SELECT statement for a single project.

205

206 Args:

207 name: Project name.

208 namespace: Project namespace.

209 tag: Project tag.

210

211 Returns:

212 SQLAlchemy SELECT statement for the Projects table.

213 """

214 statement = select(Projects)

215 statement = statement.where(

216 and_(

217 Projects.namespace == namespace,

218 Projects.name == name,

219 Projects.tag == tag,

220 )

221 )

222 return statement

223

224 def get_by_rp(

225 self,

226 registry_path: str,

227 raw: bool = False,

228 ) -> peprs.Project | dict | None:

229 """Retrieve a project from the database by registry path.

230

231 Args:

232 registry_path: Registry path, e.g., "namespace/name:tag".

233 raw: Return raw dict if True, peprs.Project object if False.

234

235 Returns:

236 Raw dict or peprs.Project object.

237

238 Raises:

239 ProjectNotFoundError: If the project does not exist.

240 """

241 namespace, name, tag = registry_path_converter(registry_path)

242 return self.get(namespace=namespace, name=name, tag=tag, raw=raw)

243

244 def delete(

245 self,

246 namespace: str = None,

247 name: str = None,

248 tag: str = None,

249 ) -> None:

250 """Delete a project from the database.

251

252 Args:

253 namespace: Project namespace.

254 name: Project name.

255 tag: Project tag.

256

257 Raises:

258 ProjectNotFoundError: If the project does not exist.

259 """

260 # name = name.lower()

261 namespace = namespace.lower()

262

263 if not self.exists(namespace=namespace, name=name, tag=tag):

264 raise ProjectNotFoundError(

265 f"Can't delete unexciting project: '{namespace}/{name}:{tag}'."

266 )

267

268 with Session(self._sa_engine) as session:

269 session.execute(

270 delete(Projects).where(

271 and_(

272 Projects.namespace == namespace,

273 Projects.name == name,

274 Projects.tag == tag,

275 )

276 )

277 )

278

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

280 user = session.scalar(statement)

281 if user:

282 user.number_of_projects -= 1

283 session.commit()

284

285 def delete_by_rp(

286 self,

287 registry_path: str,

288 ) -> None:

289 """Delete a project from the database by registry path.

290

291 Args:

292 registry_path: Registry path of the project, e.g., "namespace/name:tag".

293

294 Raises:

295 ProjectNotFoundError: If the project does not exist.

296 """

297 namespace, name, tag = registry_path_converter(registry_path)

298 return self.delete(namespace=namespace, name=name, tag=tag)

299

300 def create(

301 self,

302 project: peprs.Project | dict,

303 namespace: str,

304 name: str = None,

305 tag: str = DEFAULT_TAG,

306 description: str = None,

307 is_private: bool = False,

308 pop: bool = False,

309 pep_schema: str = None,

310 overwrite: bool = False,

311 update_only: bool = False,

312 ) -> None:

313 """Upload a project to the database.

314

315 Args:

316 project: peprs.Project or dict with config, samples, and subsamples keys.

317 namespace: Namespace to upload to.

318 name: Project name (default: taken from project config).

319 tag: Project tag.

320 description: Project description.

321 is_private: Mark project as private if True.

322 pop: Mark as a PEP-of-PEPs (POP) if True.

323 pep_schema: Schema to associate, e.g., "namespace/name".

324 overwrite: Overwrite the project if it already exists.

325 update_only: Update the project if it exists; do nothing otherwise.

326

327 Raises:

328 ProjectUniqueNameError: If the project already exists and overwrite is False.

329 SchemaDoesNotExistError: If pep_schema is provided but does not exist.

330 """

331 if isinstance(project, peprs.Project):

332 proj_dict = project.to_dict(raw=True, by_sample=True)

333 elif isinstance(project, dict):

334 # verify if the dictionary has all necessary elements.

335 # samples should be always presented as list of dicts (orient="records"))

336 _LOGGER.warning(

337 f"Project f{namespace}/{name}:{tag} is provided as dictionary. Project won't be validated."

338 )

339 proj_dict = ProjectDict(**project).model_dump(by_alias=True)

340 else:

341 raise PEPDatabaseAgentError(

342 "Project has to be peprs.Project object or dictionary with PEP elements"

343 )

344

345 if not description:

346 if isinstance(project, peprs.Project):

347 description = project.description or ""

348 else:

349 description = proj_dict.get(CONFIG_KEY, {}).get(DESCRIPTION_KEY, "")

350 proj_dict[CONFIG_KEY][DESCRIPTION_KEY] = description

351

352 namespace = namespace.lower()

353 if name:

354 proj_name = name.lower()

355 elif proj_dict[CONFIG_KEY][NAME_KEY]:

356 proj_name = proj_dict[CONFIG_KEY][NAME_KEY].lower()

357 else:

358 raise ValueError(

359 "Name of the project wasn't provided. Project will not be uploaded."

360 )

361

362 proj_dict[CONFIG_KEY][NAME_KEY] = proj_name

363

364 proj_digest = create_digest(proj_dict)

365 try:

366 number_of_samples = len(project.samples)

367 except AttributeError:

368 number_of_samples = len(proj_dict[SAMPLE_RAW_DICT_KEY])

369

370 if pep_schema:

371 schema_namespace, schema_name, schema_version = schema_path_converter(

372 pep_schema

373 )

374 with Session(self._sa_engine) as session:

375 if schema_version == LATEST_SCHEMA_VERSION:

376 schema_mapping = session.scalar(

377 select(SchemaVersions)

378 .join(

379 SchemaRecords, SchemaRecords.id == SchemaVersions.schema_id

380 )

381 .where(

382 and_(

383 SchemaRecords.namespace == schema_namespace,

384 SchemaRecords.name == schema_name,

385 )

386 )

387 .order_by(SchemaVersions.version.desc())

388 )

389

390 else:

391 where_clause = and_(

392 SchemaRecords.namespace == schema_namespace,

393 SchemaRecords.name == schema_name,

394 SchemaVersions.version == schema_version,

395 )

396

397 schema_mapping = session.scalar(

398 select(SchemaVersions).join(SchemaRecords).where(where_clause)

399 )

400 if not schema_mapping:

401 raise SchemaDoesNotExistError(

402 f"Schema {schema_namespace}/{schema_name} does not exist. "

403 f"Project won't be uploaded."

404 )

405 pep_schema = schema_mapping.id

406

407 if update_only:

408 _LOGGER.info(

409 f"Update_only argument is set True. Updating project {proj_name} ..."

410 )

411 self._overwrite(

412 project_dict=proj_dict,

413 namespace=namespace,

414 proj_name=proj_name,

415 tag=tag,

416 project_digest=proj_digest,

417 number_of_samples=number_of_samples,

418 private=is_private,

419 pep_schema=pep_schema,

420 description=description,

421 pop=pop,

422 )

423 return None

424 else:

425 try:

426 _LOGGER.info(f"Uploading {namespace}/{proj_name}:{tag} project...")

427 new_prj = Projects(

428 namespace=namespace,

429 name=proj_name,

430 tag=tag,

431 digest=proj_digest,

432 config=proj_dict[CONFIG_KEY],

433 number_of_samples=number_of_samples,

434 private=is_private,

435 submission_date=datetime.datetime.now(datetime.timezone.utc),

436 last_update_date=datetime.datetime.now(datetime.timezone.utc),

437 # pep_schema=pep_schema,

438 schema_id=pep_schema,

439 description=description,

440 pop=pop,

441 )

442

443 self._add_samples_to_project(

444 new_prj,

445 proj_dict[SAMPLE_RAW_DICT_KEY],

446 sample_table_index=proj_dict[CONFIG_KEY].get(

447 SAMPLE_TABLE_INDEX_KEY, SAMPLE_NAME_ATTR

448 ),

449 )

450

451 if proj_dict.get(SUBSAMPLE_RAW_DICT_KEY):

452 subsamples = proj_dict[SUBSAMPLE_RAW_DICT_KEY]

453 self._add_subsamples_to_project(new_prj, subsamples)

454

455 with Session(self._sa_engine) as session:

456 user = session.scalar(

457 select(User).where(User.namespace == namespace)

458 )

459

460 if not user:

461 user = User(namespace=namespace)

462 session.add(user)

463 session.commit()

464

465 user.number_of_projects += 1

466

467 session.add(new_prj)

468 session.commit()

469

470 return None

471

472 except IntegrityError:

473 if overwrite:

474 self._overwrite(

475 project_dict=proj_dict,

476 namespace=namespace,

477 proj_name=proj_name,

478 tag=tag,

479 project_digest=proj_digest,

480 number_of_samples=number_of_samples,

481 private=is_private,

482 pep_schema=pep_schema,

483 description=description,

484 )

485 return None

486

487 else:

488 raise ProjectUniqueNameError(

489 "Namespace, name and tag already exists. Project won't be "

490 "uploaded. Solution: Set overwrite value as True"

491 " (project will be overwritten), or change tag!"

492 )

493

494 def _overwrite(

495 self,

496 project_dict: dict,

497 namespace: str,

498 proj_name: str,

499 tag: str,

500 project_digest: str,

501 number_of_samples: int,

502 private: bool = False,

503 pep_schema: int = None,

504 description: str = "",

505 pop: bool = False,

506 ) -> None:

507 """Overwrite an existing project with new content.

508

509 Args:

510 project_dict: Full project dict (config, samples, subsamples).

511 namespace: Project namespace.

512 proj_name: Project name.

513 tag: Project tag.

514 project_digest: Pre-computed digest.

515 number_of_samples: Sample count.

516 private: Mark as private if True.

517 pep_schema: Schema id to associate (integer FK).

518 description: Project description.

519 pop: Mark as POP if True.

520

521 Raises:

522 ProjectNotFoundError: If the project does not exist.

523 """

524 proj_name = proj_name.lower()

525 namespace = namespace.lower()

526 if self.exists(namespace=namespace, name=proj_name, tag=tag):

527 _LOGGER.info(f"Updating {proj_name} project...")

528 statement = self._create_select_statement(proj_name, namespace, tag)

529

530 with Session(self._sa_engine) as session:

531 found_prj = session.scalar(statement)

532

533 if found_prj:

534 _LOGGER.debug(

535 f"Project has been found: {found_prj.namespace}, {found_prj.name}"

536 )

537

538 found_prj.digest = project_digest

539 found_prj.number_of_samples = number_of_samples

540 found_prj.private = private

541 # found_prj.pep_schema = pep_schema

542 found_prj.schema_id = pep_schema

543 found_prj.config = project_dict[CONFIG_KEY]

544 found_prj.description = description

545 found_prj.last_update_date = datetime.datetime.now(

546 datetime.timezone.utc

547 )

548 found_prj.pop = pop

549

550 # Deleting old samples and subsamples

551 if found_prj.samples_mapping:

552 for sample in found_prj.samples_mapping:

553 _LOGGER.debug(f"deleting samples: {str(sample)}")

554 session.delete(sample)

555

556 if found_prj.subsamples_mapping:

557 for subsample in found_prj.subsamples_mapping:

558 _LOGGER.debug(f"deleting subsamples: {str(subsample)}")

559 session.delete(subsample)

560

561 # Adding new samples and subsamples

562 self._add_samples_to_project(

563 found_prj,

564 project_dict[SAMPLE_RAW_DICT_KEY],

565 sample_table_index=project_dict[CONFIG_KEY].get(

566 SAMPLE_TABLE_INDEX_KEY

567 ),

568 )

569

570 if project_dict.get(SUBSAMPLE_RAW_DICT_KEY):

571 self._add_subsamples_to_project(

572 found_prj, project_dict[SUBSAMPLE_RAW_DICT_KEY]

573 )

574

575 session.commit()

576

577 _LOGGER.info(

578 f"Project '{namespace}/{proj_name}:{tag}' has been successfully updated!"

579 )

580 return None

581

582 else:

583 raise ProjectNotFoundError(

584 "Project does not exist! No project will be updated!"

585 )

586

587 def update(

588 self,

589 update_dict: dict | UpdateItems,

590 namespace: str,

591 name: str,

592 tag: str = DEFAULT_TAG,

593 user: str = None,

594 ) -> None:

595 """Update specific fields of a project.

596

597 Args:

598 update_dict: Fields to update — any combination of project, is_private, tag, name,

599 description, pep_schema, config, samples, subsamples, pop.

600 namespace: Project namespace.

601 name: Project name.

602 tag: Project tag.

603 user: User performing the update (default: namespace).

604

605 Raises:

606 ProjectNotFoundError: If the project does not exist.

607 """

608 if self.exists(namespace=namespace, name=name, tag=tag):

609 if isinstance(update_dict, UpdateItems):

610 update_values = update_dict

611 else:

612 if "project" in update_dict:

613 project_dict = update_dict.pop("project").to_dict(

614 raw=True, by_sample=True

615 )

616 update_dict["config"] = project_dict[CONFIG_KEY]

617 update_dict["samples"] = project_dict[SAMPLE_RAW_DICT_KEY]

618 update_dict["subsamples"] = project_dict.get(

619 SUBSAMPLE_RAW_DICT_KEY, []

620 )

621

622 update_values = UpdateItems(**update_dict)

623

624 update_values = self.__create_update_dict(update_values)

625

626 statement = self._create_select_statement(name, namespace, tag)

627

628 with Session(self._sa_engine) as session:

629 found_prj: Projects = session.scalar(statement)

630

631 if not found_prj:

632 raise ProjectNotFoundError(

633 f"Pep {namespace}/{name}:{tag} was not found. No items will be updated!"

634 )

635

636 self._convert_update_schema_id(session, update_values)

637

638 for k, v in update_values.items():

639 if getattr(found_prj, k) != v:

640 setattr(found_prj, k, v)

641

642 # standardizing project name

643 if k == NAME_KEY:

644 if "config" in update_values:

645 update_values["config"][NAME_KEY] = v

646 else:

647 found_prj.config[NAME_KEY] = v

648 flag_modified(found_prj, "config")

649 found_prj.name = found_prj.config[NAME_KEY]

650

651 if k == DESCRIPTION_KEY:

652 if "config" in update_values:

653 update_values["config"][DESCRIPTION_KEY] = v

654 else:

655 found_prj.config[DESCRIPTION_KEY] = v

656 # This line needed due to: https://github.com/sqlalchemy/sqlalchemy/issues/5218

657 flag_modified(found_prj, "config")

658

659 if "samples" in update_dict:

660 if PEPHUB_SAMPLE_ID_KEY not in update_dict["samples"][0]:

661 raise SampleTableUpdateError(

662 f"pephub_sample_id '{PEPHUB_SAMPLE_ID_KEY}' is missing in samples."

663 f"Please provide it to update samples, or use overwrite method."

664 )

665 if len(update_dict["samples"]) > MAX_HISTORY_SAMPLES_NUMBER:

666 _LOGGER.warning(

667 f"Number of samples in the project exceeds the limit of {MAX_HISTORY_SAMPLES_NUMBER}."

668 f"Samples won't be updated."

669 )

670 new_history = None

671 else:

672 new_history = HistoryProjects(

673 project_id=found_prj.id,

674 user=user or namespace,

675 project_yaml=self.get_config(namespace, name, tag),

676 )

677 session.add(new_history)

678

679 self._update_samples(

680 project_id=found_prj.id,

681 samples_list=update_dict["samples"],

682 sample_name_key=update_dict["config"].get(

683 SAMPLE_TABLE_INDEX_KEY, "sample_name"

684 ),

685 history_sa_model=new_history,

686 )

687 found_prj.number_of_samples = len(update_dict["samples"])

688

689 if "subsamples" in update_dict:

690 if found_prj.subsamples_mapping:

691 for subsample in found_prj.subsamples_mapping:

692 _LOGGER.debug(f"deleting subsamples: {str(subsample)}")

693 session.delete(subsample)

694

695 # Adding new subsamples

696 if update_dict["subsamples"]:

697 self._add_subsamples_to_project(

698 found_prj, update_dict["subsamples"]

699 )

700

701 found_prj.last_update_date = datetime.datetime.now(

702 datetime.timezone.utc

703 )

704

705 session.commit()

706

707 return None

708

709 else:

710 raise ProjectNotFoundError("No items will be updated!")

711

712 @staticmethod

713 def _convert_update_schema_id(session: Session, update_values: dict) -> None:

714 """Resolve pep_schema path to a schema_id in the update dict.

715

716 Args:

717 session: Open SQLAlchemy session.

718 update_values: Update dict that may contain a "pep_schema" key.

719

720 Raises:

721 SchemaDoesNotExistError: If the schema path does not resolve to an existing schema.

722 """

723 if "pep_schema" in update_values:

724 schema_namespace, schema_name, schema_version = schema_path_converter(

725 update_values["pep_schema"]

726 )

727 if schema_version == LATEST_SCHEMA_VERSION:

728 schema_mapping = session.scalar(

729 select(SchemaVersions)

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

731 .where(

732 and_(

733 SchemaRecords.namespace == schema_namespace,

734 SchemaRecords.name == schema_name,

735 )

736 )

737 .order_by(SchemaVersions.version.desc())

738 )

739

740 else:

741 where_clause = and_(

742 SchemaRecords.namespace == schema_namespace,

743 SchemaRecords.name == schema_name,

744 SchemaVersions.version == schema_version,

745 )

746

747 schema_mapping = session.scalar(

748 select(SchemaVersions).join(SchemaRecords).where(where_clause)

749 )

750 if not schema_mapping:

751 raise SchemaDoesNotExistError(

752 f"Schema {schema_namespace}/{schema_name} does not exist. "

753 f"Project won't be updated."

754 )

755 update_values["schema_id"] = schema_mapping.id

756 del update_values["pep_schema"]

757

758 def _update_samples(

759 self,

760 project_id: int,

761 samples_list: list[dict[str, str]],

762 sample_name_key: str = "sample_name",

763 history_sa_model: HistoryProjects | None = None,

764 ) -> None:

765 """Update samples in a project using a linked-list diff approach.

766

767 Finds differences between old and new sample lists, then applies inserts, updates,

768 deletes, and order changes.

769

770 Args:

771 project_id: Project id in the database.

772 samples_list: New list of sample dicts (must include pephub_sample_id).

773 sample_name_key: Sample table index key.

774 history_sa_model: HistoryProjects row to append change records to.

775

776 Raises:

777 ProjectDuplicatedSampleGUIDsError: If any pephub_sample_id is duplicated.

778 """

779

780 with Session(self._sa_engine) as session:

781 old_samples = session.scalars(

782 select(Samples).where(Samples.project_id == project_id)

783 )

784

785 old_samples_mapping: dict = {sample.guid: sample for sample in old_samples}

786

787 # old_child_parent_id needed because of the parent_guid is sometimes set to none in sqlalchemy mapping :( bug

788 old_child_parent_id: dict[str, str] = {

789 child: mapping.parent_guid

790 for child, mapping in old_samples_mapping.items()

791 }

792

793 old_samples_ids_set: set = set(old_samples_mapping.keys())

794 new_samples_ids_list: list = [

795 new_sample[PEPHUB_SAMPLE_ID_KEY]

796 for new_sample in samples_list

797 if new_sample[PEPHUB_SAMPLE_ID_KEY] != ""

798 and new_sample[PEPHUB_SAMPLE_ID_KEY] is not None

799 ]

800 new_samples_ids_set: set = set(new_samples_ids_list)

801 new_samples_dict: dict = {

802 new_sample[PEPHUB_SAMPLE_ID_KEY] or generate_guid(): new_sample

803 for new_sample in samples_list

804 }

805

806 if len(new_samples_ids_list) != len(new_samples_ids_set):

807 raise ProjectDuplicatedSampleGUIDsError(

808 f"Samples have to have unique pephub_sample_id: '{PEPHUB_SAMPLE_ID_KEY}'."

809 f"If ids are duplicated, overwrite the project."

810 )

811

812 # Check if something was deleted:

813 deleted_ids = old_samples_ids_set - new_samples_ids_set

814

815 del new_samples_ids_list, new_samples_ids_set

816

817 for remove_id in deleted_ids:

818 if history_sa_model:

819 history_sa_model.sample_changes_mapping.append(

820 HistorySamples(

821 guid=old_samples_mapping[remove_id].guid,

822 parent_guid=old_child_parent_id[remove_id],

823 sample_json=old_samples_mapping[remove_id].sample,

824 change_type=UpdateTypes.DELETE,

825 )

826 )

827 session.delete(old_samples_mapping[remove_id])

828

829 parent_id = None

830 parent_mapping = None

831

832 # Main loop to update samples

833 for current_id, sample_value in new_samples_dict.items():

834 new_sample = None

835 del sample_value[PEPHUB_SAMPLE_ID_KEY]

836

837 if current_id not in old_samples_ids_set:

838 new_sample = Samples(

839 sample=sample_value,

840 guid=current_id,

841 sample_name=sample_value[sample_name_key],

842 project_id=project_id,

843 parent_mapping=parent_mapping,

844 )

845 session.add(new_sample)

846

847 if history_sa_model:

848 history_sa_model.sample_changes_mapping.append(

849 HistorySamples(

850 guid=new_sample.guid,

851 parent_guid=new_sample.parent_guid,

852 sample_json=new_sample.sample,

853 change_type=UpdateTypes.INSERT,

854 )

855 )

856

857 else:

858 current_history = None

859 if old_samples_mapping[current_id].sample != sample_value:

860 if history_sa_model:

861 current_history = HistorySamples(

862 guid=old_samples_mapping[current_id].guid,

863 parent_guid=old_samples_mapping[current_id].parent_guid,

864 sample_json=old_samples_mapping[current_id].sample,

865 change_type=UpdateTypes.UPDATE,

866 )

867

868 old_samples_mapping[current_id].sample = sample_value

869 old_samples_mapping[current_id].sample_name = sample_value[

870 sample_name_key

871 ]

872

873 # !bug workaround: if project was deleted and sometimes old_samples_mapping[current_id].parent_guid

874 # and it can cause an error in history. For this we have `old_child_parent_id` dict

875 if old_samples_mapping[current_id].parent_guid != parent_id:

876 if history_sa_model:

877 if current_history:

878 current_history.parent_guid = parent_id

879 else:

880 current_history = HistorySamples(

881 guid=old_samples_mapping[current_id].guid,

882 parent_guid=old_child_parent_id[current_id],

883 sample_json=old_samples_mapping[current_id].sample,

884 change_type=UpdateTypes.UPDATE,

885 )

886 old_samples_mapping[current_id].parent_mapping = parent_mapping

887

888 if history_sa_model and current_history:

889 history_sa_model.sample_changes_mapping.append(current_history)

890

891 parent_id = current_id

892 parent_mapping = new_sample or old_samples_mapping[current_id]

893

894 session.commit()

895

896 @staticmethod

897 def __create_update_dict(update_values: UpdateItems) -> dict:

898 """Build a normalised update dict from an UpdateItems model.

899

900 Args:

901 update_values: UpdateItems with fields to apply.

902

903 Returns:

904 Dict of field→value pairs ready for setattr on the Projects row.

905 """

906 update_final = UpdateModel.model_construct()

907

908 if update_values.name is not None:

909 if update_values.config is not None:

910 update_values.config[NAME_KEY] = update_values.name

911 update_final = UpdateModel(

912 name=update_values.name,

913 **update_final.model_dump(exclude_unset=True),

914 )

915

916 if update_values.description is not None:

917 if update_values.config is not None:

918 update_values.config[DESCRIPTION_KEY] = update_values.description

919 update_final = UpdateModel(

920 description=update_values.description,

921 **update_final.model_dump(exclude_unset=True),

922 )

923 if update_values.config is not None:

924 update_final = UpdateModel(

925 config=update_values.config,

926 **update_final.model_dump(exclude_unset=True),

927 )

928 name = update_values.config.get(NAME_KEY)

929 description = update_values.config.get(DESCRIPTION_KEY)

930 if name:

931 update_final = UpdateModel(

932 name=name,

933 **update_final.model_dump(exclude_unset=True, exclude={NAME_KEY}),

934 )

935 if description:

936 update_final = UpdateModel(

937 description=description,

938 **update_final.model_dump(

939 exclude_unset=True, exclude={DESCRIPTION_KEY}

940 ),

941 )

942

943 if update_values.tag is not None:

944 update_final = UpdateModel(

945 tag=update_values.tag, **update_final.model_dump(exclude_unset=True)

946 )

947

948 if update_values.is_private is not None:

949 update_final = UpdateModel(

950 is_private=update_values.is_private,

951 **update_final.model_dump(exclude_unset=True),

952 )

953 if update_values.pop is not None:

954 update_final = UpdateModel(

955 pop=update_values.pop,

956 **update_final.model_dump(exclude_unset=True),

957 )

958

959 if update_values.pep_schema is not None:

960 update_final = UpdateModel(

961 pep_schema=update_values.pep_schema,

962 **update_final.model_dump(exclude_unset=True),

963 )

964

965 if update_values.number_of_samples is not None:

966 update_final = UpdateModel(

967 number_of_samples=update_values.number_of_samples,

968 **update_final.model_dump(exclude_unset=True),

969 )

970

971 return update_final.model_dump(exclude_unset=True, exclude_none=True)

972

973 def exists(

974 self,

975 namespace: str,

976 name: str,

977 tag: str = DEFAULT_TAG,

978 ) -> bool:

979 """Check if a project exists in the database.

980

981 Args:

982 namespace: Project namespace.

983 name: Project name.

984 tag: Project tag.

985

986 Returns:

987 True if the project exists.

988 """

989

990 statement = select(Projects.id)

991 statement = statement.where(

992 and_(

993 Projects.namespace == namespace,

994 Projects.name == name,

995 Projects.tag == tag,

996 )

997 )

998 found_prj = self._pep_db_engine.session_execute(statement).all()

999

1000 if len(found_prj) > 0:

1001 return True

1002 else:

1003 return False

1004

1005 @staticmethod

1006 def _add_samples_to_project(

1007 projects_sa: Projects,

1008 samples: list[dict],

1009 sample_table_index: str = "sample_name",

1010 ) -> None:

1011 """Append sample rows to a Projects ORM object.

1012

1013 Args:

1014 projects_sa: Projects ORM object in an open session.

1015 samples: List of sample dicts to add.

1016 sample_table_index: Column name to use as the sample identifier.

1017 """

1018 previous_sample_guid = None

1019 for sample in samples:

1020 sample = Samples(

1021 sample=sample,

1022 sample_name=sample.get(sample_table_index),

1023 parent_guid=previous_sample_guid,

1024 guid=generate_guid(),

1025 )

1026 projects_sa.samples_mapping.append(sample)

1027 previous_sample_guid = sample.guid

1028

1029 return None

1030

1031 @staticmethod

1032 def _add_subsamples_to_project(

1033 projects_sa: Projects, subsamples: list[list[dict]]

1034 ) -> None:

1035 """Append subsample rows to a Projects ORM object.

1036

1037 Args:

1038 projects_sa: Projects ORM object in an open session.

1039 subsamples: List of subsample groups (list of list of dicts).

1040 """

1041 for i, subs in enumerate(subsamples):

1042 for row_number, sub_item in enumerate(subs):

1043 projects_sa.subsamples_mapping.append(

1044 Subsamples(

1045 subsample=sub_item, subsample_number=i, row_number=row_number

1046 )

1047 )

1048

1049 def get_project_id(self, namespace: str, name: str, tag: str) -> int | None:

1050 """Get the database id of a project.

1051

1052 Args:

1053 namespace: Project namespace.

1054 name: Project name.

1055 tag: Project tag.

1056

1057 Returns:

1058 Project id, or None if the project does not exist.

1059 """

1060 statement = select(Projects.id).where(

1061 and_(

1062 Projects.namespace == namespace,

1063 Projects.name == name,

1064 Projects.tag == tag,

1065 )

1066 )

1067 with Session(self._sa_engine) as session:

1068 result = session.execute(statement).one_or_none()

1069

1070 if result:

1071 return result[0]

1072 return None

1073

1074 def fork(

1075 self,

1076 original_namespace: str,

1077 original_name: str,

1078 original_tag: str,

1079 fork_namespace: str,

1080 fork_name: str = None,

1081 fork_tag: str = None,

1082 description: str = None,

1083 private: bool = False,

1084 ) -> None:

1085 """Fork a project into a new namespace.

1086

1087 Args:

1088 original_namespace: Namespace of the source project.

1089 original_name: Name of the source project.

1090 original_tag: Tag of the source project.

1091 fork_namespace: Target namespace.

1092 fork_name: Name for the fork (default: same as original).

1093 fork_tag: Tag for the fork (default: same as original).

1094 description: Description for the fork.

1095 private: Mark fork as private if True.

1096

1097 Raises:

1098 ProjectNotFoundError: If the source project does not exist.

1099 """

1100

1101 self.create(

1102 project=self.get(

1103 namespace=original_namespace,

1104 name=original_name,

1105 tag=original_tag,

1106 raw=True,

1107 ),

1108 namespace=fork_namespace,

1109 name=fork_name,

1110 tag=fork_tag,

1111 description=description or None,

1112 is_private=private,

1113 )

1114 original_statement = select(Projects).where(

1115 and_(

1116 Projects.namespace == original_namespace,

1117 Projects.name == original_name,

1118 Projects.tag == original_tag,

1119 )

1120 )

1121 fork_statement = select(Projects).where(

1122 and_(

1123 Projects.namespace == fork_namespace,

1124 Projects.name == fork_name,

1125 Projects.tag == fork_tag,

1126 )

1127 )

1128

1129 with Session(self._sa_engine) as session:

1130 original_prj = session.scalar(original_statement)

1131 fork_prj = session.scalar(fork_statement)

1132

1133 fork_prj.forked_from_id = original_prj.id

1134 fork_prj.pop = original_prj.pop

1135 fork_prj.submission_date = original_prj.submission_date

1136 fork_prj.schema_id = original_prj.schema_id

1137 fork_prj.description = description or original_prj.description

1138

1139 session.commit()

1140

1141 def get_config(self, namespace: str, name: str, tag: str) -> dict | None:

1142 """Get the config dict for a project.

1143

1144 Args:

1145 namespace: Project namespace.

1146 name: Project name.

1147 tag: Project tag.

1148

1149 Returns:

1150 Config dict, or None if the project does not exist.

1151 """

1152 statement = select(Projects.config).where(

1153 and_(

1154 Projects.namespace == namespace,

1155 Projects.name == name,

1156 Projects.tag == tag,

1157 )

1158 )

1159 with Session(self._sa_engine) as session:

1160 result = session.execute(statement).one_or_none()

1161

1162 if result:

1163 return result[0]

1164 return None

1165

1166 def get_subsamples(self, namespace: str, name: str, tag: str) -> list | None:

1167 """Get the subsample list for a project.

1168

1169 Args:

1170 namespace: Project namespace.

1171 name: Project name.

1172 tag: Project tag.

1173

1174 Returns:

1175 List of subsample groups, or an empty list if there are no subsamples.

1176

1177 Raises:

1178 ProjectNotFoundError: If the project does not exist.

1179 """

1180 statement = self._create_select_statement(name, namespace, tag)

1181

1182 with Session(self._sa_engine) as session:

1183 found_prj = session.scalar(statement)

1184

1185 if found_prj:

1186 _LOGGER.info(

1187 f"Project has been found: {found_prj.namespace}, {found_prj.name}"

1188 )

1189 subsample_dict = {}

1190 if found_prj.subsamples_mapping:

1191 for subsample in found_prj.subsamples_mapping:

1192 if subsample.subsample_number not in subsample_dict.keys():

1193 subsample_dict[subsample.subsample_number] = []

1194 subsample_dict[subsample.subsample_number].append(

1195 subsample.subsample

1196 )

1197 return list(subsample_dict.values())

1198 else:

1199 return []

1200 else:

1201 raise ProjectNotFoundError(

1202 f"No project found for supplied input: '{namespace}/{name}:{tag}'. "

1203 f"Did you supply a valid namespace and project?"

1204 )

1205

1206 def get_samples(

1207 self,

1208 namespace: str,

1209 name: str,

1210 tag: str,

1211 raw: bool = True,

1212 with_ids: bool = False,

1213 ) -> list:

1214 """Get samples for a project.

1215

1216 Args:

1217 namespace: Project namespace.

1218 name: Project name.

1219 tag: Project tag.

1220 raw: Return raw dicts if True, records-orient pandas dicts if False.

1221 with_ids: Include pephub_sample_id in each sample if True.

1222

1223 Returns:

1224 List of sample dicts.

1225 """

1226 if raw:

1227 return self.get(

1228 namespace=namespace, name=name, tag=tag, raw=True, with_id=with_ids

1229 ).get(SAMPLE_RAW_DICT_KEY)

1230 return (

1231 self.get(

1232 namespace=namespace, name=name, tag=tag, raw=False, with_id=with_ids

1233 )

1234 .to_pandas()

1235 .replace({np.nan: None})

1236 .to_dict(orient="records")

1237 )

1238

1239 def get_history(

1240 self, namespace: str, name: str, tag: str

1241 ) -> HistoryAnnotationModel:

1242 """Get the list of history entries for a project.

1243

1244 Args:

1245 namespace: Project namespace.

1246 name: Project name.

1247 tag: Project tag.

1248

1249 Returns:

1250 HistoryAnnotationModel with a list of HistoryChangeModel entries.

1251 """

1252

1253 with Session(self._sa_engine) as session:

1254 statement = (

1255 select(HistoryProjects)

1256 .where(

1257 HistoryProjects.project_id

1258 == select(Projects.id)

1259 .where(

1260 and_(

1261 Projects.namespace == namespace,

1262 Projects.name == name,

1263 Projects.tag == tag,

1264 )

1265 )

1266 .scalar_subquery()

1267 )

1268 .order_by(HistoryProjects.update_time.desc())

1269 )

1270 results = session.scalars(statement)

1271 return_results: list = []

1272

1273 if results:

1274 for result in results:

1275 return_results.append(

1276 HistoryChangeModel(

1277 change_id=result.id,

1278 change_date=result.update_time,

1279 user=result.user,

1280 )

1281 )

1282 return HistoryAnnotationModel(

1283 namespace=namespace,

1284 name=name,

1285 tag=tag,

1286 history=return_results,

1287 )

1288

1289 def get_project_from_history(

1290 self,

1291 namespace: str,

1292 name: str,

1293 tag: str,

1294 history_id: int,

1295 raw: bool = True,

1296 with_id: bool = False,

1297 ) -> dict | peprs.Project:

1298 """Reconstruct a project at a past history checkpoint.

1299

1300 Args:

1301 namespace: Project namespace.

1302 name: Project name.

1303 tag: Project tag.

1304 history_id: Id of the history entry to restore to.

1305 raw: Return raw dict if True, peprs.Project object if False.

1306 with_id: Include pephub_sample_id in each sample if True.

1307

1308 Returns:

1309 Raw dict or peprs.Project at the requested history state.

1310

1311 Raises:

1312 ProjectNotFoundError: If the project does not exist.

1313 HistoryNotFoundError: If the history entry does not exist.

1314 """

1315

1316 with Session(self._sa_engine) as session:

1317 project_mapping = session.scalar(

1318 select(Projects).where(

1319 and_(

1320 Projects.namespace == namespace,

1321 Projects.name == name,

1322 Projects.tag == tag,

1323 )

1324 )

1325 )

1326 if not project_mapping:

1327 raise ProjectNotFoundError(

1328 f"No project found for supplied input: '{namespace}/{name}:{tag}'. "

1329 f"Did you supply a valid namespace and project?"

1330 )

1331

1332 sample_dict = self._get_samples_dict(

1333 prj_id=project_mapping.id, session=session, with_id=True

1334 )

1335

1336 main_history = session.scalar(

1337 select(HistoryProjects)

1338 .where(

1339 and_(

1340 HistoryProjects.project_id == project_mapping.id,

1341 HistoryProjects.id == history_id,

1342 )

1343 )

1344 .order_by(HistoryProjects.update_time.desc())

1345 )

1346 if not main_history:

1347 raise HistoryNotFoundError(

1348 f"No history found for supplied input: '{namespace}/{name}:{tag}'. "

1349 f"Did you supply a valid history id?"

1350 )

1351

1352 changes_mappings = session.scalars(

1353 select(HistoryProjects)

1354 .where(

1355 and_(

1356 HistoryProjects.project_id == project_mapping.id,

1357 )

1358 )

1359 .order_by(HistoryProjects.update_time.desc())

1360 )

1361

1362 # Changes mapping is a ordered list from most early to latest changes

1363 # We have to loop through each change and apply it to the sample list

1364 # It should be done before we found the history_id that user is looking for

1365 project_config = None

1366

1367 for result in changes_mappings:

1368 sample_dict = self._apply_history_changes(sample_dict, result)

1369

1370 if result.id == history_id:

1371 project_config = result.project_yaml

1372 break

1373

1374 samples_list = order_samples(sample_dict)

1375 ordered_samples_list = [sample["sample"] for sample in samples_list]

1376

1377 if not with_id:

1378 for sample in ordered_samples_list:

1379 try:

1380 del sample[PEPHUB_SAMPLE_ID_KEY]

1381 except KeyError:

1382 pass

1383

1384 if raw:

1385 return {

1386 CONFIG_KEY: project_config or project_mapping.config,

1387 SAMPLE_RAW_DICT_KEY: ordered_samples_list,

1388 SUBSAMPLE_RAW_DICT_KEY: self.get_subsamples(namespace, name, tag),

1389 }

1390 return peprs.Project.from_dict(

1391 pep_dictionary={

1392 CONFIG_KEY: project_config or project_mapping.config,

1393 SAMPLE_RAW_DICT_KEY: ordered_samples_list,

1394 SUBSAMPLE_RAW_DICT_KEY: self.get_subsamples(namespace, name, tag),

1395 }

1396 )

1397

1398 @staticmethod

1399 def _apply_history_changes(sample_dict: dict, change: HistoryProjects) -> dict:

1400 """Apply a history change record to a sample dict.

1401

1402 Args:

1403 sample_dict: Current sample dict keyed by guid.

1404 change: HistoryProjects entry whose sample_changes_mapping to apply.

1405

1406 Returns:

1407 Updated sample dict.

1408 """

1409 for sample_change in change.sample_changes_mapping:

1410 sample_id = sample_change.guid

1411

1412 if sample_change.change_type == UpdateTypes.UPDATE:

1413 sample_dict[sample_id]["sample"] = sample_change.sample_json

1414 sample_dict[sample_id]["sample"][PEPHUB_SAMPLE_ID_KEY] = (

1415 sample_change.guid

1416 )

1417 sample_dict[sample_id]["parent_guid"] = sample_change.parent_guid

1418

1419 elif sample_change.change_type == UpdateTypes.DELETE:

1420 sample_dict[sample_id] = {

1421 "sample": sample_change.sample_json,

1422 "guid": sample_id,

1423 "parent_guid": sample_change.parent_guid,

1424 }

1425

1426 elif sample_change.change_type == UpdateTypes.INSERT:

1427 del sample_dict[sample_id]

1428

1429 return sample_dict

1430

1431 def delete_history(

1432 self, namespace: str, name: str, tag: str, history_id: int | None = None

1433 ) -> None:

1434 """Delete one or all history entries for a project.

1435

1436 Args:

1437 namespace: Project namespace.

1438 name: Project name.

1439 tag: Project tag.

1440 history_id: Id of the entry to delete; deletes all history if None.

1441

1442 Raises:

1443 ProjectNotFoundError: If the project does not exist.

1444 HistoryNotFoundError: If the specified history entry does not exist.

1445 """

1446 with Session(self._sa_engine) as session:

1447 project_mapping = session.scalar(

1448 select(Projects).where(

1449 and_(

1450 Projects.namespace == namespace,

1451 Projects.name == name,

1452 Projects.tag == tag,

1453 )

1454 )

1455 )

1456 if not project_mapping:

1457 raise ProjectNotFoundError(

1458 f"No project found for supplied input: '{namespace}/{name}:{tag}'. "

1459 f"Did you supply a valid namespace and project?"

1460 )

1461

1462 if history_id is None:

1463 session.execute(

1464 delete(HistoryProjects).where(

1465 HistoryProjects.project_id == project_mapping.id

1466 )

1467 )

1468 session.commit()

1469 return None

1470

1471 history_mapping = session.scalar(

1472 select(HistoryProjects).where(

1473 and_(

1474 HistoryProjects.project_id == project_mapping.id,

1475 HistoryProjects.id == history_id,

1476 )

1477 )

1478 )

1479 if not history_mapping:

1480 raise HistoryNotFoundError(

1481 f"No history found for supplied input: '{namespace}/{name}:{tag}'. "

1482 f"Did you supply a valid history id?"

1483 )

1484

1485 session.delete(history_mapping)

1486 session.commit()

1487

1488 def restore(

1489 self,

1490 namespace: str,

1491 name: str,

1492 tag: str,

1493 history_id: int,

1494 user: str = None,

1495 ) -> None:

1496 """Restore a project to a past history state.

1497

1498 Args:

1499 namespace: Project namespace.

1500 name: Project name.

1501 tag: Project tag.

1502 history_id: Id of the history entry to restore to.

1503 user: User performing the restore (default: namespace).

1504 """

1505

1506 restore_project = self.get_project_from_history(

1507 namespace=namespace,

1508 name=name,

1509 tag=tag,

1510 history_id=history_id,

1511 raw=True,

1512 with_id=True,

1513 )

1514 self.update(

1515 update_dict={"project": peprs.Project.from_dict(restore_project)},

1516 namespace=namespace,

1517 name=name,

1518 tag=tag,

1519 user=user or namespace,

1520 )

1521

1522 def clean_history(self, days: int = 90) -> None:

1523 """Delete history entries older than a given number of days.

1524

1525 Args:

1526 days: Retain history newer than this many days (default: 90).

1527 """

1528

1529 with Session(self._sa_engine) as session:

1530 session.execute(

1531 delete(HistoryProjects).where(

1532 HistoryProjects.update_time

1533 < (datetime.datetime.now() - datetime.timedelta(days=days))

1534 )

1535 )

1536 session.commit()

1537 _LOGGER.info("History was cleaned successfully!")