diff --git a/api/graphql/schema.py b/api/graphql/schema.py index d0ac1619f..9af3c8f8a 100644 --- a/api/graphql/schema.py +++ b/api/graphql/schema.py @@ -33,6 +33,7 @@ AssayInternal, AuditLogInternal, FamilyInternal, + FileInternal, ParticipantInternal, Project, SampleInternal, @@ -239,6 +240,7 @@ class GraphQLAnalysis: timestamp_completed: datetime.datetime | None = None active: bool meta: strawberry.scalars.JSON + output_files: strawberry.scalars.JSON @staticmethod def from_internal(internal: AnalysisInternal) -> 'GraphQLAnalysis': @@ -250,6 +252,7 @@ def from_internal(internal: AnalysisInternal) -> 'GraphQLAnalysis': timestamp_completed=internal.timestamp_completed, active=internal.active, meta=internal.meta, + output_files=FileInternal.reconstruct_json(internal.output_files), ) @strawberry.field diff --git a/db/project.xml b/db/project.xml index 92ae64a5c..9965db9a0 100644 --- a/db/project.xml +++ b/db/project.xml @@ -1149,7 +1149,12 @@ nullable="false" /> - + + + + @@ -1171,6 +1176,10 @@ /> + ALTER TABLE `file` ADD SYSTEM VERSIONING; diff --git a/db/python/connect.py b/db/python/connect.py index 971058f8f..09c8d5cf0 100644 --- a/db/python/connect.py +++ b/db/python/connect.py @@ -26,6 +26,7 @@ 'sequencing_group', 'assay', 'sequencing_group_assay', + 'analysis_file', 'analysis_sequencing_group', 'analysis_sample', 'assay_external_id', diff --git a/db/python/layers/analysis.py b/db/python/layers/analysis.py index f54c91eaa..1a28db98b 100644 --- a/db/python/layers/analysis.py +++ b/db/python/layers/analysis.py @@ -7,6 +7,7 @@ from db.python.layers.base import BaseLayer from db.python.layers.sequencing_group import SequencingGroupLayer from db.python.tables.analysis import AnalysisFilter, AnalysisTable +from db.python.tables.file import FileTable from db.python.tables.sample import SampleTable from db.python.tables.sequencing_group import SequencingGroupFilter from db.python.utils import GenericFilter, get_logger @@ -48,6 +49,7 @@ def __init__(self, connection: Connection): self.sampt = SampleTable(connection) self.at = AnalysisTable(connection) + self.ft = FileTable(connection) # GETS @@ -543,7 +545,7 @@ async def create_analysis( project: ProjectId = None, ) -> int: """Create a new analysis""" - return await self.at.create_analysis( + new_analysis_id = await self.at.create_analysis( analysis_type=analysis.type, status=analysis.status, sequencing_group_ids=analysis.sequencing_group_ids, @@ -553,6 +555,10 @@ async def create_analysis( project=project, ) + await self.ft.create_or_update_analysis_output_files_from_json(analysis_id=new_analysis_id, json_dict=analysis.output) + + return new_analysis_id + async def add_sequencing_groups_to_analysis( self, analysis_id: int, sequencing_group_ids: list[int], check_project_id=True ): @@ -591,6 +597,8 @@ async def update_analysis( output=output, ) + await self.ft.create_or_update_analysis_output_files_from_json(analysis_id=analysis_id, json_dict=output) + async def get_analysis_runner_log( self, project_ids: list[int] = None, diff --git a/db/python/tables/analysis.py b/db/python/tables/analysis.py index ee445345e..6ecfc65d6 100644 --- a/db/python/tables/analysis.py +++ b/db/python/tables/analysis.py @@ -15,6 +15,7 @@ from models.enums import AnalysisStatus from models.models.analysis import AnalysisInternal from models.models.audit_log import AuditLogInternal +from models.models.file import FileInternal from models.models.project import ProjectId @@ -73,7 +74,7 @@ async def create_analysis( ('type', analysis_type), ('status', status.value), ('meta', to_db_json(meta or {})), - ('output', output), + ('output', output), # can keep for now but will be deprecated eventually as a column ('audit_log_id', await self.audit_log_id()), ('project', project or self.project), ('active', active if active is not None else True), @@ -226,15 +227,36 @@ async def query(self, filter_: AnalysisFilter) -> List[AnalysisInternal]: rows = await self.connection.fetch_all(_query, values) retvals: Dict[int, AnalysisInternal] = {} + analysis_ids: list = [] for row in rows: key = row['id'] + analysis_ids.append(key) if key in retvals: retvals[key].sequencing_group_ids.append(row['sequencing_group_id']) else: retvals[key] = AnalysisInternal.from_db(**dict(row)) + analysis_outputs_by_aid = await self.get_file_outputs_by_analysis_ids(analysis_ids) + for analysis_id, analysis in retvals.items(): + retvals[analysis_id] = analysis.copy(update={'output_files': analysis_outputs_by_aid.get(analysis_id, [])}) return list(retvals.values()) + async def get_file_outputs_by_analysis_ids(self, analysis_ids: list[int]) -> dict[int, list[FileInternal]]: + """Fetches all output files for a list of analysis IDs""" + + _query = """ + SELECT af.analysis_id, f.* + FROM analysis_file af + INNER JOIN file f ON af.file_id = f.id + WHERE af.analysis_id IN :analysis_ids + """ + rows = await self.connection.fetch_all(_query, {'analysis_ids': analysis_ids}) + analysis_files: dict[int, list[FileInternal]] = defaultdict(list) + for row in rows: + analysis_files[row['analysis_id']].append(FileInternal.from_db(**dict(row))) + + return analysis_files + async def get_latest_complete_analysis_for_type( self, project: ProjectId, diff --git a/db/python/tables/file.py b/db/python/tables/file.py new file mode 100644 index 000000000..b98804ff4 --- /dev/null +++ b/db/python/tables/file.py @@ -0,0 +1,89 @@ +from textwrap import dedent +from typing import Optional, Union + +from db.python.tables.base import DbBase +from models.models.file import FileInternal + + +class FileTable(DbBase): + """ + Capture Analysis table operations and queries + """ + + table_name = 'file' + + async def create_analysis_output_file( + self, + path: str, + analysis_id: int, + json_structure: Optional[str] + ) -> int: + """Create a new file, and add it to an analysis via the join table""" + + id_of_new_file = await self.create_or_update_output_file(path, json_structure) + await self.add_output_file_to_analysis(analysis_id=analysis_id, file_id=id_of_new_file) + + return id_of_new_file + + async def create_or_update_output_file( + self, + path: str, + json_structure: Optional[str] = None + ) -> int: + """ + Create a new file, and add it to database + """ + kv_pairs = [ + ('path', path), + ('basename', FileInternal.get_basename(path)), + ('dirname', FileInternal.get_dirname(path)), + ('nameroot', FileInternal.get_nameroot(path)), + ('nameext', FileInternal.get_extension(path)), + ('checksum', FileInternal.get_checksum(path)), + ('size', FileInternal.get_size(path)), + ('json_structure', json_structure), + ] + + kv_pairs = [(k, v) for k, v in kv_pairs if v is not None] + keys = [k for k, _ in kv_pairs] + cs_keys = ', '.join(keys) + cs_id_keys = ', '.join(f':{k}' for k in keys) + # non_pk_keys = [k for k in keys if k != 'path'] + # update_clause = ', '.join([f"{k} = VALUES({k})" for k in non_pk_keys]) # ON DUPLICATE KEY UPDATE {update_clause} + + _query = dedent(f"""INSERT INTO file ({cs_keys}) VALUES ({cs_id_keys}) RETURNING id""") + + id_of_new_file = await self.connection.fetch_val( + _query, + dict(kv_pairs), + ) + + return id_of_new_file + + async def add_output_file_to_analysis(self, analysis_id: int, file_id: int): + """Add file to an analysis (through the join table)""" + _query = dedent(""" + INSERT INTO analysis_file + (analysis_id, file_id) + VALUES (:analysis_id, :file_id) + ON DUPLICATE KEY UPDATE + analysis_id = VALUES(analysis_id), + file_id = VALUES(file_id) + """) + await self.connection.execute( + _query, + {'analysis_id': analysis_id, 'file_id': file_id} + ) + + async def create_or_update_analysis_output_files_from_json( + self, + analysis_id: int, + json_dict: Union[dict, str], + ) -> None: + """ + Create analysis files from JSON + """ + async with self.connection.transaction(): + for json_path, path in FileInternal.find_files_from_dict(json_dict=json_dict): + file_id = await self.create_or_update_output_file(path=path, json_structure='.'.join(json_path)) + await self.add_output_file_to_analysis(analysis_id, file_id) diff --git a/db/python/tables/project.py b/db/python/tables/project.py index d87c28960..f8cb63525 100644 --- a/db/python/tables/project.py +++ b/db/python/tables/project.py @@ -442,6 +442,9 @@ async def delete_project_data( INNER JOIN sample ON sample.id = sg.sample_id WHERE sample.project = :project ); +DELETE FROM analysis_file WHERE analysis_id in ( + SELECT id FROM analysis WHERE project = :project +); DELETE FROM analysis_sample WHERE sample_id in ( SELECT s.id FROM sample s WHERE s.project = :project diff --git a/models/models/__init__.py b/models/models/__init__.py index 4b81e4c4e..dbb955bb8 100644 --- a/models/models/__init__.py +++ b/models/models/__init__.py @@ -27,7 +27,8 @@ PedRowInternal, ) from models.models.file import ( - File + File, + FileInternal ) from models.models.participant import ( NestedParticipant, diff --git a/models/models/analysis.py b/models/models/analysis.py index f16564854..4ac517a41 100644 --- a/models/models/analysis.py +++ b/models/models/analysis.py @@ -3,7 +3,7 @@ from datetime import date, datetime from typing import Any -from pydantic import BaseModel +from pydantic import BaseModel, ConfigDict from models.base import SMBase from models.enums import AnalysisStatus @@ -15,6 +15,7 @@ class AnalysisInternal(SMBase): """Model for Analysis""" + model_config = ConfigDict(extra='allow') id: int | None = None type: str diff --git a/models/models/file.py b/models/models/file.py index c6db4ad6f..6bb6eaa36 100644 --- a/models/models/file.py +++ b/models/models/file.py @@ -1,8 +1,10 @@ import hashlib -from typing import Generator import json from collections import defaultdict +from typing import Any, Generator, Union +import cloudpathlib.exceptions as cloudpathlib_exceptions +import google.api_core.exceptions as google_exceptions from cloudpathlib.anypath import AnyPath from pydantic import BaseModel @@ -13,23 +15,23 @@ class FileInternal(SMBase): """File model for internal use""" id: int - analysis_id: int path: str basename: str - dirname: AnyPath + dirname: str nameroot: str nameext: str | None checksum: str | None size: int + meta: str | None = None + # exists: bool = False json_path: str | None = None - secondary_files: list[str] = [] + # secondary_files: list[str] | None = [] @staticmethod def from_db(**kwargs): """ Convert from db keys, mainly converting id to id_ """ - analysis_id = kwargs.get('analysis_id') path = kwargs.get('path') basename = kwargs.get('basename') dirname = kwargs.get('dirname') @@ -37,12 +39,13 @@ def from_db(**kwargs): nameext = kwargs.get('nameext') checksum = kwargs.get('checksum') size = kwargs.get('size') + meta = kwargs.get('meta') + # exists = kwargs.get('exists') json_path = kwargs.get('json_path') - secondary_files = kwargs.get('secondary_files') + # secondary_files = kwargs.get('secondary_files') return File( id=kwargs.pop('id'), - analysis_id=analysis_id, path=path, basename=basename, dirname=dirname, @@ -50,8 +53,10 @@ def from_db(**kwargs): nameext=nameext, checksum=checksum, size=size, + meta=meta, + # exists=exists, json_path=json_path, - secondary_files=secondary_files, + # secondary_files=secondary_files, ) def to_external(self): @@ -60,7 +65,6 @@ def to_external(self): """ return File( id=self.id, - analysis_id=self.analysis_id, path=self.path, basename=self.basename, dirname=self.dirname, @@ -68,7 +72,10 @@ def to_external(self): nameext=self.nameext, checksum=self.checksum, size=self.size, - secondary_files=self.secondary_files, + meta=self.meta, + # exists=self.exists, + json_path=self.json_path, + # secondary_files=self.secondary_files, ) @staticmethod @@ -94,9 +101,11 @@ def get_extension(path: str) -> str: @staticmethod def get_checksum(path: str) -> str: """Get checksum for file at given path""" + if FileInternal.get_extension(path) == '.mt': + return None try: return hashlib.sha256(AnyPath(path).read_bytes()).hexdigest() # pylint: disable=E1101 - except FileNotFoundError: + except (google_exceptions.NotFound, FileNotFoundError): return None @staticmethod @@ -104,50 +113,64 @@ def get_size(path: str) -> int: """Get file size""" try: return AnyPath(path).stat().st_size # pylint: disable=E1101 - except FileNotFoundError: + except (google_exceptions.NotFound, cloudpathlib_exceptions.NoStatError, FileNotFoundError): return 0 @staticmethod - def find_files_from_dict(json_dict: dict, path=None) -> Generator[tuple[str, dict], None, None]: + def find_files_from_dict(json_dict: Union[dict, str], json_path=None) -> Generator[tuple[Any, Any], None, None]: """Retrieve filepaths from a dict of outputs""" - if not path: - path = [] + if isinstance(json_dict, str): + try: + json_dict = json.loads(json_dict) + except json.JSONDecodeError: + print(f'Error parsing JSON: {json_dict}') + + if not json_path: + json_path = [] if isinstance(json_dict, dict): for key, value in json_dict.items(): if isinstance(value, dict): # If the value is a dictionary, continue the recursion - yield from FileInternal.find_files_from_dict(value, path + [key]) + yield from FileInternal.find_files_from_dict(value, json_path + [key]) else: - # Found a leaf, yield the path and the value - yield path, json_dict + # Found a leaf, yield the json_path and the value dict containing file_path + yield json_path, json_dict else: # If the input is not a dictionary, just return the value - yield path, json_dict + yield json_path, json_dict # type: ignore[misc] @staticmethod - def reconstruct_json(data) -> dict: + def reconstruct_json(data: list) -> list | dict: """Reconstruct a JSON object from a list of paths and values""" - root: dict = defaultdict(dict) - for path_str, content_str in data: - # Split the path into components and parse the JSON content - path = path_str.split('.') - - try: - content = json.loads(content_str) - except json.JSONDecodeError: - print(f'Error parsing JSON: {content_str}') - - # Navigate down the tree to the correct position, creating dictionaries as needed - current = root - for key in path[:-1]: - current = current.setdefault(key, {}) - - if path[-1] in current: - current[path[-1]].update(content) - else: - current[path[-1]] = content + # TODO change to dict once new input structure is in place + root: list = [] + for file in data: + file_root: dict = defaultdict(dict) + fields = FileInternal.__fields__.keys() # type:ignore[attr-defined] + for field in fields: + file_root[field] = getattr(file, field) + + # if file.json_structure: + # # Split the path into components and parse the JSON content + # path = file.json_path.split('.') + # try: + # content = json.loads(file_root) + # except json.JSONDecodeError: + # print(f'Error parsing JSON: {file_root}') + + # # Navigate down the tree to the correct position, creating dictionaries as needed + # current = root + # for key in path[:-1]: + # current = current.setdefault(key, []) + + # if path[-1] in current: + # current[path[-1]].append(content) + # else: + # current[path[-1]] = content + # else: + root.append(file_root) return root @@ -155,16 +178,17 @@ class File(BaseModel): """File model for external use""" id: int - analysis_id: int path: str basename: str - dirname: AnyPath + dirname: str nameroot: str nameext: str | None checksum: str | None size: int + meta: str | None = None + # exists: bool = False json_path: str | None = None - secondary_files: list[str] = [] + # secondary_files: list[str] | None = [] def to_internal(self): """ @@ -172,7 +196,6 @@ def to_internal(self): """ return FileInternal( id=self.id, - analysis_id=self.analysis_id, path=self.path, basename=self.basename, dirname=self.dirname, @@ -180,6 +203,8 @@ def to_internal(self): nameext=self.nameext, checksum=self.checksum, size=self.size, + meta=self.meta, + # exists=self.exists, json_path=self.json_path, - secondary_files=self.secondary_files, + # secondary_files=self.secondary_files, ) diff --git a/scripts/20240124_migrate_output_to_file.py b/scripts/20240124_migrate_output_to_file.py index 13eda8029..9051f15f1 100644 --- a/scripts/20240124_migrate_output_to_file.py +++ b/scripts/20240124_migrate_output_to_file.py @@ -55,7 +55,7 @@ def get_file_info(path: str) -> Dict: 'nameext': FileInternal.get_extension(path), 'checksum': FileInternal.get_checksum(path), 'size': FileInternal.get_size(path), - 'secondary_files': '[]', + 'json_structure': '', } @@ -101,8 +101,8 @@ async def prepare_files(analyses): async def insert_files(connection, files): """Insert files""" query = dedent( - """INSERT INTO file (path, basename, dirname, nameroot, nameext, checksum, size, secondary_files) - VALUES (:path, :basename, :dirname, :nameroot, :nameext, :checksum, :size, :secondary_files) + """INSERT INTO file (path, basename, dirname, nameroot, nameext, checksum, size, json_structure) + VALUES (:path, :basename, :dirname, :nameroot, :nameext, :checksum, :size, :json_structure) RETURNING id""" ) af_query = dedent(