From 044ce7d4700ec7136b39d0889bba0b7fed8ecc02 Mon Sep 17 00:00:00 2001 From: Max Levinson Date: Thu, 4 Jun 2026 10:28:30 -0400 Subject: [PATCH 1/2] ROCrate Zip debugging --- mds/src/fairscape_mds/crud/rocrate.py | 388 +++++++++++++++++++------- 1 file changed, 289 insertions(+), 99 deletions(-) diff --git a/mds/src/fairscape_mds/crud/rocrate.py b/mds/src/fairscape_mds/crud/rocrate.py index 127f392..9187254 100644 --- a/mds/src/fairscape_mds/crud/rocrate.py +++ b/mds/src/fairscape_mds/crud/rocrate.py @@ -32,9 +32,100 @@ import re import botocore import mimetypes +import io +import zipfile # ROCrate Helper Functions + +class _S3SeekableFile(io.RawIOBase): + """Seekable file-like object backed by S3 range requests. + + Implements only the methods zipfile.ZipFile needs so it can navigate + the ZIP central directory without downloading the whole object. + """ + + def __init__(self, s3_client, bucket: str, key: str): + self._client = s3_client + self._bucket = bucket + self._key = key + self._pos = 0 + self._size: int | None = None + + def _object_size(self) -> int: + if self._size is None: + resp = self._client.head_object(Bucket=self._bucket, Key=self._key) + self._size = resp["ContentLength"] + return self._size + + def seek(self, offset: int, whence: int = 0) -> int: + size = self._object_size() + if whence == 0: + self._pos = offset + elif whence == 1: + self._pos += offset + elif whence == 2: + self._pos = size + offset + self._pos = max(0, min(self._pos, size)) + return self._pos + + def tell(self) -> int: + return self._pos + + def read(self, n: int = -1) -> bytes: + size = self._object_size() + if self._pos >= size: + return b"" + end = (size - 1) if n < 0 else min(self._pos + n - 1, size - 1) + resp = self._client.get_object( + Bucket=self._bucket, + Key=self._key, + Range=f"bytes={self._pos}-{end}", + ) + data = resp["Body"].read() + self._pos += len(data) + return data + + def readable(self) -> bool: + return True + + def seekable(self) -> bool: + return True + + +def get_s3_zip_infolist(s3_client, bucket: str, key: str) -> list[zipfile.ZipInfo]: + """Return ZipFile.infolist() for a ZIP stored in S3 using range requests. + + Makes ~3 S3 API calls regardless of ZIP size: + 1. HeadObject to get file size + 2. GetObject(Range) for the last ~64 KB (EOCD + any comment) + 3. GetObject(Range) for the central directory + """ + with zipfile.ZipFile(_S3SeekableFile(s3_client, bucket, key)) as zf: + return zf.infolist() + + +def read_s3_zip_member(s3_client, bucket: str, key: str, member: str) -> bytes: + """Read a single member from a ZIP stored in S3 without downloading the whole file. + + Fetches the central directory (via range requests) to locate the member's + local header, then downloads only that member's compressed bytes. + + Args: + s3_client: boto3 S3 client + bucket: S3 bucket name + key: S3 object key for the ZIP file + member: path of the member inside the ZIP (as it appears in infolist) + + Raises: + KeyError: if the member is not found in the archive + """ + with zipfile.ZipFile(_S3SeekableFile(s3_client, bucket, key)) as zf: + with zf.open(member) as f: + return f.read() + + + def userPath(inputEmail): searchResults = re.search("(^[a-zA-Z-1-9_.+-]+)@", inputEmail) @@ -141,12 +232,125 @@ def buildContentSummary(rocrateInstance: ROCrateV1_2) -> ROCrateContentSummary: ) +class ROCrateUploadException(Exception): + def __init__(self, message: str): + super().__init__(message) + + +class ROCrateUploadCrateNotFound(ROCrateUploadException): + def __init__(self, message: str, job: ROCrateUploadRequest): + super().__init__(message) + self.job = job + + + +def findRootCrate(infolist: list[zipfile.ZipInfo]) -> tuple[str | None, list[str]]: + """ Given an Infolist from a Zip Archive, find all `ro-crate-metadata.json` files and return a tuple with the first member as the path for the root crate and the second as the list of all subcrates + """ + + metadataFiles = [ + elem.filename for elem in + filter(lambda x: 'ro-crate-metadata.json' in x.filename, infolist) + ] + + # No ROCrate Metadata Found + if len(metadataFiles) == 0: + return None, [] + + # Exactly one rocrate found + elif len(metadataFiles) == 1: + return metadataFiles[0], [] + + # multiple rocrates found + elif len(metadataFiles) > 1: + subdirectoryCount = [ elem.count("/") for elem in metadataFiles ] + + # TODO if there are multiple root crates + if subdirectoryCount.count(min(subdirectoryCount)) > 1: + raise Exception() + + rootCrate = metadataFiles[subdirectoryCount.index(min(subdirectoryCount))] + + metadataFiles.pop[metadataFiles.index(rootCrate)] + + return rootCrate, metadataFiles + + +def getROCrateMetadata( + uploadJob: ROCrateUploadRequest, + s3Client, + s3Bucket: str + ) -> tuple[bytes, list[bytes], bool]: + + infolist = get_s3_zip_infolist( + s3Client, + s3Bucket, + uploadJob.uploadPath + ) + + # find root crate from zipfile + rootCratePath, subcrates = findRootCrate(infolist) + + try: + rootCrateMetadata = read_s3_zip_member( + s3Client, + s3Bucket, + uploadJob.uploadPath, + rootCratePath + ) + + # TODO handle errors for rocrate not found + except KeyError: + raise ROCrateUploadCrateNotFound( + message="ROCrateException: Root Crate Not Found", + job=uploadJob + ) + + subcrateMetadata = [] + + for subcratePath in subcrates: + try: + subcrateMetadataElem = read_s3_zip_member( + s3Client, + s3Bucket, + uploadJob.uploadPath, + subcratePath + ) + subcrateMetadata.append(subcrateMetadataElem) + + except KeyError: + raise ROCrateUploadCrateNotFound( + message="ROCrateException: Sub Crate Not Found", + job=uploadJob + ) + + + return rootCrateMetadata, subcrateMetadata + + + + class FairscapeROCrateRequest(FairscapeRequest): def __init__(self, config): super().__init__(config) self.config = config + def updateJobStatus( + self, + uploadJobGUID: str, + update: dict + )-> None: + + updateResponse = self.config.asyncCollection.update_one( + {"guid": uploadJobGUID}, + update + ) + + assert updateResponse.matched_count == 1 + assert updateResponse.modified_count == 1 + + def uploadROCrate( self, userInstance: UserWriteModel, @@ -508,6 +712,7 @@ def processTaskWriteMLModels( pass + def processTaskWriteMetadataElements( self, userInstance, @@ -626,7 +831,7 @@ def getROCrateContentsMinio(self, zipCratePath: str): return objectList - def processTaskGetInitialJobMetadata(self, transactionGUID: str): + def processTaskGetInitialJobMetadata(self, transactionGUID: str) -> tuple[ UserWriteModel, ROCrateUploadRequest]: uploadMetadata = self.config.asyncCollection.find_one( {"guid": transactionGUID}, {"_id": 0} @@ -693,14 +898,10 @@ def processROCrate(self, transactionGUID: str): now = datetime.datetime.now() - self.config.asyncCollection.update_one( - {"guid": transactionGUID}, - {"$set": - { - "stage": "starting job" - } - } - ) + self.updateJobStatus( + transactionGUID, + {"$set": {"stage": "starting job"}} + ) foundUser, uploadInstance = self.processTaskGetInitialJobMetadata(transactionGUID) zippedCratePath = uploadInstance.uploadPath @@ -715,76 +916,89 @@ def processROCrate(self, transactionGUID: str): uploadPathString = uploadInstance.uploadPath if not uploadPathString: - raise Exception( - "ROCrate Upload Job Missing Upload Path Property" - ) + self.updateJobStatus( + transactionGUID, + {"$set": { + "stage": "job failed", + "timeFinished": datetime.datetime.now(), + "success": False, + "completed": True, + "error": "ROCrate Upload Job Missing Upload Path Property" + }} + ) + return False + jobUploadPath = pathlib.PurePosixPath(uploadInstance.uploadPath) baseDirectory = uploadInstance.uploadPath - metadataKey = baseDirectory + "/ro-crate-metadata.json" - metadataFound = False - stem = jobUploadPath.stem - includeStem = False + # determine if path requires stem + infolist = get_s3_zip_infolist( + self.config.minioClient, + self.config.minioBucket, + uploadInstance.uploadPath + ) - try: - s3Response = self.config.minioClient.get_object( - Bucket = self.config.minioBucket, - Key = metadataKey - ) - metadataFound = True - except self.config.minioClient.exceptions.NoSuchKey: - metadataFound = False + rootCrate, subcrates = findRootCrate(infolist) - if not metadataFound: + if rootCrate.count("/") > 0: + stem = pathlib.Path(rootCrate).parent._str + includeStem = True + else: + stem = "" + includeStem = False - metadataKey = f"{baseDirectory}/{jobUploadPath.stem}/ro-crate-metadata.json" - try: - s3Response = self.config.minioClient.get_object( - Bucket = self.config.minioBucket, - Key = metadataKey - ) - metadataFound = True - except self.config.minioClient.exceptions.NoSuchKey: - metadataFound = False + # get rocrate metadata + try: + roCrateJSON, subcrateJSON = getROCrateMetadata( + uploadInstance, + self.config.minioClient, + self.config.minioBucket + ) - if metadataFound: - includeStem = True - try: - content = s3Response['Body'] - roCrateJSON = json.loads(content.read()) - - except json.JSONDecodeError as e: - - self.config.asyncCollection.update_one( - {"guid": transactionGUID}, - {"$set": - { - "stage": "reading metadata", - "error": str(e), - "timeFinished": datetime.datetime.now(), - "success": False, - "completed": True - } - } - ) - raise Exception("Failed to Decode Metadata JSON") + self.updateJobStatus( + transactionGUID, + {"$set": {"stage": "found ro-crate-metadata"}} + ) - else: - raise Exception("Metadata Not Found in RO-Crate") + # TODO + except ROCrateUploadException as e: + self.updateJobStatus( + transactionGUID, + {"$set": { + "status": "job failed", + "timeFinished": datetime.datetime.now(), + "success": False, + "completed": True, + "error": str(e) + }} + ) + return False + + except Exception as e: + self.updateJobStatus( + transactionGUID, + {"$set": { + "status": "job failed", + "timeFinished": datetime.datetime.now(), + "success": False, + "completed": True, + "error": str(e) + }} + ) + return False - # validate try: - roCrateModel = ROCrateV1_2.model_validate(roCrateJSON) + roCrateModel = ROCrateV1_2.model_validate_json(roCrateJSON) except pydantic.ValidationError as e: print(f"ValidationError: {str(e)}") traceback.print_exc() # update job as failure - self.config.asyncCollection.update_one( - {"guid": transactionGUID}, + self.updateJobStatus( + transactionGUID, {"$set": { "stage": "reading metadata", @@ -796,7 +1010,7 @@ def processROCrate(self, transactionGUID: str): } ) - return None + return False crateMetadataElem = roCrateModel.getCrateMetadata() @@ -807,22 +1021,13 @@ def processROCrate(self, transactionGUID: str): roCrateGUID = crateMetadataElem.guid - self.config.asyncCollection.update_one( - {"guid": transactionGUID}, - {"$set": - { - "stage": "found metadata", - "rocrateGUID": roCrateGUID - } - } - ) - # check identifier conflicts - self.config.asyncCollection.update_one( - {"guid": transactionGUID}, + self.updateJobStatus( + transactionGUID, {"$set": { - "stage": "checking for identifier conflicts", + "stage": "processing metadata", + "rocrateGUID": roCrateGUID } } ) @@ -834,8 +1039,8 @@ def processROCrate(self, transactionGUID: str): if foundROCrateMetadata: # check identifier conflicts - self.config.asyncCollection.update_one( - {"guid": transactionGUID}, + self.updateJobStatus( + transactionGUID, {"$set": { "timeFinished": datetime.datetime.now(), @@ -852,11 +1057,11 @@ def processROCrate(self, transactionGUID: str): Key=uploadInstance.uploadPath ) - raise Exception(f"ROCrate Identifier Conflict: {roCrateGUID}") + return False - self.config.asyncCollection.update_one( - {"guid": transactionGUID}, + self.updateJobStatus( + transactionGUID, {"$set": { "stage": "minting datasets", @@ -936,22 +1141,11 @@ def processROCrate(self, transactionGUID: str): ) ) - # TODO check insert result is correct - - # TODO documents too large causes errors - # write the whole ROCrateV1_2 model into the rocrate collection - #rocrate_doc_for_collection = { - # "@id": metadataElem.guid, - # "@type": ['Dataset', "https://w3id.org/EVI#ROCrate"], - # "owner": foundUser.email, - # "permissions": foundUser.getPermissions().model_dump(mode='json', by_alias=True), - # "metadata": roCrateModel.model_dump(mode='json', by_alias=True) - #} - #self.config.rocrateCollection.insert_one(rocrate_doc_for_collection) - + assert insertResult.inserted_id + # update process as success - updateResult = self.config.asyncCollection.update_one( - {"guid": uploadInstance.guid}, + self.updateJobStatus( + uploadInstance.guid, {"$set": { "completed": True, "identifiersMinted": len(datasetGUIDS + nonDatasetGUIDS)+1, @@ -963,10 +1157,6 @@ def processROCrate(self, transactionGUID: str): }} ) - # check update result - if updateResult.modified_count != 1: - raise Exception(f"Failed to Update Job Metadata: {uploadInstance.guid}") - return metadataElem.guid From 51251040c3e75bde8a008255c9091d8eae964bb2 Mon Sep 17 00:00:00 2001 From: Max Levinson Date: Fri, 5 Jun 2026 12:50:00 -0400 Subject: [PATCH 2/2] Update .gitignore --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index 2610193..ec41a3d 100644 --- a/.gitignore +++ b/.gitignore @@ -169,3 +169,4 @@ tests/model-tests/ppi_gene_node_attributes.tsv mds/src/fairscape_mds/tests/descriptive_statistics.ipynb /.devcontainer .devcontainer/devcontainer.json +CLAUDE.md