Function get_metadata_from_s3

src/server/s3_utils.py:345–386

Retrieve metadata JSON from S3. Parameters ---------- s3_file_path : str The S3 file path for the digest (metadata will use .json extension). Returns ------- S3Metadata | None The metadata struct if found, None otherwise.

(s3_file_path: str)

def get_metadata_from_s3(s3_file_path: str) -> S3Metadata | None:
    """Retrieve metadata JSON from S3.

    Parameters
    ----------
    s3_file_path : str
        The S3 file path for the digest (metadata will use .json extension).

    Returns
    -------
    S3Metadata | None
        The metadata struct if found, None otherwise.

    """
    if not is_s3_enabled():
        return None

    # Generate metadata file path by replacing .txt with .json
    metadata_file_path = s3_file_path.replace(".txt", ".json")

    try:
        s3_client = create_s3_client()
        bucket_name = get_s3_bucket_name()

        # Get the metadata object
        response = s3_client.get_object(Bucket=bucket_name, Key=metadata_file_path)
        metadata_content = response["Body"].read().decode("utf-8")

        return S3Metadata.model_validate_json(metadata_content)
    except ClientError as err:
        # Object doesn't exist if we get a 404 error
        error_code = err.response.get("Error", {}).get("Code")
        if error_code == "404":
            logger.info("Metadata file not found", extra={"metadata_file_path": metadata_file_path})
            return None
        # Log other errors but don't fail
        logger.warning("Failed to retrieve metadata from S3", extra={"error": str(err)})
        return None
    except Exception as exc:
        # For any other exception, log and return None
        logger.warning("Unexpected error retrieving metadata from S3", extra={"error": str(exc)})
        return None


def _build_s3_url(key: str) -> str: