Modeling errors with python using exceptions

example.py

class TokenNotPresent(Exception):
  """
  The token is missing from the URL
  """
  pass

class ValidationError(Exception):
  """
  This is an exception from a library like [Pydantic](https://docs.pydantic.dev/latest/)
  """
  pass

CustomerId = str

ProductCode = str

CustomerAccountId = str

@dataclass
class CustomerInformation:
  customer_id: CustomerId
  customer_account_id: CustommerAccountId
  product_code: ProductCode

class ExternalApiError(Exception):
  pass

class UnknownError(Exception):
  pass

def exchange_customer_information(token):

  try:
    response = external_client.exchange_customer_information(token)
    # This will throw a validation error
    return validate_customer_exchange_schema(response)
  except ClientError as e:
    error_code = e.response["Error"]["Code"]
    match error_code:
      case "ExpiredCredentials":
        raise ExternalApiError(error_code="expired_credentials") from e
      case "InvalidCredentials":
        raise ExternalApiError(error_code="invalid_credentials") from e
      case "InvalidParameter":
        raise ExternalApiError(error_code="invalid_parameter") from e
      case _:
        raise ExternalApiError() from e
  except Exception as e:
      raise UnknownError() from e

def validate_subscriptions_schema(data):
  """
  Validates the response data matches the expected schema.
  Raises a `ValidationError` when the schema does not match.
  """
  pass


def find_active_subscription(customer_info, subscriptions):
  """
  Finds a subscription that is active (i.e. end date is in the future) for the specified product
  Raises `SubscriptionMissing` when there are no subscriptions or `SubscriptionExpired` if there is no active subscription.
  """
  pass


def fetch_subscription(customer_information):

  try:
    response = external_client.fetch_subscriptions(customer_information)
    # This will throw a validation error
    subscriptions = validate_subscriptions_schema(response.data)

    # Will raise SubscriptionExpired or SubscriptionMissing
    return find_active_subscription(customer_info, subscriptions)
  except ClientError as e:
    # same as previous function
  except Exception as e:
    raise UnknownError() from e

def persist_customer_information(customer_info):
  """
  Stores the customer information into the database for future usage
  """
  pass


def request_handler(request: Request):
  try:
    token = extract_token_from_url(request.url)
  except TokenNotPresent:
    logger.info("The URL token parameter was not present in", request.url)
    return redirect_with_error("TokenMissing")
  except ValidationError:
    logger.error("The URL token was present, but the length was invalid", request.url)
    return redirect_with_error("TokenInvalid")

  try:
    customer_info = exchange_customer_information(token)
  except ValidationError as val_error:
    logger.error("There was a problem with the response schema of the exchange user information API", val_error)
    return redirect_with_error("InternalError")
  except ExternalApiError as external_e:
    logger.error("There was a problem calling the exchange user information API", external_e)
    return redirect_with_error("InternalError")
  except UnknownError as unknown:
    logger.error("There was an unknown problem calling the exchange user information API", unknown)
    return redirect_with_error("UnknownError")
  
  try:
    active_subscription = fetch_subscription(customer_info)
  except SubscriptionMissing:
    logger.error("There are no subscription for the product", customer_info.product_code)
    return redirect_with_error("SubscriptionMissing")
  except ValidationError as val_error:
    logger.error("There was a problem with the response schema of the fetch subscription API", val_error)
    return redirect_with_error("InternalError")
  except ExternalApiError as xae:
    logger.error("There was a problem calling fetching subscriptions API", xae)
    return redirect_with_error("InternalError")
  except UnknownError as unknown:
    logger.error("There was an unknown problem calling fetching subscriptions API", unknown)
    return redirect_with_error("UnknownError")

  try:
    persist_customer_information(customer_info)
  except Exception as db_e:
    logger.error("There was a problem calling the database", db_e)
    return redirect_with_error("UnknownError")

  return redirect_with(active_subscription)