azad.ainetwork.errors Module ¶

azad.ainetwork.errors ¶

Classes ¶

AINetworkError ¶

AINetworkError(original_exception: Exception = None)

Bases: Exception

Base error class for AINetwork exceptions.

Source code in azad/ainetwork/errors.py

def __init__(self, original_exception: Exception = None): # type: ignore

    if original_exception:
        full_message = f"{self.default_message}: {original_exception}"
    else:
        full_message = self.default_message
    self.error_message = full_message
    super().__init__(full_message)
    self.original_exception = original_exception

Attributes ¶

default_message `class-attribute` `instance-attribute` ¶

default_message = 'An unknown error occurred in the AI network layer.'

error_message `instance-attribute` ¶

error_message = full_message

original_exception `instance-attribute` ¶

original_exception = original_exception

AIAuthenticationError ¶

AIAuthenticationError(original_exception: Exception = None)

Bases: AINetworkError

Raised when authentication fails.

Source code in azad/ainetwork/errors.py

def __init__(self, original_exception: Exception = None): # type: ignore

    if original_exception:
        full_message = f"{self.default_message}: {original_exception}"
    else:
        full_message = self.default_message
    self.error_message = full_message
    super().__init__(full_message)
    self.original_exception = original_exception

Attributes ¶

default_message `class-attribute` `instance-attribute` ¶

default_message = 'Authentication failed in the AI network layer'

error_message `instance-attribute` ¶

error_message = full_message

original_exception `instance-attribute` ¶

original_exception = original_exception

AIInvalidRequestError ¶

AIInvalidRequestError(original_exception: Exception = None)

Bases: AINetworkError

Raised when the request is malformed or invalid.

Source code in azad/ainetwork/errors.py

def __init__(self, original_exception: Exception = None): # type: ignore

    if original_exception:
        full_message = f"{self.default_message}: {original_exception}"
    else:
        full_message = self.default_message
    self.error_message = full_message
    super().__init__(full_message)
    self.original_exception = original_exception

Attributes ¶

default_message `class-attribute` `instance-attribute` ¶

default_message = 'Invalid request in the AI network layer'

error_message `instance-attribute` ¶

error_message = full_message

original_exception `instance-attribute` ¶

original_exception = original_exception

AIRateLimitError ¶

AIRateLimitError(original_exception: Exception = None)

Bases: AINetworkError

Raised when rate limits have been exceeded.

Source code in azad/ainetwork/errors.py

def __init__(self, original_exception: Exception = None): # type: ignore

    if original_exception:
        full_message = f"{self.default_message}: {original_exception}"
    else:
        full_message = self.default_message
    self.error_message = full_message
    super().__init__(full_message)
    self.original_exception = original_exception

Attributes ¶

default_message `class-attribute` `instance-attribute` ¶

default_message = 'Rate limit exceeded in the AI network layer'

error_message `instance-attribute` ¶

error_message = full_message

original_exception `instance-attribute` ¶

original_exception = original_exception

AIServiceUnavailableError ¶

AIServiceUnavailableError(original_exception: Exception = None)

Bases: AINetworkError

Raised when the underlying service is temporarily unavailable.

Source code in azad/ainetwork/errors.py

def __init__(self, original_exception: Exception = None): # type: ignore

    if original_exception:
        full_message = f"{self.default_message}: {original_exception}"
    else:
        full_message = self.default_message
    self.error_message = full_message
    super().__init__(full_message)
    self.original_exception = original_exception

Attributes ¶

default_message `class-attribute` `instance-attribute` ¶

default_message = 'Service is unavailable in the AI network layer'

error_message `instance-attribute` ¶

error_message = full_message

original_exception `instance-attribute` ¶

original_exception = original_exception

AITimeoutError ¶

AITimeoutError(original_exception: Exception = None)

Bases: AINetworkError

Raised when the request or streaming times out.

Source code in azad/ainetwork/errors.py

def __init__(self, original_exception: Exception = None): # type: ignore

    if original_exception:
        full_message = f"{self.default_message}: {original_exception}"
    else:
        full_message = self.default_message
    self.error_message = full_message
    super().__init__(full_message)
    self.original_exception = original_exception

Attributes ¶

default_message `class-attribute` `instance-attribute` ¶

default_message = 'Operation timed out in the AI network layer'

error_message `instance-attribute` ¶

error_message = full_message

original_exception `instance-attribute` ¶

original_exception = original_exception

AIConnectionError ¶

AIConnectionError(original_exception: Exception = None)

Bases: AINetworkError

Raised when a network connection error occurs.

Source code in azad/ainetwork/errors.py

def __init__(self, original_exception: Exception = None): # type: ignore

    if original_exception:
        full_message = f"{self.default_message}: {original_exception}"
    else:
        full_message = self.default_message
    self.error_message = full_message
    super().__init__(full_message)
    self.original_exception = original_exception

Attributes ¶

default_message `class-attribute` `instance-attribute` ¶

default_message = 'Network connection error in the AI network layer'

error_message `instance-attribute` ¶

error_message = full_message

original_exception `instance-attribute` ¶

original_exception = original_exception

AIInvalidResponseError ¶

