61 lines
1.9 KiB
Python
61 lines
1.9 KiB
Python
from typing import Union, Optional
|
|
from core.models.session.SessionProfile import SessionProfile
|
|
from core.models.system.SystemProfile import SystemProfile
|
|
from core.Errors import MissingSubscriptionError, InvalidSubscriptionError
|
|
from core.services.WebServiceApiService import WebServiceApiService
|
|
from core.controllers.ConnectionController import ConnectionController
|
|
from core.observers.ConnectionObserver import ConnectionObserver
|
|
|
|
def activate_subscription(
|
|
profile: Union[SessionProfile, SystemProfile],
|
|
connection_observer: Optional[ConnectionObserver] = None
|
|
) -> bool:
|
|
"""
|
|
Purpose:
|
|
Ensure subscription is ready.
|
|
|
|
Features:
|
|
Idempotent.
|
|
Checks local first
|
|
|
|
Confusion:
|
|
This returns True both if was already active or just activated.
|
|
|
|
Returns:
|
|
Returns True if subscription was already active.
|
|
Returns True if subscription was just activated.
|
|
|
|
Errors:
|
|
True.
|
|
Raises if subscription is missing or invalid. (not false)
|
|
"""
|
|
|
|
if not profile.has_subscription():
|
|
raise MissingSubscriptionError()
|
|
|
|
# Already activated—nothing to do
|
|
if profile.subscription.has_been_activated():
|
|
return True
|
|
|
|
# Fetch and activate
|
|
subscription = ConnectionController.with_preferred_connection(
|
|
profile.subscription.billing_code,
|
|
task=WebServiceApiService.get_subscription,
|
|
connection_observer=connection_observer
|
|
)
|
|
|
|
if subscription is None:
|
|
raise InvalidSubscriptionError()
|
|
|
|
profile.subscription = subscription
|
|
profile.save()
|
|
return True
|
|
|
|
|
|
def is_subscription_ready(profile: Union[SessionProfile, SystemProfile]) -> bool:
|
|
"""Check if subscription exists and is activated (no side effects)."""
|
|
return (
|
|
profile.has_subscription()
|
|
and profile.subscription.has_been_activated()
|
|
)
|
|
|