def _process_license()

in src/lambdas/custodian_license_updater/handler.py [0:0]

• 38 lines of code
• 10 McCabe index (conditional complexity)

    def _process_license(self, lic: License) -> None:
        """
        Can raise LicenseSyncException
        :param lic:
        :return:
        """
        _LOG.info('Making request to LM ')
        data = self.license_manager_service.cl.sync_license(
            license_key=lic.license_key,
            customer=lic.customer
        )
        if not data:
            raise LicenseSyncError('Request to the License manager failed')
        new_rulesets = [
            self._process_ruleset(i, lic) for i in data.get('rulesets') or ()
        ]

        old_rulesets_ids = set(lic.ruleset_ids)
        new_rulesets_ids = {r.license_manager_id for r in new_rulesets}
        for to_remove in old_rulesets_ids - new_rulesets_ids:
            _LOG.info(f'Removing old licensed ruleset {to_remove}')
            item = self.ruleset_service.by_lm_id(
                to_remove, attributes_to_get=[Ruleset.id]
            )
            if item:
                self.ruleset_service.delete(item)

        _LOG.info('Saving new rulesets')
        self.ruleset_service.batch_save(new_rulesets)

        event_driven = data.get('event_driven') or {}
        if quota := event_driven.get('quota'):  # fixing some bug in cslm
            event_driven['quota'] = int(quota)
        _LOG.info('Saving new license')
        self.license_service.update(
            item=lic,
            description=data.get('description'),
            allowance=data.get('allowance'),
            customers=data.get('customers'),
            event_driven=event_driven,
            rulesets=[r.license_manager_id for r in new_rulesets],
            latest_sync=utc_iso(),
            valid_until=data.get('valid_until')
        )

        _LOG.info('Updating license activation')
        self.update_license_activation(lic, new_rulesets)