AIInvalidResponseError(original_exception: Exception = None)

Bases: AINetworkError

Raised when an invalid response is received from the API.

Source code in azad/ainetwork/errors.py

def __init__(self, original_exception: Exception = None): # type: ignore

    if original_exception:
        full_message = f"{self.default_message}: {original_exception}"
    else:
        full_message = self.default_message
    self.error_message = full_message
    super().__init__(full_message)
    self.original_exception = original_exception

Attributes ¶

default_message `class-attribute` `instance-attribute` ¶

default_message = 'Invalid response received in the AI network layer'

error_message `instance-attribute` ¶

error_message = full_message

original_exception `instance-attribute` ¶

original_exception = original_exception

AIProviderError ¶

AIProviderError(original_exception: Exception = None)

Bases: AINetworkError

Raised when a provider-specific error occurs.

Source code in azad/ainetwork/errors.py

def __init__(self, original_exception: Exception = None): # type: ignore

    if original_exception:
        full_message = f"{self.default_message}: {original_exception}"
    else:
        full_message = self.default_message
    self.error_message = full_message
    super().__init__(full_message)
    self.original_exception = original_exception

Attributes ¶

default_message `class-attribute` `instance-attribute` ¶

default_message = 'Provider-specific error in the AI network layer'

error_message `instance-attribute` ¶

error_message = full_message

original_exception `instance-attribute` ¶

original_exception = original_exception

AIInsufficientCreditsError ¶

AIInsufficientCreditsError(original_exception: Exception = None)

Bases: AINetworkError

Raised when the user doesn't have enough credits.

Source code in azad/ainetwork/errors.py

def __init__(self, original_exception: Exception = None): # type: ignore

    if original_exception:
        full_message = f"{self.default_message}: {original_exception}"
    else:
        full_message = self.default_message
    self.error_message = full_message
    super().__init__(full_message)
    self.original_exception = original_exception

Attributes ¶

default_message `class-attribute` `instance-attribute` ¶

default_message = 'Insufficient credits to process the request'

error_message `instance-attribute` ¶

error_message = full_message

original_exception `instance-attribute` ¶

original_exception = original_exception

AIUserNotFoundError ¶

AIUserNotFoundError(original_exception: Exception = None)

Bases: AINetworkError

Raised when the user is not found.

Source code in azad/ainetwork/errors.py

def __init__(self, original_exception: Exception = None): # type: ignore

    if original_exception:
        full_message = f"{self.default_message}: {original_exception}"
    else:
        full_message = self.default_message
    self.error_message = full_message
    super().__init__(full_message)
    self.original_exception = original_exception

Attributes ¶

default_message `class-attribute` `instance-attribute` ¶

default_message = 'User not found in the Kodu tunnel network layer db'

error_message `instance-attribute` ¶

error_message = full_message

original_exception `instance-attribute` ¶

original_exception = original_exception

AINetwork ¶

Functions ¶

complete `async` ¶

complete(prompt, **kwargs)

Wraps litellm.exceptions.accompletion. It first makes the initial call (one line) and then loops over the asynchronous iterator to process streaming tokens.

All litellm errors are caught and re-raised as custom AINetwork errors, with the default error message defined in the error class and the original exception appended automatically.

Source code in azad/ainetwork/errors.py

async def complete(self, prompt, **kwargs):
    """
    Wraps litellm.exceptions.accompletion. It first makes the initial call (one line) and then
    loops over the asynchronous iterator to process streaming tokens.

    All litellm errors are caught and re-raised as custom AINetwork errors,
    with the default error message defined in the error class and the original exception
    appended automatically.
    """

Functions ¶

create ¶

create(err: Exception) -> AINetworkError

Creates a wrapping error for a litellm error using a match statement.

Source code in azad/ainetwork/errors.py

def create(err: Exception) -> AINetworkError:
    """Creates a wrapping error for a litellm error using a match statement."""
    match err:
        # Authentication failures
        case litellm.exceptions.AuthenticationError():
            return AIAuthenticationError(err)
        # Bad requests – note: InvalidRequestError is deprecated in favor of BadRequestError
        case litellm.exceptions.BadRequestError() | litellm.exceptions.InvalidRequestError():
            return AIInvalidRequestError(err)
        # Rate limiting errors
        case litellm.exceptions.RateLimitError():
            return AIRateLimitError(err)
        # Service unavailable errors
        case litellm.exceptions.ServiceUnavailableError():
            return AIServiceUnavailableError(err)
        # Timeout errors (e.g. APITimeoutError)
        case litellm.exceptions.Timeout():
            return AITimeoutError(err)
        # Connection-related errors (APIConnectionError or ConnectionError)
        case litellm.exceptions.APIConnectionError():
            return AIConnectionError(err)
        # Errors due to invalid responses (e.g. response schema mismatches)
        case litellm.exceptions.APIResponseValidationError():
            return AIInvalidResponseError(err)
        # Insufficient credits error
        case AIInsufficientCreditsError():
            return AIInsufficientCreditsError(err)  # Return the error as is, no need to wrap it
        # Default fallback: if nothing else matches, return a generic network error.
        case _:
            return AINetworkError(err)