(self)
return Subscription(
            account_id=real_response[b"account_id"].decode(),
            subscription_blocker=subscription_blocker,
        )
Get subscription information for installation.
Get subscription information for installation.
Get subscription information for installation .
async def get_subscription(self) -> Optional[Subscription]:
        """
        Get subscription information for installation.
        """
        from kodiak.queue import get_redis

        redis = await get_redis()
        res = await redis.hgetall(
            f"kodiak:subscription:{self.installation_id}".encode()
        )
        if not res:
            return None
        real_response = await res.asdict()
        if not real_response:
            return None
        subscription_blocker_kind = (
            real_response.get(b"subscription_blocker") or b""
        ).decode()
        subscription_blocker: Optional[
            Union[SubscriptionExpired, TrialExpired, SeatsExceeded]
        ] = None
        if subscription_blocker_kind == "seats_exceeded":
            # to be backwards compatible we must handle the case of `data` missing.
            try:
                subscription_blocker = SeatsExceeded.parse_raw(
                    # Pydantic says it doesn't allow Nones, but passing a None
                    # raises a ValidationError which is fine.
                    real_response.get(b"data")  # type: ignore
                )
            except pydantic.ValidationError:
                logger.exception("failed to parse seats_exceeded data", exc_info=True)
                subscription_blocker = SeatsExceeded(allowed_user_ids=[])
        if subscription_blocker_kind == "trial_expired":
            subscription_blocker = TrialExpired()
        if subscription_blocker_kind == "subscription_expired":
            subscription_blocker = SubscriptionExpired()
        return Subscription(
            account_id=real_response[b"account_id"].decode(),
            subscription_blocker=subscription_blocker,
        )
async def get_subscription ( self ) -> Optional [ Subscription ] : from kodiak . queue import get_redis redis = await get_redis ( ) res = await redis . hgetall ( f"kodiak:subscription:{self.installation_id}" . encode ( ) ) if not res : return None real_response = await res . asdict ( ) if not real_response : return None subscription_blocker_kind = ( real_response . get ( b"subscription_blocker" ) or b"" ) . decode ( ) subscription_blocker : Optional [ Union [ SubscriptionExpired , TrialExpired , SeatsExceeded ] ] = None if subscription_blocker_kind == "seats_exceeded" : # to be backwards compatible we must handle the case of `data` missing. try : subscription_blocker = SeatsExceeded . parse_raw ( # Pydantic says it doesn't allow Nones, but passing a None # raises a ValidationError which is fine. real_response . get ( b"data" ) # type: ignore ) except pydantic . ValidationError : logger . exception ( "failed to parse seats_exceeded data" , exc_info = True ) subscription_blocker = SeatsExceeded ( allowed_user_ids = [ ] ) if subscription_blocker_kind == "trial_expired" : subscription_blocker = TrialExpired ( ) if subscription_blocker_kind == "subscription_expired" : subscription_blocker = SubscriptionExpired ( ) return Subscription ( account_id = real_response [ b"account_id" ] . decode ( ) , subscription_blocker = subscription_blocker , )