partnercenter/azext_partnercenter/clients/offer_listing_client.py

# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
# pylint: disable=line-too-long
# pylint: disable=protected-access

from azext_partnercenter.models import ListingContact, Listing, ListingUri, Resource
from azext_partnercenter.vendored_sdks.v1.partnercenter.models import (
    MicrosoftIngestionApiModelsListingsAzureListing,
    MicrosoftIngestionApiModelsListingsListingContact,
    MicrosoftIngestionApiModelsListingsListingUri)
from azext_partnercenter.clients import OfferClient, PlanClient
from ._base_client import BaseClient


class OfferListingClient(BaseClient):
    def __init__(self, cli_ctx, *_):
        super().__init__(cli_ctx, *_)
        self._offer_client = OfferClient(cli_ctx, *_)
        self._plan_client = PlanClient(cli_ctx, *_)

    def get_plan_listing(self, offer_external_id, plan_external_id):
        product_listing_branches = self._get_product_listing_branches(offer_external_id)
        if not product_listing_branches:
            return None
        product = self._offer_client.get(offer_external_id)
        if product is None:
            return None
        product_id = product.resource.durable_id
        plan = self._plan_client.find_by_external_id(offer_external_id, plan_external_id)
        if plan is None:
            return None
        plan_durable_id = plan._resource.durable_id

        branch = next(filter(lambda b: b.variant_id == plan_durable_id, product_listing_branches), None)
        if branch is None:
            return None

        instance_id = branch.current_draft_instance_id
        result = self._sdk.listing_client.products_product_id_listings_get_by_instance_id_instance_i_dinstance_id_get(
            product_id,
            instance_id,
            self._get_access_token())
        listing = result.value[0]
        return Listing(
            description=listing.description,
            title=listing.title,
            summary=listing.summary if hasattr(listing, 'summary') else '',
            language_code=listing.language_code if hasattr('language_code', 'summary') else '',
            short_description=listing.short_description if hasattr('short_description', 'summary') else '',
            contacts=list(map(lambda c: ListingContact(**c.to_dict()), listing.listing_contacts)),
            uris=list(map(lambda c: ListingUri(**c.to_dict()), listing.listing_uris)),
            odata_etag=listing.odata_etag,
            offer_id=offer_external_id,
            resource=Resource(id=listing.id, type=listing.resource_type)
        )

    def update(self, offer_external_id, parameters: Listing):
        """Update an offer listing"""
        return self.create_or_update(offer_external_id, parameters)

    def create_or_update(self, offer_external_id, listing_model: Listing, plan_external_id=None):
        offer = self._offer_client.get(offer_external_id)
        if offer is None:
            return None
        product_id = offer.resource.durable_id

        listing = None
        if plan_external_id is None:
            listing = self._offer_client.get_listing(offer_external_id)
        else:
            listing = self.get_plan_listing(offer_external_id, plan_external_id)

        if listing is None:
            return None

        listing_id = listing._resource.durable_id

        listing_contacts = self._get_api_listing_contacts(listing_model)
        listing_uris = self._get_api_listing_uris(listing_model)

        updated_listing = MicrosoftIngestionApiModelsListingsAzureListing(
            resource_type='AzureListing',
            description=listing_model.description,
            title=listing_model.title,
            summary=listing_model.summary,
            short_description=listing_model.short_description,
            odata_etag=listing.odata_etag,
            listing_uris=listing_uris,
            listing_contacts=listing_contacts)
        update_result = self._sdk.listing_client.products_product_id_listings_listing_id_put(
            product_id,
            listing_id,
            self._get_access_token(),
            microsoft_ingestion_api_models_listings_azure_listing=updated_listing)
        return Listing(
            description=update_result.description,
            title=update_result.title,
            summary=update_result.summary,
            language_code=update_result.language_code,
            short_description=update_result.short_description,
            contacts=list(map(lambda c: ListingContact(**c.to_dict()), update_result.listing_contacts)),
            uris=list(map(
                lambda c: ListingUri(type=c.type, subtype=c.subtype, uri=c.uri, display_text=c.display_text), update_result.listing_uris)
            ),
            odata_etag=update_result.odata_etag,
            resource=Resource(id=update_result.id, type=update_result.resource_type)
        )

    def get_contacts(self, offer_external_id):
        listing = self.get(offer_external_id)
        return listing.contacts

    def update_contacts(self, offer_external_id, parameters=list[ListingContact]):
        listing = self.get(offer_external_id)
        listing.contacts = parameters
        result = self.create_or_update(offer_external_id, listing)
        return result.contacts

    def get_uris(self, offer_external_id):
        listing = self.get(offer_external_id)
        return listing.uris

    def update_uris(self, offer_external_id, parameters=list[ListingUri]):
        listing = self.get(offer_external_id)
        listing.uris = parameters
        result = self.create_or_update(offer_external_id, listing)
        return result.uris

    def _get_api_listing_uris(self, listing_model: Listing):
        return list(map(lambda u: MicrosoftIngestionApiModelsListingsListingUri(
            type=u.type if u.type is not None else '',
            subtype=u.subtype if u.subtype is not None else '',
            display_text=u.display_text if u.display_text is not None else '',
            uri=u.uri if u.uri is not None else ''),
            listing_model.uris))

    def _get_api_listing_contacts(self, listing_model: Listing):
        return list(map(lambda c: MicrosoftIngestionApiModelsListingsListingContact(
            type=c.type,
            email=c.email,
            name=c.name,
            phone=c.phone,
            uri=c.uri),
            listing_model.contacts))

    def delete_offer_listing(self, product_external_id):
        product = self._offer_client.get(product_external_id)
        if product is None:
            return None
        product_id = product.resource.durable_id

        listing = self._offer_client.get_listing(product_external_id)
        if listing is None:
            return None
        listing_id = listing._resource.durable_id

        return self._sdk.listing_client.products_product_id_listings_listing_id_delete(product_id, listing_id, self._get_access_token())

    def delete_listing_contact(self, offer_external_id, contact: ListingContact):
        listing = self._offer_client.get_listing(offer_external_id)
        try:
            listing.contacts.remove(contact)
        except ValueError:
            return None
        return self.create_or_update(offer_external_id, listing)

    def delete_uri(self, offer_external_id, uri: ListingUri):
        listing = self._offer_client.get_listing(offer_external_id)
        try:
            listing.uris.remove(uri)
        except ValueError:
            return None

        return self.create_or_update(offer_external_id, listing)

    def delete_latest_listing_uri(self, product_external_id):
        listing = self._offer_client.get_listing(product_external_id)
        if len(listing.uris) > 0:
            del listing.uris[0]
        return self.create_or_update(product_external_id, listing)

    def get(self, offer_external_id):
        return self._offer_client.get_listing(offer_external_id)

    def _get_product_listing_branches(self, product_external_id):
        offer = self._offer_client.get(product_external_id)
        product_id = offer.resource.durable_id

        module = 'Listing'
        branches = self._sdk.branches_client.products_product_id_branches_get_by_module_modulemodule_get(
            product_id, module, self._get_access_token())

        return list(filter(lambda x: hasattr(x, 'variant_id'), branches.value))