From 707e8731720a55e0b3334f546b97df532e7c30ad Mon Sep 17 00:00:00 2001 From: Douglas Rioux Date: Thu, 29 Jun 2023 15:00:09 -0400 Subject: [PATCH 1/7] Make first cut at item model --- .gitignore | 2 + dcicutils/cgap_items.py | 565 +++++++++++++++++++++++++++++++++++++ dcicutils/item_models.py | 211 ++++++++++++++ dcicutils/testing_utils.py | 16 ++ test/test_cgap_items.py | 29 ++ test/test_item_models.py | 256 +++++++++++++++++ 6 files changed, 1079 insertions(+) create mode 100644 dcicutils/cgap_items.py create mode 100644 dcicutils/item_models.py create mode 100644 dcicutils/testing_utils.py create mode 100644 test/test_cgap_items.py create mode 100644 test/test_item_models.py diff --git a/.gitignore b/.gitignore index 9bd970b02..9a07e7fbe 100644 --- a/.gitignore +++ b/.gitignore @@ -117,3 +117,5 @@ ENV/ # PyCharm metadata .idea/ +# Vim +*.swp diff --git a/dcicutils/cgap_items.py b/dcicutils/cgap_items.py new file mode 100644 index 000000000..c168d9a7d --- /dev/null +++ b/dcicutils/cgap_items.py @@ -0,0 +1,565 @@ +from __future__ import annotations +import math +from dataclasses import dataclass +from typing import List, Union + +import structlog + +from .item_models import JsonObject, PortalItem, SubembeddedProperty + +logger = structlog.getLogger(__name__) + + +@dataclass(frozen=True) +class CommonPropertiesMixin: + ACCESSION = "accession" + ALIASES = "aliases" + DATE_MODIFIED = "date_modified" + DISPLAY_TITLE = "display_title" + INSTITUTION = "institution" + LAST_MODIFIED = "last_modified" + MODIFIED_BY = "modified_by" + PROJECT = "project" + SCHEMA_VERSION = "schema_version" + STATUS = "status" + TAGS = "tags" + + @property + def accession(self) -> str: + return self._properties.get(self.ACCESSION, "") + + @property + def display_title(self) -> str: + return self._properties.get(self.DISPLAY_TITLE, "") + + @property + def schema_version(self) -> str: + return self._properties.get(self.SCHEMA_VERSION, "") + + @property + def aliases(self) -> List[str]: + return self._properties.get(self.ALIASES, []) + + @property + def status(self) -> str: + return self._properties.get(self.STATUS, "") + + @property + def date_created(self) -> str: + return self._properties.get(self.DATE_CREATED, "") + + @property + def last_modified(self) -> JsonObject: + return self._properties.get(self.LAST_MODIFIED, {}) + + @property + def date_modified(self) -> str: + return self.last_modified.get(self.DATE_MODIFIED, "") + + @property + def modified_by(self) -> Union[User, str]: + property_value = self.last_modified.get(self.MODIFIED_BY, "") + return self._get_embeddable_property(property_value, User) + + @property + def tags(self) -> List[str]: + return self._properties.get(self.TAGS, []) + + @property + def project(self) -> Union[Project, str]: + property_value = self._properties.get(self.PROJECT, "") + return self._get_embeddable_property(property_value, Project) + + @property + def institution(self) -> Union[Institution, str]: + property_value = self._properties.get(self.INSTITUTION, "") + return self._get_embeddable_property(property_value, Institution) + + +@dataclass(frozen=True) +class CGAPItem(PortalItem, CommonPropertiesMixin): + pass + + +@dataclass(frozen=True) +class Project(CGAPItem): + NAME = "name" + TITLE = "title" + + @property + def name(self) -> str: + return self._properties.get(self.NAME, "") + + @property + def title(self) -> str: + return self._properties.get(self.TITLE, "") + + +@dataclass(frozen=True) +class Institution(CGAPItem): + NAME = "name" + TITLE = "title" + + @property + def name(self) -> str: + return self._properties.get(self.NAME, "") + + @property + def title(self) -> str: + return self._properties.get(self.TITLE, "") + + +@dataclass(frozen=True) +class User(CGAPItem): + FIRST_NAME = "first_name" + INSTITUTION = "institution" + LAST_NAME = "last_name" + PROJECT = "project" + USER_INSTITUTION = "user_institution" + + @property + def first_name(self) -> str: + return self._properties.get(self.FIRST_NAME, "") + + @property + def last_name(self) -> str: + return self._properties.get(self.LAST_NAME, "") + + @property + def institution(self) -> Union[str, JsonObject]: + property_value = self._properties.get(self.INSTITUTION, "") + return self._get_embeddable_property(property_value, Institution) + + @property + def project(self) -> Union[str, JsonObject]: + property_value = self._properties.get(self.PROJECT, "") + return self._get_embeddable_property(property_value, Project) + + @property + def user_institution(self) -> Union[str, JsonObject]: + property_value = self._properties.get(self.USER_INSTITUTION, "") + return self._get_embeddable_property(property_value, Institution) + + +@dataclass(frozen=True) +class FileFormat(CGAPItem): + @property + def file_format(self) -> str: + return self._properties.get(self.FILE_FORMAT, "") + + +@dataclass(frozen=True) +class File(CGAPItem): + @property + def file_format(self): + property_value = self._properties.get(self.FILE_FORMAT, "") + return self._get_embeddable_property(property_value, FileFormat) + + +@dataclass(frozen=True) +class VariantConsequence(CGAPItem): + # Schema constants + IMPACT = "impact" + IMPACT_HIGH = "HIGH" + IMPACT_LOW = "LOW" + IMPACT_MODERATE = "MODERATE" + IMPACT_MODIFIER = "MODIFIER" + VAR_CONSEQ_NAME = "var_conseq_name" + + DOWNSTREAM_GENE_CONSEQUENCE = "downstream_gene_variant" + FIVE_PRIME_UTR_CONSEQUENCE = "5_prime_UTR_variant" + THREE_PRIME_UTR_CONSEQUENCE = "3_prime_UTR_variant" + UPSTREAM_GENE_CONSEQUENCE = "upstream_gene_variant" + + @property + def impact(self) -> str: + return self._properties.get(self.IMPACT, "") + + @property + def name(self) -> str: + return self._properties.get(self.VAR_CONSEQ_NAME, "") + + def get_name(self) -> str: + return self.name + + def get_impact(self) -> str: + return self.impact + + def is_downstream(self) -> str: + return self.name == self.DOWNSTREAM_GENE_CONSEQUENCE + + def is_upstream(self) -> str: + return self.name == self.UPSTREAM_GENE_CONSEQUENCE + + def is_three_prime_utr(self) -> str: + return self.name == self.THREE_PRIME_UTR_CONSEQUENCE + + def is_five_prime_utr(self) -> str: + return self.name == self.FIVE_PRIME_UTR_CONSEQUENCE + + +@dataclass(frozen=True) +class Transcript(SubembeddedProperty): + # Schema constants + CSQ_CANONICAL = "csq_canonical" + CSQ_CONSEQUENCE = "csq_consequence" + CSQ_DISTANCE = "csq_distance" + CSQ_EXON = "csq_exon" + CSQ_FEATURE = "csq_feature" + CSQ_INTRON = "csq_intron" + CSQ_MOST_SEVERE = "csq_most_severe" + + # Class constants + LOCATION_EXON = "Exon" + LOCATION_INTRON = "Intron" + LOCATION_DOWNSTREAM = "bp downstream" + LOCATION_UPSTREAM = "bp upstream" + LOCATION_FIVE_PRIME_UTR = "5' UTR" + LOCATION_THREE_PRIME_UTR = "3' UTR" + IMPACT_RANKING = { + VariantConsequence.IMPACT_HIGH: 0, + VariantConsequence.IMPACT_MODERATE: 1, + VariantConsequence.IMPACT_LOW: 2, + VariantConsequence.IMPACT_MODIFIER: 3, + } + + @property + def canonical(self) -> bool: + return self._properties.get(self.CSQ_CANONICAL, False) + + @property + def most_severe(self) -> bool: + return self._properties.get(self.CSQ_MOST_SEVERE, False) + + @property + def exon(self) -> str: + return self._properties.get(self.CSQ_EXON, "") + + @property + def intron(self) -> str: + return self._properties.get(self.CSQ_INTRON, "") + + @property + def distance(self) -> str: + return self._properties.get(self.CSQ_DISTANCE, "") + + @property + def feature(self) -> str: + return self._properties.get(self.CSQ_FEATURE, "") + + @property + def consequences(self) -> List[Union[VariantConsequence, str]]: + consequences = self._properties.get(self.CSQ_CONSEQUENCE, []) + return [ + self._get_embeddedable_property(consequence, VariantConsequence) + for consequence in consequences + ] + + def _are_consequences_embedded(self) -> bool: + return [ + isinstance(consequence, VariantConsequence) + for consequence in self.consequences + ] + + def is_canonical(self) -> bool: + return self.canonical + + def is_most_severe(self) -> bool: + return self.most_severe + + def get_feature(self) -> str: + return self.feature + + def get_location(self) -> str: + if not self._are_consequences_embedded(): + raise ValueError(f"Consequence calculations not possible") + result = "" + most_severe_consequence = self._get_most_severe_consequence() + if most_severe_consequence: + result = self._get_location(most_severe_consequence) + return result + + def _get_most_severe_consequence(self) -> Union[VariantConsequence, None]: + result = None + most_severe_rank = math.inf + for consequence in self.consequences: + impact = consequence.get_impact() + impact_rank = self.IMPACT_RANKING.get(impact, math.inf) + if impact_rank < most_severe_rank: + result = consequence + return result + + def _get_location(self, most_severe_consequence: VariantConsequence) -> str: + result = "" + if self.exon: + result = self._get_exon_location() + elif self.intron: + result = self._get_intron_location() + elif self.distance: + result = self._get_distance_location(most_severe_consequence) + return self._add_utr_suffix_if_needed(result, most_severe_consequence) + + def _get_exon_location(self) -> str: + return f"{self.LOCATION_EXON} {self.exon}" + + def _get_intron_location(self) -> str: + return f"{self.LOCATION_INTRON} {self.intron}" + + def _get_distance_location(self, consequence: VariantConsequence) -> str: + if consequence.is_upstream(): + return f"{self.distance} {self.LOCATION_UPSTREAM}" + if consequence.is_downstream(): + return f"{self.distance} {self.LOCATION_DOWNSTREAM}" + return "" + + def _add_utr_suffix_if_needed( + self, location: str, consequence: VariantConsequence + ) -> str: + if consequence.is_three_prime_utr(): + return self._add_three_prime_utr_suffix(location) + if consequence.is_five_prime_utr(): + return self._add_five_prime_utr_suffix(location) + return location + + def _add_three_prime_utr_suffix(self, location: str) -> str: + return self._add_utr_suffix(location, self.LOCATION_THREE_PRIME_UTR) + + def _add_five_prime_utr_suffix(self, location: str) -> str: + return self._add_utr_suffix(location, self.LOCATION_FIVE_PRIME_UTR) + + def _add_utr_suffix(self, location: str, utr_suffix: str) -> str: + if location: + return f"{location} ({utr_suffix})" + return utr_suffix + + def get_consequence_names(self) -> str: + if not self._are_consequences_embedded(): + raise ValueError(f"Consequence calculations not possible") + return ", ".join([consequence.get_name() for consequence in self.consequences]) + + +@dataclass(frozen=True) +class Variant(CGAPItem): + # Schema constants + CSQ_CANONICAL = "csq_canonical" + CSQ_CONSEQUENCE = "csq_consquence" + CSQ_FEATURE = "csq_feature" + CSQ_GNOMADE2_AF_POPMAX = "csq_gnomade2_af_popmax" + CSQ_GNOMADG_AF_POPMAX = "csq_gnomadg_af_popmax" + CSQ_MOST_SEVERE = "csq_most_severe" + DISTANCE = "distance" + EXON = "exon" + INTRON = "intron" + MOST_SEVERE_LOCATION = "most_severe_location" + TRANSCRIPT = "transcript" + + GNOMAD_V2_AF_PREFIX = "csq_gnomade2_af-" + GNOMAD_V3_AF_PREFIX = "csq_gnomadg_af-" + GNOMAD_POPULATION_SUFFIX_TO_NAME = { + "afr": "African-American/African", + "ami": "Amish", + "amr": "Latino", + "asj": "Ashkenazi Jewish", + "eas": "East Asian", + "fin": "Finnish", + "mid": "Middle Eastern", + "nfe": "Non-Finnish European", + "oth": "Other Ancestry", + "sas": "South Asian", + } + + @property + def transcript(self) -> List[JsonObject]: + return self._properties.get(self.TRANSCRIPT, []) + + @property + def _transcripts(self) -> List[Transcript]: + return [ + Transcript.from_properties(transcript, self) + for transcript in self.transcript + ] + + @property + def most_severe_location(self) -> str: + return self._properties.get(self.MOST_SEVERE_LOCATION, "") + + @property + def csq_gnomadg_af_popmax(self) -> Union[float, None]: + return self._properties.get(self.CSQ_GNOMADG_AF_POPMAX) + + @property + def csq_gnomade2_af_popmax(self) -> Union[float, None]: + return self._properties.get(self.CSQ_GNOMADE2_AF_POPMAX) + + @property + def _canonical_transcript(self) -> Union[None, Transcript]: + for transcript in self._transcripts: + if transcript.is_canonical(): + return transcript + + @property + def _most_severe_transcript(self) -> Union[None, Transcript]: + for transcript in self._transcripts: + if transcript.is_most_severe(): + return transcript + + def get_most_severe_location(self) -> str: + return self.most_severe_location + + def get_canonical_transcript_feature(self) -> str: + if self._canonical_transcript: + return self._canonical_transcript.get_feature() + return "" + + def get_most_severe_transcript_feature(self) -> str: + if self._most_severe_transcript: + return self._most_severe_transcript.get_feature() + return "" + + def get_canonical_transcript_consequence_names(self) -> str: + if self._canonical_transcript: + return self._canonical_transcript.get_consequence_names() + return "" + + def get_most_severe_transcript_consequence_names(self) -> str: + if self._most_severe_transcript: + return self._most_severe_transcript.get_consequence_names() + return "" + + def get_canonical_transcript_location(self) -> str: + if self._canonical_transcript: + return self._canonical_transcript.get_location() + return "" + + def get_most_severe_transcript_location(self) -> str: + if self._most_severe_transcript: + return self._most_severe_transcript.get_location() + return "" + + def get_gnomad_v3_popmax_population(self) -> str: + result = "" + gnomad_v3_af_popmax = self.csq_gnomadg_af_popmax + if gnomad_v3_af_popmax: + result = self._get_gnomad_v3_population_for_allele_fraction( + gnomad_v3_af_popmax + ) + return result + + def get_gnomad_v2_popmax_population(self) -> str: + result = "" + gnomad_v2_af_popmax = self.csq_gnomade2_af_popmax + if gnomad_v2_af_popmax: + result = self._get_gnomad_v2_population_for_allele_fraction( + gnomad_v2_af_popmax + ) + return result + + def _get_gnomad_v3_population_for_allele_fraction( + self, allele_fraction: float + ) -> str: + return self._get_gnomad_population_for_allele_fraction( + self.GNOMAD_V3_AF_PREFIX, allele_fraction + ) + + def _get_gnomad_v2_population_for_allele_fraction( + self, allele_fraction: float + ) -> str: + return self._get_gnomad_population_for_allele_fraction( + self.GNOMAD_V2_AF_PREFIX, allele_fraction + ) + + def _get_gnomad_population_for_allele_fraction( + self, gnomad_af_prefix: str, allele_fraction: float + ) -> str: + result = "" + for ( + gnomad_suffix, + population_name, + ) in self.GNOMAD_POPULATION_SUFFIX_TO_NAME.items(): + population_property_name = gnomad_af_prefix + gnomad_suffix + allele_frequency = self._properties.get(population_property_name) + if allele_frequency == allele_fraction: + result = population_name + break + return result + + +@dataclass(frozen=True) +class Note(CGAPItem): + pass + + +@dataclass(frozen=True) +class VariantSample(CGAPItem): + # Schema constants + VARIANT = "variant" + + @property + def variant(self) -> Variant: + return Variant(self._properties.get(self.VARIANT, {})) + + def get_canonical_transcript_feature(self) -> str: + return self.variant.get_canonical_transcript_feature() + + def get_canonical_transcript_location(self) -> str: + return self.variant.get_canonical_transcript_location() + + def get_canonical_transcript_consequence_names(self) -> str: + return self.variant.get_canonical_transcript_consequence_names() + + def get_most_severe_transcript_feature(self) -> str: + return self.variant.get_most_severe_transcript_feature() + + def get_most_severe_transcript_location(self) -> str: + return self.variant.get_most_severe_transcript_location() + + def get_most_severe_transcript_consequence_names(self) -> str: + return self.variant.get_most_severe_transcript_consequence_names() + + def get_gnomad_v3_popmax_population(self) -> str: + return self.variant.get_gnomad_v3_popmax_population() + + def get_gnomad_v2_popmax_population(self) -> str: + return self.variant.get_gnomad_v2_popmax_population() + + +@dataclass(frozen=True) +class VariantSampleSelectionFromVariantSampleList(SubembeddedProperty): + VARIANT_SAMPLE_ITEM = "variant_sample_item" + + @property + def variant_sample_item(self) -> Union[VariantSample, str]: + variant_sample = self._properties.get(self.VARIANT_SAMPLE_ITEM, "") + return self._get_embeddable_property(variant_sample, VariantSample) + + def get_variant_sample(self): + return self.variant_sample_item + + +@dataclass(frozen=True) +class VariantSampleList(CGAPItem): + CREATED_FOR_CASE = "created_for_case" + VARIANT_SAMPLES = "variant_samples" + + @property + def created_for_case(self) -> str: + return self._properties.get(self.CREATED_FOR_CASE, "") + + @property + def variant_samples(self) -> List[VariantSampleSelectionFromVariantSampleList]: + variant_samples = self._properties.get(self.VARIANT_SAMPLES, []) + return [ + VariantSampleSelectionFromVariantSampleList.from_properties( + variant_sample, parent_item=self + ) + for variant_sample in variant_samples + ] + + def get_associated_case_accession(self) -> str: + return self.created_for_case + + def get_variant_samples(self) -> List[Union[VariantSample, str]]: + return [ + variant_sample_selection.get_variant_sample() + for variant_sample_selection in self.variant_samples + ] diff --git a/dcicutils/item_models.py b/dcicutils/item_models.py new file mode 100644 index 000000000..7c8ac7f6e --- /dev/null +++ b/dcicutils/item_models.py @@ -0,0 +1,211 @@ +from __future__ import annotations +from dataclasses import dataclass, field +from functools import lru_cache +from typing import Any, Mapping, Optional, Tuple, Union + +import structlog + +from .ff_utils import get_metadata + + +logger = structlog.getLogger(__name__) + +JsonObject = Mapping[str, Any] + + +def _make_embeddable_property( + existing_item: PortalItem, + property_value: [str, JsonObject], + item_to_create: PortalItem, +) -> Union[str, JsonObject, PortalItem]: + embed_items = existing_item.embed_items() + identifier = _get_item_identifier(property_value) + if embed_items and identifier: + return item_to_create.from_identifier_and_existing_item( + identifier, existing_item + ) + if isinstance(property_value, Mapping): + return item_to_create.from_properties_and_existing_item( + property_value, existing_item + ) + return property_value + + +def _get_item_identifier(item: Union[str, JsonObject]) -> str: + if isinstance(item, str): + return item + if isinstance(item, Mapping): + return item.get(PortalItem.UUID, "") + raise ValueError() + + +@dataclass(frozen=True) +class PortalItem: + AT_ID = "@id" + UUID = "uuid" + + IDENTIFYING_PROPERTIES = [AT_ID, UUID] + + properties: Optional[JsonObject] = field(default=None, hash=False) + auth: Optional[JsonObject] = field(default=None, hash=False) + do_embeds: Optional[bool] = field(default=False, hash=False) + + def __repr__(self) -> str: + return f"{self.__class__.__name__}({self.uuid})" + + def __eq__(self, other: Any) -> bool: + if isinstance(other, PortalItem) and self._uuid: + return self._uuid == other._uuid + return False + + @property + def _properties(self) -> JsonObject: + return self.properties or {} + + @property + def _uuid(self) -> str: + return self._properties.get(self.UUID, "") + + @property + def _at_id(self) -> str: + return self._properties.get(self.AT_ID, "") + + def is_same_item(self, item: Any) -> bool: + if isinstance(item, PortalItem): + return self == item + if isinstance(item, dict): + return self._has_same_identifier(item) + if isinstance(item, str): + return self._is_same_identifier(item) + return False + + def _has_same_identifier(self, item: Mapping) -> bool: + return any( + [ + self._properties.get(identifying_property) == item.get(identifying_property) + for identifying_property in self.IDENTIFYING_PROPERTIES + ] + ) + + def _is_same_identifier(self, identifier: str) -> bool: + return any( + [ + self._properties.get(identifying_property) == identifier + for identifying_property in self.IDENTIFYING_PROPERTIES + ] + ) + + def do_embeds(self) -> bool: + return self.do_embeds + + def get_auth(self) -> Union[JsonObject, None]: + return self.auth + + def get_properties(self) -> JsonObject: + return self._properties + + def _get_embeddable_property( + self, + property_value: [str, JsonObject], + item_to_create: PortalItem, + ) -> Union[str, JsonObject, PortalItem]: + return _make_embeddable_property(self, property_value, item_to_create) + + @classmethod + def from_properties( + cls, properties: JsonObject, embed_items=False, auth=None, **kwargs: Any + ) -> PortalItem: + return cls(properties=properties, embed_items=embed_items, auth=auth) + + @classmethod + def from_identifier_and_auth( + cls, identifier: str, auth: JsonObject, embed_items=False, **kwargs: Any + ) -> PortalItem: + properties = cls._get_item_via_auth(identifier, auth) + return cls.from_properties(properties=properties, auth=auth, embed_items=embed_items) + + @classmethod + def _get_item_via_auth( + cls, identifier: str, auth: JsonObject, add_on: Optional[str] = "frame=object" + ) -> JsonObject: + hashable_auth = cls._make_hashable_auth(auth) + return cls._get_and_cache_item_via_auth(identifier, hashable_auth, add_on) + + @classmethod + def _make_hashable_auth(cls, auth: Mapping[str, str]) -> Tuple[Tuple[str, str]]: + """Assuming nothing nested here...""" + return tuple(auth.items()) + + @classmethod + def _undo_make_hashable_auth( + cls, hashable_auth: Tuple[Tuple[str, str]] + ) -> JsonObject: + return dict(hashable_auth) + + @classmethod + @lru_cache(maxsize=256) + def _get_and_cache_item_via_auth( + cls, identifier: str, hashable_auth: Tuple[Tuple[str, Any]], add_on: Optional[str] = None + ) -> JsonObject: + """Save on requests by caching items.""" + auth = cls._undo_make_hashable_auth(hashable_auth) + try: + result = get_metadata(identifier, key=auth, add_on=add_on) + except Exception as e: + result = {} + logger.error(f"Error getting metadata for {identifier}: {e}") + return result + + @classmethod + def from_identifier_and_existing_item( + cls, identifier: str, existing_item: PortalItem, **kwargs: Any + ) -> PortalItem: + embed_items = existing_item.embed_items() + auth = existing_item.get_auth() + if auth: + return cls.from_identifier_and_auth( + identifier, auth, embed_items=embed_items + ) + raise ValueError("Unable to create item from existing item") + + @classmethod + def from_properties_and_existing_item( + cls, properties: JsonObject, existing_item: PortalItem, **kwargs: Any + ) -> PortalItem: + embed_items = existing_item.embed_items() + auth = existing_item.get_auth() + return cls.from_properties( + properties, embed_items=embed_items, auth=auth + ) + + +@dataclass(frozen=True) +class SubembeddedProperty: + properties: Optional[JsonObject] = field(default=None, hash=False) + parent_item: Optional[PortalItem] = field(default=None, hash=False) + + def __repr__(self) -> str: + return f"{self.__class__.__name__}(parent={self.parent_item.__repr__()})" + + @property + def _properties(self) -> JsonObject: + return self.properties or {} + + def _get_embeddable_property( + self, + property_value: [str, JsonObject], + item_to_create: PortalItem, + ) -> Union[str, JsonObject, PortalItem]: + if self.parent_item: + return _make_embeddable_property( + self.parent_item, property_value, item_to_create + ) + if isinstance(property_value, Mapping): + return item_to_create.from_properties(property_value) + return property_value + + @classmethod + def from_properties( + cls, properties: JsonObject, parent_item: Optional[PortalItem], **kwargs: Any + ) -> SubembeddedProperty: + return cls(properties=properties, parent_item=parent_item) diff --git a/dcicutils/testing_utils.py b/dcicutils/testing_utils.py new file mode 100644 index 000000000..fb8e25e28 --- /dev/null +++ b/dcicutils/testing_utils.py @@ -0,0 +1,16 @@ +from contextlib import contextmanager +from typing import Any, Iterator, Optional +from unittest import mock + + +@contextmanager +def patch_context( + object_to_patch: object, + attribute_to_patch: str, + return_value: Optional[Any] = None, + **kwargs, +) -> Iterator[mock.MagicMock]: + with mock.patch.object(object_to_patch, attribute_to_patch, **kwargs) as mocked_item: + if return_value is not None: + mocked_item.return_value = return_value + yield mocked_item diff --git a/test/test_cgap_items.py b/test/test_cgap_items.py new file mode 100644 index 000000000..655df26c1 --- /dev/null +++ b/test/test_cgap_items.py @@ -0,0 +1,29 @@ +import json +from pathlib import Path + +from dcicutils.cgap_items import User + + +SOME_USER = { + "uuid": "some-uuid", + "email": "some-email", + "first_name": "some-first-name", + "last_name": "some-last-name", + "title": "some-title", + "project": "some-project", +} +KEYS_FILE = Path.expanduser(Path("~/.cgap-keys.json")).absolute() +USER_IDENTIFIER = "5196db84-f3d9-44bd-bba0-59e9e83634a1" + + +def get_keys(): + keys = json.loads(KEYS_FILE.read_text()) + return keys["msa"] + + +def test_user(): + auth = get_keys() + user = User.from_identifier_and_auth(USER_IDENTIFIER, auth, embed_items=True) + import pdb; pdb.set_trace() + project = user.project + assert user.uuid == USER_IDENTIFIER diff --git a/test/test_item_models.py b/test/test_item_models.py new file mode 100644 index 000000000..d183c22d0 --- /dev/null +++ b/test/test_item_models.py @@ -0,0 +1,256 @@ +from contextlib import contextmanager +from random import random +from typing import Any, Iterator, Optional, Tuple, Union +from unittest import mock + +import pytest + +from dcicutils import item_models as item_models_module +from dcicutils.item_models import ( + JsonObject, + _make_embeddable_property, + _get_item_identifier, + PortalItem, + SubembeddedProperty, +) +from dcicutils.testing_utils import patch_context + + +SOME_UUID = "uuid1234" +SOME_AT_ID = "/foo/bar/" +SOME_ITEM_PROPERTIES = {"uuid": SOME_UUID, "@id": SOME_AT_ID} +OTHER_ITEM_PROPERTIES = {"uuid": "foo"} +SOME_AUTH = {"key": "some_key", "secret": "some_secret"} +HASHABLE_SOME_AUTH = (("key", "some_key"), ("secret", "some_secret")) + + +@contextmanager +def patch_make_embeddable_property(**kwargs: Any) -> Iterator[mock.MagicMock]: + with patch_context( + item_models_module, "_make_embeddable_property", **kwargs + ) as mock_item: + yield mock_item + + +@contextmanager +def patch_get_and_cache_item_via_auth(**kwargs: Any) -> Iterator[mock.MagicMock]: + with patch_context( + item_models_module.PortalItem, + "_get_and_cache_item_via_auth", + **kwargs, + ) as mock_item: + yield mock_item + + +@contextmanager +def patch_get_metadata(**kwargs: Any) -> Iterator[mock.MagicMock]: + with patch_context( + item_models_module, + "get_metadata", + **kwargs, + ) as mock_item: + yield mock_item + + +def mock_portal_item(embed_items: bool = False): + mock_item = mock.create_autospec(PortalItem, instance=True) + mock_item.embed_items.return_value = embed_items + return mock_item + + +@pytest.mark.parametrize( + "embed_items,property_value,expected_from_identifier,expected_from_properties", + [ + (False, "foo", False, False), + (True, "foo", True, False), + (False, {}, False, True), + (False, SOME_ITEM_PROPERTIES, False, True), + (True, SOME_ITEM_PROPERTIES, True, False), + ] +) +def test_make_embeddable_property( + embed_items: bool, property_value: Union[str, JsonObject], + expected_from_identifier: bool, + expected_from_properties: bool, +) -> None: + existing_item = mock_portal_item(embed_items) + item_to_create = mock_portal_item() + result = _make_embeddable_property(existing_item, property_value, item_to_create) + if expected_from_identifier: + identifier = _get_item_identifier(property_value) + assert result == item_to_create.from_identifier_and_existing_item.return_value + item_to_create.from_identifier_and_existing_item.assert_called_once_with( + identifier, existing_item + ) + elif expected_from_properties: + assert result == item_to_create.from_properties_and_existing_item.return_value + item_to_create.from_properties_and_existing_item.assert_called_once_with( + property_value, existing_item + ) + else: + assert result == property_value + + +@pytest.mark.parametrize( + "item,expected", + [ + ("", ""), + ({}, ""), + ("foo", "foo"), + ({"uuid": "foo"}, "foo"), + ] +) +def test_get_item_identifier(item: Union[str, JsonObject], expected: str) -> None: + result = _get_item_identifier(item) + assert result == expected + + +def get_portal_item( + properties: Optional[JsonObject] = None, + embed_items: Optional[bool] = False, + auth: Optional[bool] = None, +) -> PortalItem: + if properties is None: + properties = SOME_ITEM_PROPERTIES + if auth is None: + auth = SOME_AUTH + return PortalItem(auth=auth, embed_items=embed_items, properties=properties) + + +class TestPortalItem: + + @pytest.mark.parametrize( + "item_1_properties,item_2_properties,expected", + [ + ({}, {}, False), + ({}, SOME_ITEM_PROPERTIES, False), + (SOME_ITEM_PROPERTIES, OTHER_ITEM_PROPERTIES, False), + (SOME_ITEM_PROPERTIES, SOME_ITEM_PROPERTIES, True), + ] + ) + def test_equality( + self, + item_1_properties: JsonObject, item_2_properties: JsonObject, + expected: bool, + ) -> None: + item_1 = get_portal_item(properties=item_1_properties) + item_2 = get_portal_item(properties=item_2_properties) + assert (item_1 == item_2) == expected + + def test_properties(self) -> None: + assert PortalItem()._properties == {} + assert get_portal_item()._properties == SOME_ITEM_PROPERTIES + + @pytest.mark.parametrize( + "properties,expected", [({}, ""), (SOME_ITEM_PROPERTIES, SOME_UUID)] + ) + def test_uuid(self, properties: JsonObject, expected: str) -> None: + portal_item = get_portal_item(properties=properties) + assert portal_item.uuid == expected + + @pytest.mark.parametrize( + "properties,expected", [({}, ""), (SOME_ITEM_PROPERTIES, SOME_AT_ID)] + ) + def test_at_id(self, properties: JsonObject, expected: str) -> None: + portal_item = get_portal_item(properties=properties) + assert portal_item.at_id == expected + + @pytest.mark.parametrize( + "comparison_item,expected", + [ + (get_portal_item(), True), + (get_portal_item(properties=OTHER_ITEM_PROPERTIES), False), + (SOME_UUID, True), + (SOME_AT_ID, True), + ("some_string", False), + ] + ) + def test_is_same_item(self, comparison_item: Any, expected: bool) -> None: + portal_item = get_portal_item() + result = portal_item.is_same_item(comparison_item) + assert result == expected + + def test_get_embeddable_property(self) -> None: + property_value = "foo" + item_to_create = mock_portal_item() + portal_item = get_portal_item() + with patch_make_embeddable_property() as mock_make_embed: + portal_item._get_embeddable_property(property_value, item_to_create) + assert mock_make_embed.called_once_with( + portal_item, property_value, item_to_create + ) + + def test_get_item_via_auth(self) -> None: + with patch_get_and_cache_item_via_auth( + return_value=SOME_ITEM_PROPERTIES + ) as mock_get_and_cache_item: + portal_item = get_portal_item() + result = portal_item._get_item_via_auth(SOME_UUID, SOME_AUTH) + assert result == SOME_ITEM_PROPERTIES + mock_get_and_cache_item.assert_called_once_with( + SOME_UUID, HASHABLE_SOME_AUTH, "frame=object" + ) + + @pytest.mark.parametrize( + "auth,expected", [({}, tuple()), (SOME_AUTH, HASHABLE_SOME_AUTH)] + ) + def test_make_hashable_auth(self, auth: JsonObject, expected: Tuple) -> None: + portal_item = get_portal_item() + result = portal_item._make_hashable_auth(auth) + assert result == expected + + @pytest.mark.parametrize( + "hashable_auth,expected", [(tuple(), {}), (HASHABLE_SOME_AUTH, SOME_AUTH)] + ) + def test_undo_make_hashable_auth(self, hashable_auth: Tuple, expected: JsonObject) -> None: + portal_item = get_portal_item() + result = portal_item._undo_make_hashable_auth(hashable_auth) + assert result == expected + + @pytest.mark.parametrize( + "raise_exception,expected", [(False, SOME_ITEM_PROPERTIES), (True, {})] + ) + def test_get_ad_cache_item_via_auth(self, raise_exception: bool, expected: JsonObject) -> None: + side_effect = Exception if raise_exception else None + random_add_on = str(random()) # To differentiate parametrized calls + with patch_get_metadata(side_effect=side_effect, return_value=SOME_ITEM_PROPERTIES) as mock_get_metadata: + portal_item = get_portal_item() + result = portal_item._get_and_cache_item_via_auth( + SOME_UUID, HASHABLE_SOME_AUTH, add_on=random_add_on + ) + assert result == expected + assert len(mock_get_metadata.call_args_list) == 1 + mock_get_metadata.assert_called_once_with(SOME_UUID, key=SOME_AUTH, add_on=random_add_on) + + # Ensure cached + portal_item._get_and_cache_item_via_auth(SOME_UUID, HASHABLE_SOME_AUTH, add_on=random_add_on) + assert len(mock_get_metadata.call_args_list) == 1 + + +def get_subembedded_property(properties: Optional[JsonObject] = None, parent_item: Optional[PortalItem] = None) -> SubembeddedProperty: + properties = properties or SOME_ITEM_PROPERTIES + return SubembeddedProperty(properties=properties, parent_item=parent_item) + + +class TestSubembeddedProperty: + + @pytest.mark.parametrize( + ( + "property_value,parent_item_exists,expected_make_embeddable_property_call," + "expected" + ), + [ + ] + ) + def test_get_embeddable_property(self, property_value: Union[JsonObject, str], parent_item_exists: bool, expected_make_embeddable_property_call: bool, expected: Union[PortalItem, str]) -> None: + parent_item = mock_portal_item() if parent_item_exists else None + item_to_create = mock_portal_item() + with patch_make_embeddable_property() as mock_make_embeddable_property: + subembedded_property = get_subembedded_property(parent_item=parent_item) + result = subembedded_property._get_embeddable_property(property_value, item_to_create) + if expected_make_embeddable_property_call: + mock_make_embeddable_property.assert_called_once_with(parent_item, property_value, item_to_create) + assert result == mock_make_embeddable_property.return_value + else: + assert result == expected + mock_make_embeddable_property.assert_not_called() From 2afdb25779ad4909a3c01887c694beb665ea1555 Mon Sep 17 00:00:00 2001 From: Douglas Rioux Date: Mon, 17 Jul 2023 14:34:04 -0400 Subject: [PATCH 2/7] Tidy up + increase test coverage --- dcicutils/cgap_items.py | 565 ---------------------------------- dcicutils/item_model_utils.py | 222 +++++++++++++ dcicutils/item_models.py | 211 ------------- dcicutils/testing_utils.py | 34 +- test/test_cgap_items.py | 29 -- test/test_item_model_utils.py | 388 +++++++++++++++++++++++ test/test_item_models.py | 256 --------------- 7 files changed, 638 insertions(+), 1067 deletions(-) delete mode 100644 dcicutils/cgap_items.py create mode 100644 dcicutils/item_model_utils.py delete mode 100644 dcicutils/item_models.py delete mode 100644 test/test_cgap_items.py create mode 100644 test/test_item_model_utils.py delete mode 100644 test/test_item_models.py diff --git a/dcicutils/cgap_items.py b/dcicutils/cgap_items.py deleted file mode 100644 index c168d9a7d..000000000 --- a/dcicutils/cgap_items.py +++ /dev/null @@ -1,565 +0,0 @@ -from __future__ import annotations -import math -from dataclasses import dataclass -from typing import List, Union - -import structlog - -from .item_models import JsonObject, PortalItem, SubembeddedProperty - -logger = structlog.getLogger(__name__) - - -@dataclass(frozen=True) -class CommonPropertiesMixin: - ACCESSION = "accession" - ALIASES = "aliases" - DATE_MODIFIED = "date_modified" - DISPLAY_TITLE = "display_title" - INSTITUTION = "institution" - LAST_MODIFIED = "last_modified" - MODIFIED_BY = "modified_by" - PROJECT = "project" - SCHEMA_VERSION = "schema_version" - STATUS = "status" - TAGS = "tags" - - @property - def accession(self) -> str: - return self._properties.get(self.ACCESSION, "") - - @property - def display_title(self) -> str: - return self._properties.get(self.DISPLAY_TITLE, "") - - @property - def schema_version(self) -> str: - return self._properties.get(self.SCHEMA_VERSION, "") - - @property - def aliases(self) -> List[str]: - return self._properties.get(self.ALIASES, []) - - @property - def status(self) -> str: - return self._properties.get(self.STATUS, "") - - @property - def date_created(self) -> str: - return self._properties.get(self.DATE_CREATED, "") - - @property - def last_modified(self) -> JsonObject: - return self._properties.get(self.LAST_MODIFIED, {}) - - @property - def date_modified(self) -> str: - return self.last_modified.get(self.DATE_MODIFIED, "") - - @property - def modified_by(self) -> Union[User, str]: - property_value = self.last_modified.get(self.MODIFIED_BY, "") - return self._get_embeddable_property(property_value, User) - - @property - def tags(self) -> List[str]: - return self._properties.get(self.TAGS, []) - - @property - def project(self) -> Union[Project, str]: - property_value = self._properties.get(self.PROJECT, "") - return self._get_embeddable_property(property_value, Project) - - @property - def institution(self) -> Union[Institution, str]: - property_value = self._properties.get(self.INSTITUTION, "") - return self._get_embeddable_property(property_value, Institution) - - -@dataclass(frozen=True) -class CGAPItem(PortalItem, CommonPropertiesMixin): - pass - - -@dataclass(frozen=True) -class Project(CGAPItem): - NAME = "name" - TITLE = "title" - - @property - def name(self) -> str: - return self._properties.get(self.NAME, "") - - @property - def title(self) -> str: - return self._properties.get(self.TITLE, "") - - -@dataclass(frozen=True) -class Institution(CGAPItem): - NAME = "name" - TITLE = "title" - - @property - def name(self) -> str: - return self._properties.get(self.NAME, "") - - @property - def title(self) -> str: - return self._properties.get(self.TITLE, "") - - -@dataclass(frozen=True) -class User(CGAPItem): - FIRST_NAME = "first_name" - INSTITUTION = "institution" - LAST_NAME = "last_name" - PROJECT = "project" - USER_INSTITUTION = "user_institution" - - @property - def first_name(self) -> str: - return self._properties.get(self.FIRST_NAME, "") - - @property - def last_name(self) -> str: - return self._properties.get(self.LAST_NAME, "") - - @property - def institution(self) -> Union[str, JsonObject]: - property_value = self._properties.get(self.INSTITUTION, "") - return self._get_embeddable_property(property_value, Institution) - - @property - def project(self) -> Union[str, JsonObject]: - property_value = self._properties.get(self.PROJECT, "") - return self._get_embeddable_property(property_value, Project) - - @property - def user_institution(self) -> Union[str, JsonObject]: - property_value = self._properties.get(self.USER_INSTITUTION, "") - return self._get_embeddable_property(property_value, Institution) - - -@dataclass(frozen=True) -class FileFormat(CGAPItem): - @property - def file_format(self) -> str: - return self._properties.get(self.FILE_FORMAT, "") - - -@dataclass(frozen=True) -class File(CGAPItem): - @property - def file_format(self): - property_value = self._properties.get(self.FILE_FORMAT, "") - return self._get_embeddable_property(property_value, FileFormat) - - -@dataclass(frozen=True) -class VariantConsequence(CGAPItem): - # Schema constants - IMPACT = "impact" - IMPACT_HIGH = "HIGH" - IMPACT_LOW = "LOW" - IMPACT_MODERATE = "MODERATE" - IMPACT_MODIFIER = "MODIFIER" - VAR_CONSEQ_NAME = "var_conseq_name" - - DOWNSTREAM_GENE_CONSEQUENCE = "downstream_gene_variant" - FIVE_PRIME_UTR_CONSEQUENCE = "5_prime_UTR_variant" - THREE_PRIME_UTR_CONSEQUENCE = "3_prime_UTR_variant" - UPSTREAM_GENE_CONSEQUENCE = "upstream_gene_variant" - - @property - def impact(self) -> str: - return self._properties.get(self.IMPACT, "") - - @property - def name(self) -> str: - return self._properties.get(self.VAR_CONSEQ_NAME, "") - - def get_name(self) -> str: - return self.name - - def get_impact(self) -> str: - return self.impact - - def is_downstream(self) -> str: - return self.name == self.DOWNSTREAM_GENE_CONSEQUENCE - - def is_upstream(self) -> str: - return self.name == self.UPSTREAM_GENE_CONSEQUENCE - - def is_three_prime_utr(self) -> str: - return self.name == self.THREE_PRIME_UTR_CONSEQUENCE - - def is_five_prime_utr(self) -> str: - return self.name == self.FIVE_PRIME_UTR_CONSEQUENCE - - -@dataclass(frozen=True) -class Transcript(SubembeddedProperty): - # Schema constants - CSQ_CANONICAL = "csq_canonical" - CSQ_CONSEQUENCE = "csq_consequence" - CSQ_DISTANCE = "csq_distance" - CSQ_EXON = "csq_exon" - CSQ_FEATURE = "csq_feature" - CSQ_INTRON = "csq_intron" - CSQ_MOST_SEVERE = "csq_most_severe" - - # Class constants - LOCATION_EXON = "Exon" - LOCATION_INTRON = "Intron" - LOCATION_DOWNSTREAM = "bp downstream" - LOCATION_UPSTREAM = "bp upstream" - LOCATION_FIVE_PRIME_UTR = "5' UTR" - LOCATION_THREE_PRIME_UTR = "3' UTR" - IMPACT_RANKING = { - VariantConsequence.IMPACT_HIGH: 0, - VariantConsequence.IMPACT_MODERATE: 1, - VariantConsequence.IMPACT_LOW: 2, - VariantConsequence.IMPACT_MODIFIER: 3, - } - - @property - def canonical(self) -> bool: - return self._properties.get(self.CSQ_CANONICAL, False) - - @property - def most_severe(self) -> bool: - return self._properties.get(self.CSQ_MOST_SEVERE, False) - - @property - def exon(self) -> str: - return self._properties.get(self.CSQ_EXON, "") - - @property - def intron(self) -> str: - return self._properties.get(self.CSQ_INTRON, "") - - @property - def distance(self) -> str: - return self._properties.get(self.CSQ_DISTANCE, "") - - @property - def feature(self) -> str: - return self._properties.get(self.CSQ_FEATURE, "") - - @property - def consequences(self) -> List[Union[VariantConsequence, str]]: - consequences = self._properties.get(self.CSQ_CONSEQUENCE, []) - return [ - self._get_embeddedable_property(consequence, VariantConsequence) - for consequence in consequences - ] - - def _are_consequences_embedded(self) -> bool: - return [ - isinstance(consequence, VariantConsequence) - for consequence in self.consequences - ] - - def is_canonical(self) -> bool: - return self.canonical - - def is_most_severe(self) -> bool: - return self.most_severe - - def get_feature(self) -> str: - return self.feature - - def get_location(self) -> str: - if not self._are_consequences_embedded(): - raise ValueError(f"Consequence calculations not possible") - result = "" - most_severe_consequence = self._get_most_severe_consequence() - if most_severe_consequence: - result = self._get_location(most_severe_consequence) - return result - - def _get_most_severe_consequence(self) -> Union[VariantConsequence, None]: - result = None - most_severe_rank = math.inf - for consequence in self.consequences: - impact = consequence.get_impact() - impact_rank = self.IMPACT_RANKING.get(impact, math.inf) - if impact_rank < most_severe_rank: - result = consequence - return result - - def _get_location(self, most_severe_consequence: VariantConsequence) -> str: - result = "" - if self.exon: - result = self._get_exon_location() - elif self.intron: - result = self._get_intron_location() - elif self.distance: - result = self._get_distance_location(most_severe_consequence) - return self._add_utr_suffix_if_needed(result, most_severe_consequence) - - def _get_exon_location(self) -> str: - return f"{self.LOCATION_EXON} {self.exon}" - - def _get_intron_location(self) -> str: - return f"{self.LOCATION_INTRON} {self.intron}" - - def _get_distance_location(self, consequence: VariantConsequence) -> str: - if consequence.is_upstream(): - return f"{self.distance} {self.LOCATION_UPSTREAM}" - if consequence.is_downstream(): - return f"{self.distance} {self.LOCATION_DOWNSTREAM}" - return "" - - def _add_utr_suffix_if_needed( - self, location: str, consequence: VariantConsequence - ) -> str: - if consequence.is_three_prime_utr(): - return self._add_three_prime_utr_suffix(location) - if consequence.is_five_prime_utr(): - return self._add_five_prime_utr_suffix(location) - return location - - def _add_three_prime_utr_suffix(self, location: str) -> str: - return self._add_utr_suffix(location, self.LOCATION_THREE_PRIME_UTR) - - def _add_five_prime_utr_suffix(self, location: str) -> str: - return self._add_utr_suffix(location, self.LOCATION_FIVE_PRIME_UTR) - - def _add_utr_suffix(self, location: str, utr_suffix: str) -> str: - if location: - return f"{location} ({utr_suffix})" - return utr_suffix - - def get_consequence_names(self) -> str: - if not self._are_consequences_embedded(): - raise ValueError(f"Consequence calculations not possible") - return ", ".join([consequence.get_name() for consequence in self.consequences]) - - -@dataclass(frozen=True) -class Variant(CGAPItem): - # Schema constants - CSQ_CANONICAL = "csq_canonical" - CSQ_CONSEQUENCE = "csq_consquence" - CSQ_FEATURE = "csq_feature" - CSQ_GNOMADE2_AF_POPMAX = "csq_gnomade2_af_popmax" - CSQ_GNOMADG_AF_POPMAX = "csq_gnomadg_af_popmax" - CSQ_MOST_SEVERE = "csq_most_severe" - DISTANCE = "distance" - EXON = "exon" - INTRON = "intron" - MOST_SEVERE_LOCATION = "most_severe_location" - TRANSCRIPT = "transcript" - - GNOMAD_V2_AF_PREFIX = "csq_gnomade2_af-" - GNOMAD_V3_AF_PREFIX = "csq_gnomadg_af-" - GNOMAD_POPULATION_SUFFIX_TO_NAME = { - "afr": "African-American/African", - "ami": "Amish", - "amr": "Latino", - "asj": "Ashkenazi Jewish", - "eas": "East Asian", - "fin": "Finnish", - "mid": "Middle Eastern", - "nfe": "Non-Finnish European", - "oth": "Other Ancestry", - "sas": "South Asian", - } - - @property - def transcript(self) -> List[JsonObject]: - return self._properties.get(self.TRANSCRIPT, []) - - @property - def _transcripts(self) -> List[Transcript]: - return [ - Transcript.from_properties(transcript, self) - for transcript in self.transcript - ] - - @property - def most_severe_location(self) -> str: - return self._properties.get(self.MOST_SEVERE_LOCATION, "") - - @property - def csq_gnomadg_af_popmax(self) -> Union[float, None]: - return self._properties.get(self.CSQ_GNOMADG_AF_POPMAX) - - @property - def csq_gnomade2_af_popmax(self) -> Union[float, None]: - return self._properties.get(self.CSQ_GNOMADE2_AF_POPMAX) - - @property - def _canonical_transcript(self) -> Union[None, Transcript]: - for transcript in self._transcripts: - if transcript.is_canonical(): - return transcript - - @property - def _most_severe_transcript(self) -> Union[None, Transcript]: - for transcript in self._transcripts: - if transcript.is_most_severe(): - return transcript - - def get_most_severe_location(self) -> str: - return self.most_severe_location - - def get_canonical_transcript_feature(self) -> str: - if self._canonical_transcript: - return self._canonical_transcript.get_feature() - return "" - - def get_most_severe_transcript_feature(self) -> str: - if self._most_severe_transcript: - return self._most_severe_transcript.get_feature() - return "" - - def get_canonical_transcript_consequence_names(self) -> str: - if self._canonical_transcript: - return self._canonical_transcript.get_consequence_names() - return "" - - def get_most_severe_transcript_consequence_names(self) -> str: - if self._most_severe_transcript: - return self._most_severe_transcript.get_consequence_names() - return "" - - def get_canonical_transcript_location(self) -> str: - if self._canonical_transcript: - return self._canonical_transcript.get_location() - return "" - - def get_most_severe_transcript_location(self) -> str: - if self._most_severe_transcript: - return self._most_severe_transcript.get_location() - return "" - - def get_gnomad_v3_popmax_population(self) -> str: - result = "" - gnomad_v3_af_popmax = self.csq_gnomadg_af_popmax - if gnomad_v3_af_popmax: - result = self._get_gnomad_v3_population_for_allele_fraction( - gnomad_v3_af_popmax - ) - return result - - def get_gnomad_v2_popmax_population(self) -> str: - result = "" - gnomad_v2_af_popmax = self.csq_gnomade2_af_popmax - if gnomad_v2_af_popmax: - result = self._get_gnomad_v2_population_for_allele_fraction( - gnomad_v2_af_popmax - ) - return result - - def _get_gnomad_v3_population_for_allele_fraction( - self, allele_fraction: float - ) -> str: - return self._get_gnomad_population_for_allele_fraction( - self.GNOMAD_V3_AF_PREFIX, allele_fraction - ) - - def _get_gnomad_v2_population_for_allele_fraction( - self, allele_fraction: float - ) -> str: - return self._get_gnomad_population_for_allele_fraction( - self.GNOMAD_V2_AF_PREFIX, allele_fraction - ) - - def _get_gnomad_population_for_allele_fraction( - self, gnomad_af_prefix: str, allele_fraction: float - ) -> str: - result = "" - for ( - gnomad_suffix, - population_name, - ) in self.GNOMAD_POPULATION_SUFFIX_TO_NAME.items(): - population_property_name = gnomad_af_prefix + gnomad_suffix - allele_frequency = self._properties.get(population_property_name) - if allele_frequency == allele_fraction: - result = population_name - break - return result - - -@dataclass(frozen=True) -class Note(CGAPItem): - pass - - -@dataclass(frozen=True) -class VariantSample(CGAPItem): - # Schema constants - VARIANT = "variant" - - @property - def variant(self) -> Variant: - return Variant(self._properties.get(self.VARIANT, {})) - - def get_canonical_transcript_feature(self) -> str: - return self.variant.get_canonical_transcript_feature() - - def get_canonical_transcript_location(self) -> str: - return self.variant.get_canonical_transcript_location() - - def get_canonical_transcript_consequence_names(self) -> str: - return self.variant.get_canonical_transcript_consequence_names() - - def get_most_severe_transcript_feature(self) -> str: - return self.variant.get_most_severe_transcript_feature() - - def get_most_severe_transcript_location(self) -> str: - return self.variant.get_most_severe_transcript_location() - - def get_most_severe_transcript_consequence_names(self) -> str: - return self.variant.get_most_severe_transcript_consequence_names() - - def get_gnomad_v3_popmax_population(self) -> str: - return self.variant.get_gnomad_v3_popmax_population() - - def get_gnomad_v2_popmax_population(self) -> str: - return self.variant.get_gnomad_v2_popmax_population() - - -@dataclass(frozen=True) -class VariantSampleSelectionFromVariantSampleList(SubembeddedProperty): - VARIANT_SAMPLE_ITEM = "variant_sample_item" - - @property - def variant_sample_item(self) -> Union[VariantSample, str]: - variant_sample = self._properties.get(self.VARIANT_SAMPLE_ITEM, "") - return self._get_embeddable_property(variant_sample, VariantSample) - - def get_variant_sample(self): - return self.variant_sample_item - - -@dataclass(frozen=True) -class VariantSampleList(CGAPItem): - CREATED_FOR_CASE = "created_for_case" - VARIANT_SAMPLES = "variant_samples" - - @property - def created_for_case(self) -> str: - return self._properties.get(self.CREATED_FOR_CASE, "") - - @property - def variant_samples(self) -> List[VariantSampleSelectionFromVariantSampleList]: - variant_samples = self._properties.get(self.VARIANT_SAMPLES, []) - return [ - VariantSampleSelectionFromVariantSampleList.from_properties( - variant_sample, parent_item=self - ) - for variant_sample in variant_samples - ] - - def get_associated_case_accession(self) -> str: - return self.created_for_case - - def get_variant_samples(self) -> List[Union[VariantSample, str]]: - return [ - variant_sample_selection.get_variant_sample() - for variant_sample_selection in self.variant_samples - ] diff --git a/dcicutils/item_model_utils.py b/dcicutils/item_model_utils.py new file mode 100644 index 000000000..510b64dba --- /dev/null +++ b/dcicutils/item_model_utils.py @@ -0,0 +1,222 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from functools import lru_cache +from typing import Any, Iterable, List, Mapping, Optional, Tuple, Union + +import structlog + +from . import ff_utils + + +logger = structlog.getLogger(__name__) + +JsonObject = Mapping[str, Any] +LinkTo = Union[str, JsonObject] + + +def get_link_to( + existing_item: PortalItem, + link_to: LinkTo, + item_to_create: PortalItem, +) -> Union[str, PortalItem]: + """Create new item model from existing one for given linkTo. + + LinkTos be identifiers (UUIDs) or (partially) embedded objects. + + Follow rules of existing item model for fetching linkTo via + request. If not fetching via request, then make item model from + existing properties if possible. + """ + fetch_links = existing_item.should_fetch_links() + identifier = get_item_identifier(link_to) + if fetch_links and identifier: + return item_to_create.from_identifier_and_existing_item( + identifier, existing_item + ) + if isinstance(link_to, Mapping): + return item_to_create.from_properties_and_existing_item(link_to, existing_item) + return link_to + + +def get_item_identifier(item: LinkTo) -> str: + if isinstance(item, str): + return item + if isinstance(item, Mapping): + return item.get(PortalItem.UUID, "") + + +@dataclass(frozen=True) +class PortalItem: + ACCESSION = "accession" + AT_ID = "@id" + TYPE = "@type" + UUID = "uuid" + + properties: JsonObject + auth: Optional[JsonObject] = field(default=None, hash=False) + fetch_links: Optional[bool] = field(default=False, hash=False) + + def __repr__(self) -> str: + return f"{self.__class__.__name__}({self._uuid})" + + @property + def _uuid(self) -> str: + return self.properties.get(self.UUID, "") + + @property + def _at_id(self) -> str: + return self.properties.get(self.AT_ID, "") + + @property + def _accession(self) -> str: + return self.properties.get(self.ACCESSION, "") + + @property + def _types(self) -> List[str]: + return self.properties.get(self.TYPE, []) + + def should_fetch_links(self) -> bool: + return self.fetch_links + + def get_auth(self) -> Union[JsonObject, None]: + return self.auth + + def get_properties(self) -> JsonObject: + return self.properties + + def get_accession(self) -> str: + return self._accession + + def get_uuid(self) -> str: + return self._uuid + + def get_at_id(self) -> str: + return self._at_id + + def get_types(self) -> List[str]: + return self._types + + def _get_link_tos( + self, link_tos: Iterable[LinkTo], item_to_create: PortalItem + ) -> List[Union[str, PortalItem]]: + return [self._get_link_to(link_to, item_to_create) for link_to in link_tos] + + def _get_link_to( + self, + link_to: LinkTo, + item_to_create: PortalItem, + ) -> Union[str, PortalItem]: + return get_link_to(self, link_to, item_to_create) + + @classmethod + def from_properties( + cls, + properties: JsonObject, + fetch_links: bool = False, + auth: JsonObject = None, + **kwargs: Any, + ) -> PortalItem: + return cls(properties, fetch_links=fetch_links, auth=auth) + + @classmethod + def from_identifier_and_auth( + cls, identifier: str, auth: JsonObject, fetch_links=False, **kwargs: Any + ) -> PortalItem: + properties = cls._get_item_via_auth(identifier, auth) + return cls.from_properties(properties, auth=auth, fetch_links=fetch_links) + + @classmethod + def _get_item_via_auth( + cls, identifier: str, auth: JsonObject, add_on: Optional[str] = "frame=object" + ) -> JsonObject: + hashable_auth = cls._make_hashable_auth(auth) + return cls._get_and_cache_item_via_auth(identifier, hashable_auth, add_on) + + @classmethod + def _make_hashable_auth(cls, auth: Mapping[str, str]) -> Tuple[Tuple[str, str]]: + """Assuming nothing nested here...""" + return tuple(auth.items()) + + @classmethod + def _undo_make_hashable_auth( + cls, hashable_auth: Tuple[Tuple[str, str]] + ) -> JsonObject: + return dict(hashable_auth) + + @classmethod + @lru_cache(maxsize=256) + def _get_and_cache_item_via_auth( + cls, + identifier: str, + hashable_auth: Tuple[Tuple[str, Any]], + add_on: Optional[str] = None, + ) -> JsonObject: + """Save on requests by caching items.""" + auth = cls._undo_make_hashable_auth(hashable_auth) + try: + result = ff_utils.get_metadata(identifier, key=auth, add_on=add_on) + except Exception as e: + result = {} + logger.error(f"Error getting metadata for {identifier}: {e}") + return result + + @classmethod + def from_identifier_and_existing_item( + cls, identifier: str, existing_item: PortalItem, **kwargs: Any + ) -> PortalItem: + fetch_links = existing_item.should_fetch_links() + auth = existing_item.get_auth() + if auth: + return cls.from_identifier_and_auth( + identifier, auth, fetch_links=fetch_links + ) + raise RuntimeError("Unable to fetch given identifier without auth key") + + @classmethod + def from_properties_and_existing_item( + cls, properties: JsonObject, existing_item: PortalItem, **kwargs: Any + ) -> PortalItem: + fetch_links = existing_item.should_fetch_links() + auth = existing_item.get_auth() + return cls.from_properties(properties, fetch_links=fetch_links, auth=auth) + + +@dataclass(frozen=True) +class NestedProperty: + properties: JsonObject + parent_item: Optional[PortalItem] = field(default=None, hash=False) + + def __repr__(self) -> str: + return f"{self.__class__.__name__}(parent={self.parent_item.__repr__()})" + + def get_properties(self) -> JsonObject: + return self.properties + + def get_parent_item(self) -> Union[PortalItem, None]: + return self.parent_item + + def _get_link_tos( + self, link_tos: LinkTo, item_to_create: PortalItem + ) -> List[Union[str, PortalItem]]: + return [self._get_link_to(link_to, item_to_create) for link_to in link_tos] + + def _get_link_to( + self, + link_to: LinkTo, + item_to_create: PortalItem, + ) -> Union[str, PortalItem]: + if self.parent_item: + return get_link_to(self.parent_item, link_to, item_to_create) + if isinstance(link_to, Mapping): + return item_to_create.from_properties(link_to) + return link_to + + @classmethod + def from_properties( + cls, + properties: JsonObject, + parent_item: Optional[PortalItem] = None, + **kwargs: Any, + ) -> NestedProperty: + return cls(properties, parent_item=parent_item) diff --git a/dcicutils/item_models.py b/dcicutils/item_models.py deleted file mode 100644 index 7c8ac7f6e..000000000 --- a/dcicutils/item_models.py +++ /dev/null @@ -1,211 +0,0 @@ -from __future__ import annotations -from dataclasses import dataclass, field -from functools import lru_cache -from typing import Any, Mapping, Optional, Tuple, Union - -import structlog - -from .ff_utils import get_metadata - - -logger = structlog.getLogger(__name__) - -JsonObject = Mapping[str, Any] - - -def _make_embeddable_property( - existing_item: PortalItem, - property_value: [str, JsonObject], - item_to_create: PortalItem, -) -> Union[str, JsonObject, PortalItem]: - embed_items = existing_item.embed_items() - identifier = _get_item_identifier(property_value) - if embed_items and identifier: - return item_to_create.from_identifier_and_existing_item( - identifier, existing_item - ) - if isinstance(property_value, Mapping): - return item_to_create.from_properties_and_existing_item( - property_value, existing_item - ) - return property_value - - -def _get_item_identifier(item: Union[str, JsonObject]) -> str: - if isinstance(item, str): - return item - if isinstance(item, Mapping): - return item.get(PortalItem.UUID, "") - raise ValueError() - - -@dataclass(frozen=True) -class PortalItem: - AT_ID = "@id" - UUID = "uuid" - - IDENTIFYING_PROPERTIES = [AT_ID, UUID] - - properties: Optional[JsonObject] = field(default=None, hash=False) - auth: Optional[JsonObject] = field(default=None, hash=False) - do_embeds: Optional[bool] = field(default=False, hash=False) - - def __repr__(self) -> str: - return f"{self.__class__.__name__}({self.uuid})" - - def __eq__(self, other: Any) -> bool: - if isinstance(other, PortalItem) and self._uuid: - return self._uuid == other._uuid - return False - - @property - def _properties(self) -> JsonObject: - return self.properties or {} - - @property - def _uuid(self) -> str: - return self._properties.get(self.UUID, "") - - @property - def _at_id(self) -> str: - return self._properties.get(self.AT_ID, "") - - def is_same_item(self, item: Any) -> bool: - if isinstance(item, PortalItem): - return self == item - if isinstance(item, dict): - return self._has_same_identifier(item) - if isinstance(item, str): - return self._is_same_identifier(item) - return False - - def _has_same_identifier(self, item: Mapping) -> bool: - return any( - [ - self._properties.get(identifying_property) == item.get(identifying_property) - for identifying_property in self.IDENTIFYING_PROPERTIES - ] - ) - - def _is_same_identifier(self, identifier: str) -> bool: - return any( - [ - self._properties.get(identifying_property) == identifier - for identifying_property in self.IDENTIFYING_PROPERTIES - ] - ) - - def do_embeds(self) -> bool: - return self.do_embeds - - def get_auth(self) -> Union[JsonObject, None]: - return self.auth - - def get_properties(self) -> JsonObject: - return self._properties - - def _get_embeddable_property( - self, - property_value: [str, JsonObject], - item_to_create: PortalItem, - ) -> Union[str, JsonObject, PortalItem]: - return _make_embeddable_property(self, property_value, item_to_create) - - @classmethod - def from_properties( - cls, properties: JsonObject, embed_items=False, auth=None, **kwargs: Any - ) -> PortalItem: - return cls(properties=properties, embed_items=embed_items, auth=auth) - - @classmethod - def from_identifier_and_auth( - cls, identifier: str, auth: JsonObject, embed_items=False, **kwargs: Any - ) -> PortalItem: - properties = cls._get_item_via_auth(identifier, auth) - return cls.from_properties(properties=properties, auth=auth, embed_items=embed_items) - - @classmethod - def _get_item_via_auth( - cls, identifier: str, auth: JsonObject, add_on: Optional[str] = "frame=object" - ) -> JsonObject: - hashable_auth = cls._make_hashable_auth(auth) - return cls._get_and_cache_item_via_auth(identifier, hashable_auth, add_on) - - @classmethod - def _make_hashable_auth(cls, auth: Mapping[str, str]) -> Tuple[Tuple[str, str]]: - """Assuming nothing nested here...""" - return tuple(auth.items()) - - @classmethod - def _undo_make_hashable_auth( - cls, hashable_auth: Tuple[Tuple[str, str]] - ) -> JsonObject: - return dict(hashable_auth) - - @classmethod - @lru_cache(maxsize=256) - def _get_and_cache_item_via_auth( - cls, identifier: str, hashable_auth: Tuple[Tuple[str, Any]], add_on: Optional[str] = None - ) -> JsonObject: - """Save on requests by caching items.""" - auth = cls._undo_make_hashable_auth(hashable_auth) - try: - result = get_metadata(identifier, key=auth, add_on=add_on) - except Exception as e: - result = {} - logger.error(f"Error getting metadata for {identifier}: {e}") - return result - - @classmethod - def from_identifier_and_existing_item( - cls, identifier: str, existing_item: PortalItem, **kwargs: Any - ) -> PortalItem: - embed_items = existing_item.embed_items() - auth = existing_item.get_auth() - if auth: - return cls.from_identifier_and_auth( - identifier, auth, embed_items=embed_items - ) - raise ValueError("Unable to create item from existing item") - - @classmethod - def from_properties_and_existing_item( - cls, properties: JsonObject, existing_item: PortalItem, **kwargs: Any - ) -> PortalItem: - embed_items = existing_item.embed_items() - auth = existing_item.get_auth() - return cls.from_properties( - properties, embed_items=embed_items, auth=auth - ) - - -@dataclass(frozen=True) -class SubembeddedProperty: - properties: Optional[JsonObject] = field(default=None, hash=False) - parent_item: Optional[PortalItem] = field(default=None, hash=False) - - def __repr__(self) -> str: - return f"{self.__class__.__name__}(parent={self.parent_item.__repr__()})" - - @property - def _properties(self) -> JsonObject: - return self.properties or {} - - def _get_embeddable_property( - self, - property_value: [str, JsonObject], - item_to_create: PortalItem, - ) -> Union[str, JsonObject, PortalItem]: - if self.parent_item: - return _make_embeddable_property( - self.parent_item, property_value, item_to_create - ) - if isinstance(property_value, Mapping): - return item_to_create.from_properties(property_value) - return property_value - - @classmethod - def from_properties( - cls, properties: JsonObject, parent_item: Optional[PortalItem], **kwargs: Any - ) -> SubembeddedProperty: - return cls(properties=properties, parent_item=parent_item) diff --git a/dcicutils/testing_utils.py b/dcicutils/testing_utils.py index fb8e25e28..28eb3d17e 100644 --- a/dcicutils/testing_utils.py +++ b/dcicutils/testing_utils.py @@ -1,16 +1,38 @@ from contextlib import contextmanager -from typing import Any, Iterator, Optional +from typing import Any, Iterator from unittest import mock +AN_UNLIKELY_RETURN_VALUE = "unlikely return value" + + @contextmanager def patch_context( - object_to_patch: object, - attribute_to_patch: str, - return_value: Optional[Any] = None, + to_patch: object, + return_value: Any = AN_UNLIKELY_RETURN_VALUE, **kwargs, ) -> Iterator[mock.MagicMock]: - with mock.patch.object(object_to_patch, attribute_to_patch, **kwargs) as mocked_item: - if return_value is not None: + if isinstance(to_patch, property): + to_patch = to_patch.fget + new_callable = mock.PropertyMock + else: + new_callable = mock.MagicMock + target = f"{to_patch.__module__}.{to_patch.__qualname__}" + with mock.patch(target, new_callable=new_callable, **kwargs) as mocked_item: + if return_value != AN_UNLIKELY_RETURN_VALUE: mocked_item.return_value = return_value yield mocked_item +# +# +# +# @contextmanager +# def patch_context( +# object_to_patch: object, +# attribute_to_patch: str, +# return_value: Optional[Any] = None, +# **kwargs, +# ) -> Iterator[mock.MagicMock]: +# with mock.patch.object(object_to_patch, attribute_to_patch, **kwargs) as mocked_item: +# if return_value is not None: +# mocked_item.return_value = return_value +# yield mocked_item diff --git a/test/test_cgap_items.py b/test/test_cgap_items.py deleted file mode 100644 index 655df26c1..000000000 --- a/test/test_cgap_items.py +++ /dev/null @@ -1,29 +0,0 @@ -import json -from pathlib import Path - -from dcicutils.cgap_items import User - - -SOME_USER = { - "uuid": "some-uuid", - "email": "some-email", - "first_name": "some-first-name", - "last_name": "some-last-name", - "title": "some-title", - "project": "some-project", -} -KEYS_FILE = Path.expanduser(Path("~/.cgap-keys.json")).absolute() -USER_IDENTIFIER = "5196db84-f3d9-44bd-bba0-59e9e83634a1" - - -def get_keys(): - keys = json.loads(KEYS_FILE.read_text()) - return keys["msa"] - - -def test_user(): - auth = get_keys() - user = User.from_identifier_and_auth(USER_IDENTIFIER, auth, embed_items=True) - import pdb; pdb.set_trace() - project = user.project - assert user.uuid == USER_IDENTIFIER diff --git a/test/test_item_model_utils.py b/test/test_item_model_utils.py new file mode 100644 index 000000000..7ab23e3b0 --- /dev/null +++ b/test/test_item_model_utils.py @@ -0,0 +1,388 @@ +from contextlib import contextmanager +from random import random +from typing import Any, Iterator, Optional, Tuple, Union +from unittest import mock + +import pytest + +from dcicutils import item_model_utils as item_models_module +from dcicutils.item_model_utils import ( + JsonObject, + LinkTo, + get_item_identifier, + get_link_to, + PortalItem, + NestedProperty, +) +from dcicutils.testing_utils import patch_context + + +SOME_UUID = "uuid1234" +SOME_AT_ID = "/foo/bar/" +SOME_ACCESSION = "GAPXY12345" +SOME_TYPES = ["SomeItemType", "Item"] +SOME_ITEM_PROPERTIES = { + "uuid": SOME_UUID, + "@id": SOME_AT_ID, + "accession": SOME_ACCESSION, + "@type": SOME_TYPES, +} +OTHER_ITEM_PROPERTIES = {"uuid": "foo"} +SOME_AUTH = {"key": "some_key", "secret": "some_secret"} +HASHABLE_SOME_AUTH = (("key", "some_key"), ("secret", "some_secret")) + + +@contextmanager +def patch_get_link_to(**kwargs: Any) -> Iterator[mock.MagicMock]: + with patch_context(item_models_module.get_link_to, **kwargs) as mock_item: + yield mock_item + + +@contextmanager +def patch_item_get_link_to(**kwargs: Any) -> Iterator[mock.MagicMock]: + with patch_context( + item_models_module.PortalItem._get_link_to, + **kwargs, + ) as mock_item: + yield mock_item + + +@contextmanager +def patch_nested_get_link_to(**kwargs: Any) -> Iterator[mock.MagicMock]: + with patch_context( + item_models_module.NestedProperty._get_link_to, + **kwargs, + ) as mock_item: + yield mock_item + + +@contextmanager +def patch_get_item_via_auth(**kwargs: Any) -> Iterator[mock.MagicMock]: + with patch_context( + item_models_module.PortalItem._get_item_via_auth, + **kwargs, + ) as mock_item: + yield mock_item + + +@contextmanager +def patch_get_and_cache_item_via_auth(**kwargs: Any) -> Iterator[mock.MagicMock]: + with patch_context( + item_models_module.PortalItem._get_and_cache_item_via_auth, + **kwargs, + ) as mock_item: + yield mock_item + + +@contextmanager +def patch_from_identifier_and_auth(**kwargs: Any) -> Iterator[mock.MagicMock]: + with patch_context( + item_models_module.PortalItem.from_identifier_and_auth, + **kwargs, + ) as mock_item: + yield mock_item + + +@contextmanager +def patch_get_metadata(**kwargs: Any) -> Iterator[mock.MagicMock]: + with patch_context( + item_models_module.ff_utils.get_metadata, + **kwargs, + ) as mock_item: + yield mock_item + + +def mock_portal_item(): + return mock.create_autospec(PortalItem, instance=True) + + +def get_portal_item( + properties: Optional[JsonObject] = None, + fetch_links: Optional[bool] = False, + auth: Optional[bool] = None, + use_defaults: Optional[bool] = True, +) -> PortalItem: + if properties is None and use_defaults: + properties = SOME_ITEM_PROPERTIES + if auth is None and use_defaults: + auth = SOME_AUTH + return PortalItem(auth=auth, fetch_links=fetch_links, properties=properties) + + +@pytest.mark.parametrize( + "fetch_links,link_to,expected_from_identifier,expected_from_properties", + [ + (False, SOME_UUID, False, False), + (True, SOME_UUID, True, False), + (True, SOME_ITEM_PROPERTIES, True, False), + (False, SOME_ITEM_PROPERTIES, False, True), + ], +) +def test_get_link_to( + fetch_links: bool, + link_to: LinkTo, + expected_from_identifier: bool, + expected_from_properties: bool, +) -> None: + existing_item = get_portal_item(fetch_links=fetch_links) + item_to_create = mock_portal_item() + result = get_link_to(existing_item, link_to, item_to_create) + if expected_from_identifier: + identifier = get_item_identifier(link_to) + assert result == item_to_create.from_identifier_and_existing_item.return_value + item_to_create.from_identifier_and_existing_item.assert_called_once_with( + identifier, existing_item + ) + elif expected_from_properties: + assert result == item_to_create.from_properties_and_existing_item.return_value + item_to_create.from_properties_and_existing_item.assert_called_once_with( + link_to, existing_item + ) + else: + assert result == link_to + + +@pytest.mark.parametrize( + "item,expected", + [ + ("", ""), + ({}, ""), + ("foo", "foo"), + ({"uuid": "foo"}, "foo"), + ], +) +def test_get_item_identifier(item: Union[str, JsonObject], expected: str) -> None: + result = get_item_identifier(item) + assert result == expected + + +class TestPortalItem: + @pytest.mark.parametrize( + "properties,expected", [({}, ""), (SOME_ITEM_PROPERTIES, SOME_UUID)] + ) + def test_get_uuid(self, properties: JsonObject, expected: str) -> None: + portal_item = get_portal_item(properties=properties) + assert portal_item.get_uuid() == expected + + @pytest.mark.parametrize( + "properties,expected", [({}, ""), (SOME_ITEM_PROPERTIES, SOME_AT_ID)] + ) + def test_get_at_id(self, properties: JsonObject, expected: str) -> None: + portal_item = get_portal_item(properties=properties) + assert portal_item.get_at_id() == expected + + @pytest.mark.parametrize( + "properties,expected", [({}, ""), (SOME_ITEM_PROPERTIES, SOME_ACCESSION)] + ) + def test_get_accession(self, properties: JsonObject, expected: str) -> None: + portal_item = get_portal_item(properties=properties) + assert portal_item.get_accession() == expected + + @pytest.mark.parametrize( + "properties,expected", [({}, []), (SOME_ITEM_PROPERTIES, SOME_TYPES)] + ) + def test_get_types(self, properties: JsonObject, expected: str) -> None: + portal_item = get_portal_item(properties=properties) + assert portal_item.get_types() == expected + + def test_get_link_tos(self) -> None: + link_tos = [SOME_UUID, SOME_ACCESSION] + item_to_create = mock_portal_item() + portal_item = get_portal_item() + with patch_item_get_link_to() as mock_get_link_to: + portal_item._get_link_tos(link_tos, item_to_create) + assert len(mock_get_link_to.call_args_list) == len(link_tos) + for link_to in link_tos: + mock_get_link_to.assert_any_call(link_to, item_to_create) + + def test_get_link_to(self) -> None: + link_to = "foo" + item_to_create = mock_portal_item() + portal_item = get_portal_item() + with patch_get_link_to() as mock_get_link_to: + portal_item._get_link_to(link_to, item_to_create) + assert mock_get_link_to.called_once_with( + portal_item, link_to, item_to_create + ) + + def test_from_properties(self) -> None: + properties = OTHER_ITEM_PROPERTIES + fetch_links = True + auth = SOME_AUTH + portal_item = get_portal_item() + result = portal_item.from_properties( + properties, fetch_links=fetch_links, auth=auth + ) + assert isinstance(result, PortalItem) + assert result.get_properties() == properties + assert result.should_fetch_links() == fetch_links + assert result.get_auth() == auth + + def test_from_identifier_and_auth(self) -> None: + with patch_get_item_via_auth( + return_value=OTHER_ITEM_PROPERTIES + ) as mock_get_item_via_auth: + identifier = SOME_UUID + fetch_links = True + portal_item = get_portal_item() + result = portal_item.from_identifier_and_auth( + identifier, SOME_AUTH, fetch_links=fetch_links + ) + assert isinstance(result, PortalItem) + assert result.should_fetch_links() == fetch_links + assert result.get_auth() == SOME_AUTH + mock_get_item_via_auth.assert_called_once_with(identifier, SOME_AUTH) + + def test_get_item_via_auth(self) -> None: + with patch_get_and_cache_item_via_auth( + return_value=SOME_ITEM_PROPERTIES + ) as mock_get_and_cache_item: + portal_item = get_portal_item() + result = portal_item._get_item_via_auth(SOME_UUID, SOME_AUTH) + assert result == SOME_ITEM_PROPERTIES + mock_get_and_cache_item.assert_called_once_with( + SOME_UUID, HASHABLE_SOME_AUTH, "frame=object" + ) + + @pytest.mark.parametrize( + "auth,expected", [({}, tuple()), (SOME_AUTH, HASHABLE_SOME_AUTH)] + ) + def test_make_hashable_auth(self, auth: JsonObject, expected: Tuple) -> None: + portal_item = get_portal_item() + result = portal_item._make_hashable_auth(auth) + assert result == expected + + @pytest.mark.parametrize( + "hashable_auth,expected", [(tuple(), {}), (HASHABLE_SOME_AUTH, SOME_AUTH)] + ) + def test_undo_make_hashable_auth( + self, hashable_auth: Tuple, expected: JsonObject + ) -> None: + portal_item = get_portal_item() + result = portal_item._undo_make_hashable_auth(hashable_auth) + assert result == expected + + @pytest.mark.parametrize( + "raise_exception,expected", [(False, SOME_ITEM_PROPERTIES), (True, {})] + ) + def test_get_and_cache_item_via_auth( + self, raise_exception: bool, expected: JsonObject + ) -> None: + side_effect = Exception if raise_exception else None + random_add_on = str(random()) # To differentiate parametrized calls + with patch_get_metadata( + side_effect=side_effect, return_value=SOME_ITEM_PROPERTIES + ) as mock_get_metadata: + portal_item = get_portal_item() + result = portal_item._get_and_cache_item_via_auth( + SOME_UUID, HASHABLE_SOME_AUTH, add_on=random_add_on + ) + assert result == expected + assert len(mock_get_metadata.call_args_list) == 1 + mock_get_metadata.assert_called_once_with( + SOME_UUID, key=SOME_AUTH, add_on=random_add_on + ) + + # Ensure cached + portal_item._get_and_cache_item_via_auth( + SOME_UUID, HASHABLE_SOME_AUTH, add_on=random_add_on + ) + assert len(mock_get_metadata.call_args_list) == 1 + + @pytest.mark.parametrize( + "auth,fetch_links,exception_expected", + [ + (None, False, True), + (None, True, True), + (SOME_AUTH, False, False), + (SOME_AUTH, True, False), + ], + ) + def test_from_identifier_and_existing_item( + self, auth: JsonObject, fetch_links: bool, exception_expected: bool + ) -> None: + identifier = SOME_UUID + portal_item = get_portal_item( + auth=auth, fetch_links=fetch_links, use_defaults=False + ) + with patch_from_identifier_and_auth() as mock_from_identifier_and_auth: + if exception_expected: + with pytest.raises(RuntimeError): + PortalItem.from_identifier_and_existing_item( + identifier, portal_item + ) + else: + result = PortalItem.from_identifier_and_existing_item( + identifier, portal_item + ) + assert result == mock_from_identifier_and_auth.return_value + mock_from_identifier_and_auth.assert_called_once_with( + identifier, auth, fetch_links=fetch_links + ) + + def test_from_properties_and_existing_item(self) -> None: + properties = OTHER_ITEM_PROPERTIES + portal_item = get_portal_item() + result = PortalItem.from_properties_and_existing_item(properties, portal_item) + assert isinstance(result, PortalItem) + assert result.get_auth() == portal_item.get_auth() + assert result.should_fetch_links() == portal_item.should_fetch_links() + + +def get_nested_property( + properties: Optional[JsonObject] = None, parent_item: Optional[PortalItem] = None +) -> NestedProperty: + properties = properties or SOME_ITEM_PROPERTIES + return NestedProperty(properties=properties, parent_item=parent_item) + + +class TestNestedProperty: + def test_get_link_tos(self) -> None: + link_tos = [SOME_UUID, SOME_ACCESSION] + item_to_create = mock_portal_item() + nested_property = get_nested_property() + with patch_nested_get_link_to() as mock_get_link_to: + nested_property._get_link_tos(link_tos, item_to_create) + assert len(mock_get_link_to.call_args_list) == len(link_tos) + for link_to in link_tos: + mock_get_link_to.assert_any_call(link_to, item_to_create) + + @pytest.mark.parametrize( + "link_to,parent_item,expected_get_link_to_call,expected_item_from_properties", + [ + (SOME_UUID, None, False, False), + (SOME_ITEM_PROPERTIES, None, False, True), + (SOME_UUID, get_portal_item(), True, False), + (SOME_ITEM_PROPERTIES, get_portal_item(), True, False), + ], + ) + def test_get_link_to( + self, + link_to: LinkTo, + parent_item: Union[PortalItem, None], + expected_get_link_to_call: bool, + expected_item_from_properties: bool, + ) -> None: + item_to_create = mock_portal_item() + nested_property = get_nested_property(parent_item=parent_item) + with patch_get_link_to() as mock_get_link_to: + result = nested_property._get_link_to(link_to, item_to_create) + if expected_get_link_to_call: + mock_get_link_to.assert_called_once_with( + parent_item, link_to, item_to_create + ) + assert result == mock_get_link_to.return_value + elif expected_item_from_properties: + item_to_create.from_properties.assert_called_once_with(link_to) + assert result == item_to_create.from_properties.return_value + else: + assert result == link_to + + def test_from_properties(self) -> None: + properties = OTHER_ITEM_PROPERTIES + parent_item = mock_portal_item() + nested_property = get_nested_property() + result = nested_property.from_properties(properties, parent_item=parent_item) + assert isinstance(result, NestedProperty) + assert result.get_properties() == properties + assert result.get_parent_item() == parent_item diff --git a/test/test_item_models.py b/test/test_item_models.py deleted file mode 100644 index d183c22d0..000000000 --- a/test/test_item_models.py +++ /dev/null @@ -1,256 +0,0 @@ -from contextlib import contextmanager -from random import random -from typing import Any, Iterator, Optional, Tuple, Union -from unittest import mock - -import pytest - -from dcicutils import item_models as item_models_module -from dcicutils.item_models import ( - JsonObject, - _make_embeddable_property, - _get_item_identifier, - PortalItem, - SubembeddedProperty, -) -from dcicutils.testing_utils import patch_context - - -SOME_UUID = "uuid1234" -SOME_AT_ID = "/foo/bar/" -SOME_ITEM_PROPERTIES = {"uuid": SOME_UUID, "@id": SOME_AT_ID} -OTHER_ITEM_PROPERTIES = {"uuid": "foo"} -SOME_AUTH = {"key": "some_key", "secret": "some_secret"} -HASHABLE_SOME_AUTH = (("key", "some_key"), ("secret", "some_secret")) - - -@contextmanager -def patch_make_embeddable_property(**kwargs: Any) -> Iterator[mock.MagicMock]: - with patch_context( - item_models_module, "_make_embeddable_property", **kwargs - ) as mock_item: - yield mock_item - - -@contextmanager -def patch_get_and_cache_item_via_auth(**kwargs: Any) -> Iterator[mock.MagicMock]: - with patch_context( - item_models_module.PortalItem, - "_get_and_cache_item_via_auth", - **kwargs, - ) as mock_item: - yield mock_item - - -@contextmanager -def patch_get_metadata(**kwargs: Any) -> Iterator[mock.MagicMock]: - with patch_context( - item_models_module, - "get_metadata", - **kwargs, - ) as mock_item: - yield mock_item - - -def mock_portal_item(embed_items: bool = False): - mock_item = mock.create_autospec(PortalItem, instance=True) - mock_item.embed_items.return_value = embed_items - return mock_item - - -@pytest.mark.parametrize( - "embed_items,property_value,expected_from_identifier,expected_from_properties", - [ - (False, "foo", False, False), - (True, "foo", True, False), - (False, {}, False, True), - (False, SOME_ITEM_PROPERTIES, False, True), - (True, SOME_ITEM_PROPERTIES, True, False), - ] -) -def test_make_embeddable_property( - embed_items: bool, property_value: Union[str, JsonObject], - expected_from_identifier: bool, - expected_from_properties: bool, -) -> None: - existing_item = mock_portal_item(embed_items) - item_to_create = mock_portal_item() - result = _make_embeddable_property(existing_item, property_value, item_to_create) - if expected_from_identifier: - identifier = _get_item_identifier(property_value) - assert result == item_to_create.from_identifier_and_existing_item.return_value - item_to_create.from_identifier_and_existing_item.assert_called_once_with( - identifier, existing_item - ) - elif expected_from_properties: - assert result == item_to_create.from_properties_and_existing_item.return_value - item_to_create.from_properties_and_existing_item.assert_called_once_with( - property_value, existing_item - ) - else: - assert result == property_value - - -@pytest.mark.parametrize( - "item,expected", - [ - ("", ""), - ({}, ""), - ("foo", "foo"), - ({"uuid": "foo"}, "foo"), - ] -) -def test_get_item_identifier(item: Union[str, JsonObject], expected: str) -> None: - result = _get_item_identifier(item) - assert result == expected - - -def get_portal_item( - properties: Optional[JsonObject] = None, - embed_items: Optional[bool] = False, - auth: Optional[bool] = None, -) -> PortalItem: - if properties is None: - properties = SOME_ITEM_PROPERTIES - if auth is None: - auth = SOME_AUTH - return PortalItem(auth=auth, embed_items=embed_items, properties=properties) - - -class TestPortalItem: - - @pytest.mark.parametrize( - "item_1_properties,item_2_properties,expected", - [ - ({}, {}, False), - ({}, SOME_ITEM_PROPERTIES, False), - (SOME_ITEM_PROPERTIES, OTHER_ITEM_PROPERTIES, False), - (SOME_ITEM_PROPERTIES, SOME_ITEM_PROPERTIES, True), - ] - ) - def test_equality( - self, - item_1_properties: JsonObject, item_2_properties: JsonObject, - expected: bool, - ) -> None: - item_1 = get_portal_item(properties=item_1_properties) - item_2 = get_portal_item(properties=item_2_properties) - assert (item_1 == item_2) == expected - - def test_properties(self) -> None: - assert PortalItem()._properties == {} - assert get_portal_item()._properties == SOME_ITEM_PROPERTIES - - @pytest.mark.parametrize( - "properties,expected", [({}, ""), (SOME_ITEM_PROPERTIES, SOME_UUID)] - ) - def test_uuid(self, properties: JsonObject, expected: str) -> None: - portal_item = get_portal_item(properties=properties) - assert portal_item.uuid == expected - - @pytest.mark.parametrize( - "properties,expected", [({}, ""), (SOME_ITEM_PROPERTIES, SOME_AT_ID)] - ) - def test_at_id(self, properties: JsonObject, expected: str) -> None: - portal_item = get_portal_item(properties=properties) - assert portal_item.at_id == expected - - @pytest.mark.parametrize( - "comparison_item,expected", - [ - (get_portal_item(), True), - (get_portal_item(properties=OTHER_ITEM_PROPERTIES), False), - (SOME_UUID, True), - (SOME_AT_ID, True), - ("some_string", False), - ] - ) - def test_is_same_item(self, comparison_item: Any, expected: bool) -> None: - portal_item = get_portal_item() - result = portal_item.is_same_item(comparison_item) - assert result == expected - - def test_get_embeddable_property(self) -> None: - property_value = "foo" - item_to_create = mock_portal_item() - portal_item = get_portal_item() - with patch_make_embeddable_property() as mock_make_embed: - portal_item._get_embeddable_property(property_value, item_to_create) - assert mock_make_embed.called_once_with( - portal_item, property_value, item_to_create - ) - - def test_get_item_via_auth(self) -> None: - with patch_get_and_cache_item_via_auth( - return_value=SOME_ITEM_PROPERTIES - ) as mock_get_and_cache_item: - portal_item = get_portal_item() - result = portal_item._get_item_via_auth(SOME_UUID, SOME_AUTH) - assert result == SOME_ITEM_PROPERTIES - mock_get_and_cache_item.assert_called_once_with( - SOME_UUID, HASHABLE_SOME_AUTH, "frame=object" - ) - - @pytest.mark.parametrize( - "auth,expected", [({}, tuple()), (SOME_AUTH, HASHABLE_SOME_AUTH)] - ) - def test_make_hashable_auth(self, auth: JsonObject, expected: Tuple) -> None: - portal_item = get_portal_item() - result = portal_item._make_hashable_auth(auth) - assert result == expected - - @pytest.mark.parametrize( - "hashable_auth,expected", [(tuple(), {}), (HASHABLE_SOME_AUTH, SOME_AUTH)] - ) - def test_undo_make_hashable_auth(self, hashable_auth: Tuple, expected: JsonObject) -> None: - portal_item = get_portal_item() - result = portal_item._undo_make_hashable_auth(hashable_auth) - assert result == expected - - @pytest.mark.parametrize( - "raise_exception,expected", [(False, SOME_ITEM_PROPERTIES), (True, {})] - ) - def test_get_ad_cache_item_via_auth(self, raise_exception: bool, expected: JsonObject) -> None: - side_effect = Exception if raise_exception else None - random_add_on = str(random()) # To differentiate parametrized calls - with patch_get_metadata(side_effect=side_effect, return_value=SOME_ITEM_PROPERTIES) as mock_get_metadata: - portal_item = get_portal_item() - result = portal_item._get_and_cache_item_via_auth( - SOME_UUID, HASHABLE_SOME_AUTH, add_on=random_add_on - ) - assert result == expected - assert len(mock_get_metadata.call_args_list) == 1 - mock_get_metadata.assert_called_once_with(SOME_UUID, key=SOME_AUTH, add_on=random_add_on) - - # Ensure cached - portal_item._get_and_cache_item_via_auth(SOME_UUID, HASHABLE_SOME_AUTH, add_on=random_add_on) - assert len(mock_get_metadata.call_args_list) == 1 - - -def get_subembedded_property(properties: Optional[JsonObject] = None, parent_item: Optional[PortalItem] = None) -> SubembeddedProperty: - properties = properties or SOME_ITEM_PROPERTIES - return SubembeddedProperty(properties=properties, parent_item=parent_item) - - -class TestSubembeddedProperty: - - @pytest.mark.parametrize( - ( - "property_value,parent_item_exists,expected_make_embeddable_property_call," - "expected" - ), - [ - ] - ) - def test_get_embeddable_property(self, property_value: Union[JsonObject, str], parent_item_exists: bool, expected_make_embeddable_property_call: bool, expected: Union[PortalItem, str]) -> None: - parent_item = mock_portal_item() if parent_item_exists else None - item_to_create = mock_portal_item() - with patch_make_embeddable_property() as mock_make_embeddable_property: - subembedded_property = get_subembedded_property(parent_item=parent_item) - result = subembedded_property._get_embeddable_property(property_value, item_to_create) - if expected_make_embeddable_property_call: - mock_make_embeddable_property.assert_called_once_with(parent_item, property_value, item_to_create) - assert result == mock_make_embeddable_property.return_value - else: - assert result == expected - mock_make_embeddable_property.assert_not_called() From c6003ec39cdd509efb3eca1148eaf7d9f8f0b2fe Mon Sep 17 00:00:00 2001 From: William Ronchetti Date: Thu, 20 Jul 2023 14:58:39 -0400 Subject: [PATCH 3/7] give a beta --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index c736d3d35..271490be8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "dcicutils" -version = "7.5.2" +version = "7.5.2.1b0" description = "Utility package for interacting with the 4DN Data Portal and other 4DN resources" authors = ["4DN-DCIC Team "] license = "MIT" From 8c8a62ecff5e7eba7a3b1b2de7af4c5d6dcc190f Mon Sep 17 00:00:00 2001 From: Douglas Rioux Date: Wed, 26 Jul 2023 10:53:11 -0400 Subject: [PATCH 4/7] Add module kwarg when patching --- dcicutils/item_model_utils.py | 2 +- dcicutils/testing_utils.py | 9 +++++++-- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/dcicutils/item_model_utils.py b/dcicutils/item_model_utils.py index 510b64dba..c0b4cf718 100644 --- a/dcicutils/item_model_utils.py +++ b/dcicutils/item_model_utils.py @@ -135,7 +135,7 @@ def _get_item_via_auth( @classmethod def _make_hashable_auth(cls, auth: Mapping[str, str]) -> Tuple[Tuple[str, str]]: - """Assuming nothing nested here...""" + """Assuming nothing nested here.""" return tuple(auth.items()) @classmethod diff --git a/dcicutils/testing_utils.py b/dcicutils/testing_utils.py index 28eb3d17e..0e89b387b 100644 --- a/dcicutils/testing_utils.py +++ b/dcicutils/testing_utils.py @@ -1,5 +1,6 @@ from contextlib import contextmanager -from typing import Any, Iterator +from types import ModuleType +from typing import Any, Iterator, Optional from unittest import mock @@ -10,6 +11,7 @@ def patch_context( to_patch: object, return_value: Any = AN_UNLIKELY_RETURN_VALUE, + module: Optional[ModuleType] = None, **kwargs, ) -> Iterator[mock.MagicMock]: if isinstance(to_patch, property): @@ -17,7 +19,10 @@ def patch_context( new_callable = mock.PropertyMock else: new_callable = mock.MagicMock - target = f"{to_patch.__module__}.{to_patch.__qualname__}" + if module is None: + target = f"{to_patch.__module__}.{to_patch.__qualname__}" + else: + target = f"{module.__name__}.{to_patch.__qualname__}" with mock.patch(target, new_callable=new_callable, **kwargs) as mocked_item: if return_value != AN_UNLIKELY_RETURN_VALUE: mocked_item.return_value = return_value From 000ec5faaa8ec624dfcd034473550f50dd185664 Mon Sep 17 00:00:00 2001 From: Douglas Rioux Date: Wed, 26 Jul 2023 10:54:15 -0400 Subject: [PATCH 5/7] Bump beta version --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 271490be8..a9f4b4a91 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "dcicutils" -version = "7.5.2.1b0" +version = "7.5.2.1b1" description = "Utility package for interacting with the 4DN Data Portal and other 4DN resources" authors = ["4DN-DCIC Team "] license = "MIT" From cbb954b9753ae41708433749df3c0c0e82b33546 Mon Sep 17 00:00:00 2001 From: Douglas Rioux Date: Thu, 3 Aug 2023 15:34:18 -0400 Subject: [PATCH 6/7] Clean up for PR --- dcicutils/item_model_utils.py | 40 ++++++++++++++++++----------------- dcicutils/testing_utils.py | 22 +++++++------------ test/test_item_model_utils.py | 26 +++++++++++------------ 3 files changed, 42 insertions(+), 46 deletions(-) diff --git a/dcicutils/item_model_utils.py b/dcicutils/item_model_utils.py index c0b4cf718..3f557049e 100644 --- a/dcicutils/item_model_utils.py +++ b/dcicutils/item_model_utils.py @@ -2,7 +2,7 @@ from dataclasses import dataclass, field from functools import lru_cache -from typing import Any, Iterable, List, Mapping, Optional, Tuple, Union +from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, Union import structlog @@ -11,8 +11,7 @@ logger = structlog.getLogger(__name__) -JsonObject = Mapping[str, Any] -LinkTo = Union[str, JsonObject] +LinkTo = Union[str, Dict[str, Any]] def get_link_to( @@ -22,7 +21,7 @@ def get_link_to( ) -> Union[str, PortalItem]: """Create new item model from existing one for given linkTo. - LinkTos be identifiers (UUIDs) or (partially) embedded objects. + LinkTos be identifiers (e.g. UUIDs) or (partially) embedded objects. Follow rules of existing item model for fetching linkTo via request. If not fetching via request, then make item model from @@ -53,8 +52,8 @@ class PortalItem: TYPE = "@type" UUID = "uuid" - properties: JsonObject - auth: Optional[JsonObject] = field(default=None, hash=False) + properties: Dict[str, Any] + auth: Optional[Dict[str, Any]] = field(default=None, hash=False) fetch_links: Optional[bool] = field(default=False, hash=False) def __repr__(self) -> str: @@ -79,10 +78,10 @@ def _types(self) -> List[str]: def should_fetch_links(self) -> bool: return self.fetch_links - def get_auth(self) -> Union[JsonObject, None]: + def get_auth(self) -> Union[Dict[str, Any], None]: return self.auth - def get_properties(self) -> JsonObject: + def get_properties(self) -> Dict[str, Any]: return self.properties def get_accession(self) -> str: @@ -112,24 +111,27 @@ def _get_link_to( @classmethod def from_properties( cls, - properties: JsonObject, + properties: Dict[str, Any], fetch_links: bool = False, - auth: JsonObject = None, + auth: Dict[str, Any] = None, **kwargs: Any, ) -> PortalItem: return cls(properties, fetch_links=fetch_links, auth=auth) @classmethod def from_identifier_and_auth( - cls, identifier: str, auth: JsonObject, fetch_links=False, **kwargs: Any + cls, identifier: str, auth: Dict[str, Any], fetch_links=False, **kwargs: Any ) -> PortalItem: properties = cls._get_item_via_auth(identifier, auth) return cls.from_properties(properties, auth=auth, fetch_links=fetch_links) @classmethod def _get_item_via_auth( - cls, identifier: str, auth: JsonObject, add_on: Optional[str] = "frame=object" - ) -> JsonObject: + cls, + identifier: str, + auth: Dict[str, Any], + add_on: Optional[str] = "frame=object", + ) -> Dict[str, Any]: hashable_auth = cls._make_hashable_auth(auth) return cls._get_and_cache_item_via_auth(identifier, hashable_auth, add_on) @@ -141,7 +143,7 @@ def _make_hashable_auth(cls, auth: Mapping[str, str]) -> Tuple[Tuple[str, str]]: @classmethod def _undo_make_hashable_auth( cls, hashable_auth: Tuple[Tuple[str, str]] - ) -> JsonObject: + ) -> Dict[str, Any]: return dict(hashable_auth) @classmethod @@ -151,7 +153,7 @@ def _get_and_cache_item_via_auth( identifier: str, hashable_auth: Tuple[Tuple[str, Any]], add_on: Optional[str] = None, - ) -> JsonObject: + ) -> Dict[str, Any]: """Save on requests by caching items.""" auth = cls._undo_make_hashable_auth(hashable_auth) try: @@ -175,7 +177,7 @@ def from_identifier_and_existing_item( @classmethod def from_properties_and_existing_item( - cls, properties: JsonObject, existing_item: PortalItem, **kwargs: Any + cls, properties: Dict[str, Any], existing_item: PortalItem, **kwargs: Any ) -> PortalItem: fetch_links = existing_item.should_fetch_links() auth = existing_item.get_auth() @@ -184,13 +186,13 @@ def from_properties_and_existing_item( @dataclass(frozen=True) class NestedProperty: - properties: JsonObject + properties: Dict[str, Any] parent_item: Optional[PortalItem] = field(default=None, hash=False) def __repr__(self) -> str: return f"{self.__class__.__name__}(parent={self.parent_item.__repr__()})" - def get_properties(self) -> JsonObject: + def get_properties(self) -> Dict[str, Any]: return self.properties def get_parent_item(self) -> Union[PortalItem, None]: @@ -215,7 +217,7 @@ def _get_link_to( @classmethod def from_properties( cls, - properties: JsonObject, + properties: Dict[str, Any], parent_item: Optional[PortalItem] = None, **kwargs: Any, ) -> NestedProperty: diff --git a/dcicutils/testing_utils.py b/dcicutils/testing_utils.py index 0e89b387b..dfc222b1d 100644 --- a/dcicutils/testing_utils.py +++ b/dcicutils/testing_utils.py @@ -14,6 +14,14 @@ def patch_context( module: Optional[ModuleType] = None, **kwargs, ) -> Iterator[mock.MagicMock]: + """Mock out the given object. + + Essentially mock.patch_object with some hacks to enable linting + on the object to patch instead of providing as a string. + + Depending on import structure, adding the module to patch may be + required. + """ if isinstance(to_patch, property): to_patch = to_patch.fget new_callable = mock.PropertyMock @@ -27,17 +35,3 @@ def patch_context( if return_value != AN_UNLIKELY_RETURN_VALUE: mocked_item.return_value = return_value yield mocked_item -# -# -# -# @contextmanager -# def patch_context( -# object_to_patch: object, -# attribute_to_patch: str, -# return_value: Optional[Any] = None, -# **kwargs, -# ) -> Iterator[mock.MagicMock]: -# with mock.patch.object(object_to_patch, attribute_to_patch, **kwargs) as mocked_item: -# if return_value is not None: -# mocked_item.return_value = return_value -# yield mocked_item diff --git a/test/test_item_model_utils.py b/test/test_item_model_utils.py index 7ab23e3b0..69d8b8c0c 100644 --- a/test/test_item_model_utils.py +++ b/test/test_item_model_utils.py @@ -1,13 +1,12 @@ from contextlib import contextmanager from random import random -from typing import Any, Iterator, Optional, Tuple, Union +from typing import Any, Dict, Iterator, Optional, Tuple, Union from unittest import mock import pytest from dcicutils import item_model_utils as item_models_module from dcicutils.item_model_utils import ( - JsonObject, LinkTo, get_item_identifier, get_link_to, @@ -97,7 +96,7 @@ def mock_portal_item(): def get_portal_item( - properties: Optional[JsonObject] = None, + properties: Optional[Dict[str, Any]] = None, fetch_links: Optional[bool] = False, auth: Optional[bool] = None, use_defaults: Optional[bool] = True, @@ -151,7 +150,7 @@ def test_get_link_to( ({"uuid": "foo"}, "foo"), ], ) -def test_get_item_identifier(item: Union[str, JsonObject], expected: str) -> None: +def test_get_item_identifier(item: Union[str, Dict[str, Any]], expected: str) -> None: result = get_item_identifier(item) assert result == expected @@ -160,28 +159,28 @@ class TestPortalItem: @pytest.mark.parametrize( "properties,expected", [({}, ""), (SOME_ITEM_PROPERTIES, SOME_UUID)] ) - def test_get_uuid(self, properties: JsonObject, expected: str) -> None: + def test_get_uuid(self, properties: Dict[str, Any], expected: str) -> None: portal_item = get_portal_item(properties=properties) assert portal_item.get_uuid() == expected @pytest.mark.parametrize( "properties,expected", [({}, ""), (SOME_ITEM_PROPERTIES, SOME_AT_ID)] ) - def test_get_at_id(self, properties: JsonObject, expected: str) -> None: + def test_get_at_id(self, properties: Dict[str, Any], expected: str) -> None: portal_item = get_portal_item(properties=properties) assert portal_item.get_at_id() == expected @pytest.mark.parametrize( "properties,expected", [({}, ""), (SOME_ITEM_PROPERTIES, SOME_ACCESSION)] ) - def test_get_accession(self, properties: JsonObject, expected: str) -> None: + def test_get_accession(self, properties: Dict[str, Any], expected: str) -> None: portal_item = get_portal_item(properties=properties) assert portal_item.get_accession() == expected @pytest.mark.parametrize( "properties,expected", [({}, []), (SOME_ITEM_PROPERTIES, SOME_TYPES)] ) - def test_get_types(self, properties: JsonObject, expected: str) -> None: + def test_get_types(self, properties: Dict[str, Any], expected: str) -> None: portal_item = get_portal_item(properties=properties) assert portal_item.get_types() == expected @@ -247,7 +246,7 @@ def test_get_item_via_auth(self) -> None: @pytest.mark.parametrize( "auth,expected", [({}, tuple()), (SOME_AUTH, HASHABLE_SOME_AUTH)] ) - def test_make_hashable_auth(self, auth: JsonObject, expected: Tuple) -> None: + def test_make_hashable_auth(self, auth: Dict[str, Any], expected: Tuple) -> None: portal_item = get_portal_item() result = portal_item._make_hashable_auth(auth) assert result == expected @@ -256,7 +255,7 @@ def test_make_hashable_auth(self, auth: JsonObject, expected: Tuple) -> None: "hashable_auth,expected", [(tuple(), {}), (HASHABLE_SOME_AUTH, SOME_AUTH)] ) def test_undo_make_hashable_auth( - self, hashable_auth: Tuple, expected: JsonObject + self, hashable_auth: Tuple, expected: Dict[str, Any] ) -> None: portal_item = get_portal_item() result = portal_item._undo_make_hashable_auth(hashable_auth) @@ -266,7 +265,7 @@ def test_undo_make_hashable_auth( "raise_exception,expected", [(False, SOME_ITEM_PROPERTIES), (True, {})] ) def test_get_and_cache_item_via_auth( - self, raise_exception: bool, expected: JsonObject + self, raise_exception: bool, expected: Dict[str, Any] ) -> None: side_effect = Exception if raise_exception else None random_add_on = str(random()) # To differentiate parametrized calls @@ -299,7 +298,7 @@ def test_get_and_cache_item_via_auth( ], ) def test_from_identifier_and_existing_item( - self, auth: JsonObject, fetch_links: bool, exception_expected: bool + self, auth: Dict[str, Any], fetch_links: bool, exception_expected: bool ) -> None: identifier = SOME_UUID portal_item = get_portal_item( @@ -330,7 +329,8 @@ def test_from_properties_and_existing_item(self) -> None: def get_nested_property( - properties: Optional[JsonObject] = None, parent_item: Optional[PortalItem] = None + properties: Optional[Dict[str, Any]] = None, + parent_item: Optional[PortalItem] = None, ) -> NestedProperty: properties = properties or SOME_ITEM_PROPERTIES return NestedProperty(properties=properties, parent_item=parent_item) From ed82c787b81d9af1dbbae36d615c9b426c4a5790 Mon Sep 17 00:00:00 2001 From: Douglas Rioux Date: Thu, 3 Aug 2023 15:42:23 -0400 Subject: [PATCH 7/7] Bump beta version --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 86c3d916e..ce4c20c21 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "dcicutils" -version = "7.6.0" +version = "7.7.0.1b1" description = "Utility package for interacting with the 4DN Data Portal and other 4DN resources" authors = ["4DN-DCIC Team "] license = "MIT"