Add error classes

Move version check to debug

g4f/Provider/retry_provider.py

@@ -6,6 +6,7 @@ from typing import List, Type, Dict
 from ..typing import CreateResult, Messages
 from .base_provider import BaseProvider, AsyncProvider
 from .. import debug
+from ..errors import RetryProviderError, RetryNoProviderError
 
 
 class RetryProvider(AsyncProvider):
@@ -84,8 +85,8 @@ class RetryProvider(AsyncProvider):
     
     def raise_exceptions(self) -> None:
         if self.exceptions:
-            raise RuntimeError("\n".join(["RetryProvider failed:"] + [
+            raise RetryProviderError("RetryProvider failed:\n" + "\n".join([
-                f"{p}: {self.exceptions[p].__class__.__name__}: {self.exceptions[p]}" for p in self.exceptions
+                f"{p}: {exception.__class__.__name__}: {exception}" for p, exception in self.exceptions.items()
             ]))
         
-        raise RuntimeError("RetryProvider: No provider found")
+        raise RetryNoProviderError("No provider found")

g4f/__init__.py

@@ -1,61 +1,35 @@
 from __future__ import annotations
 
 import os
-from requests import get
-from importlib.metadata import version as get_package_version, PackageNotFoundError
-from subprocess import check_output, CalledProcessError, PIPE
 
+from .errors   import *
 from .models   import Model, ModelUtils, _all_models
-from .Provider import BaseProvider, AsyncGeneratorProvider, RetryProvider
+from .Provider import BaseProvider, AsyncGeneratorProvider, RetryProvider, ProviderUtils
 from .typing   import Messages, CreateResult, AsyncResult, Union, List
 from .         import debug
 
-def get_version() -> str:
-    # Read from package manager
-    try:
-        return get_package_version("g4f")
-    except PackageNotFoundError:
-        pass
-    # Read from docker environment
-    current_version = os.environ.get("G4F_VERSION")
-    if current_version:
-        return current_version
-    # Read from git repository
-    try:
-        command = ["git", "describe", "--tags", "--abbrev=0"]
-        return check_output(command, text=True, stderr=PIPE).strip()
-    except CalledProcessError:
-        pass
-    
-def get_lastet_version() -> str:
-    response = get("https://pypi.org/pypi/g4f/json").json()
-    return response["info"]["version"]
-
-def check_pypi_version() -> None:
-    try:
-        version = get_version()
-        latest_version = get_lastet_version()
-    except Exception as e:
-        print(f'Failed to check g4f pypi version: {e}')
-    if version != latest_version:
-        print(f'New pypi version: {latest_version} (current: {version}) | pip install -U g4f')
-
 def get_model_and_provider(model    : Union[Model, str], 
-                           provider : Union[type[BaseProvider], None], 
+                           provider : Union[type[BaseProvider], str, None], 
                            stream   : bool,
                            ignored  : List[str] = None,
                            ignore_working: bool = False,
                            ignore_stream: bool = False) -> tuple[Model, type[BaseProvider]]:
     if debug.version_check:
-        check_pypi_version()
         debug.version_check = False
-    
+        debug.check_pypi_version()
+
+    if isinstance(provider, str):
+        if provider in ProviderUtils.convert:
+            provider = ProviderUtils.convert[provider]
+        else:
+            raise ProviderNotFoundError(f'Provider not found: {provider}')
+
     if isinstance(model, str):
         if model in ModelUtils.convert:
             model = ModelUtils.convert[model]
         else:
-            raise ValueError(f'The model: {model} does not exist')
+            raise ModelNotFoundError(f'The model: {model} does not exist')
-
+        
     if not provider:
         provider = model.best_provider
 
@@ -63,14 +37,14 @@ def get_model_and_provider(model    : Union[Model, str],
         provider.providers = [p for p in provider.providers if p.__name__ not in ignored]
 
     if not provider:
-        raise RuntimeError(f'No provider found for model: {model}')
+        raise ProviderNotFoundError(f'No provider found for model: {model}')
-
+    
     if not provider.working and not ignore_working:
-        raise RuntimeError(f'{provider.__name__} is not working')
+        raise ProviderNotWorkingError(f'{provider.__name__} is not working')
-
+    
     if not ignore_stream and not provider.supports_stream and stream:
-        raise ValueError(f'{provider.__name__} does not support "stream" argument')
+        raise StreamNotSupportedError(f'{provider.__name__} does not support "stream" argument')
-
+    
     if debug.logging:
         print(f'Using {provider.__name__} provider')
 
@@ -80,7 +54,7 @@ class ChatCompletion:
     @staticmethod
     def create(model    : Union[Model, str],
                messages : Messages,
-               provider : Union[type[BaseProvider], None] = None,
+               provider : Union[type[BaseProvider], str, None] = None,
                stream   : bool = False,
                auth     : Union[str, None] = None,
                ignored  : List[str] = None, 
@@ -91,15 +65,15 @@ class ChatCompletion:
         model, provider = get_model_and_provider(model, provider, stream, ignored, ignore_working, ignore_stream_and_auth)
 
         if not ignore_stream_and_auth and provider.needs_auth and not auth:
-            raise ValueError(
+            raise AuthenticationRequiredError(f'{provider.__name__} requires authentication (use auth=\'cookie or token or jwt ...\' param)')
-                f'{provider.__name__} requires authentication (use auth=\'cookie or token or jwt ...\' param)')
 
         if auth:
             kwargs['auth'] = auth
-            
+        
-        proxy = os.environ.get("G4F_PROXY")
+        if "proxy" not in kwargs:
-        if proxy and "proxy" not in kwargs:
+            proxy = os.environ.get("G4F_PROXY")
-            kwargs['proxy'] = proxy
+            if proxy:
+                kwargs['proxy'] = proxy
 
         result = provider.create_completion(model.name, messages, stream, **kwargs)
         return result if stream else ''.join(result)
@@ -107,7 +81,7 @@ class ChatCompletion:
     @staticmethod
     async def create_async(model    : Union[Model, str],
                            messages : Messages,
-                           provider : Union[type[BaseProvider], None] = None,
+                           provider : Union[type[BaseProvider], str, None] = None,
                            stream   : bool = False,
                            ignored  : List[str] = None,
                            **kwargs) -> Union[AsyncResult, str]:
@@ -116,7 +90,7 @@ class ChatCompletion:
         if stream:
             if isinstance(provider, type) and issubclass(provider, AsyncGeneratorProvider):
                 return await provider.create_async_generator(model.name, messages, **kwargs)
-            raise ValueError(f'{provider.__name__} does not support "stream" argument in "create_async"')
+            raise StreamNotSupportedError(f'{provider.__name__} does not support "stream" argument in "create_async"')
 
         return await provider.create_async(model.name, messages, **kwargs)
 
@@ -136,9 +110,8 @@ class Completion:
             'text-davinci-002',
             'text-davinci-003'
         ]
-
         if model not in allowed_models:
-            raise Exception(f'ValueError: Can\'t use {model} with Completion.create()')
+            raise ModelNotAllowed(f'Can\'t use {model} with Completion.create()')
 
         model, provider = get_model_and_provider(model, provider, stream, ignored)
 

g4f/debug.py

@@ -1,2 +1,39 @@
+from os import environ
+from requests import get
+from importlib.metadata import version, PackageNotFoundError
+from subprocess import check_output, CalledProcessError, PIPE
+from .errors import VersionNotFoundError
+
 logging = False
-version_check = True
+version_check = True
+
+def get_version() -> str:
+    # Read from package manager
+    try:
+        return version("g4f")
+    except PackageNotFoundError:
+        pass
+    # Read from docker environment
+    current_version = environ.get("G4F_VERSION")
+    if current_version:
+        return current_version
+    # Read from git repository
+    try:
+        command = ["git", "describe", "--tags", "--abbrev=0"]
+        return check_output(command, text=True, stderr=PIPE).strip()
+    except CalledProcessError:
+        pass
+    raise VersionNotFoundError("Version not found")
+    
+def get_lastet_version() -> str:
+    response = get("https://pypi.org/pypi/g4f/json").json()
+    return response["info"]["version"]
+
+def check_pypi_version() -> None:
+    try:
+        version = get_version()
+        latest_version = get_lastet_version()
+    except Exception as e:
+        print(f'Failed to check g4f pypi version: {e}')
+    if version != latest_version:
+        print(f'New pypi version: {latest_version} (current: {version}) | pip install -U g4f')

g4f/errors.py

@@ -0,0 +1,26 @@
+class ProviderNotFoundError(Exception):
+    pass
+
+class ProviderNotWorkingError(Exception):
+    pass
+
+class StreamNotSupportedError(Exception):
+    pass
+
+class AuthenticationRequiredError(Exception):
+    pass
+
+class ModelNotFoundError(Exception):
+    pass
+
+class ModelNotAllowed(Exception):
+    pass
+
+class RetryProviderError(Exception):
+    pass
+
+class RetryNoProviderError(Exception):
+    pass
+
+class VersionNotFoundError(Exception):
+    pass

g4f/gui/server/backend.py

@@ -4,8 +4,9 @@ from g4f.Provider import __providers__
 import json
 from flask      import request, Flask
 from .internet  import get_search_message
+from g4f import debug
 
-g4f.debug.logging = True
+debug.logging = True
 
 class Backend_Api:
     def __init__(self, app: Flask) -> None:
@@ -52,8 +53,8 @@ class Backend_Api:
         
     def version(self):
         return {
-            "version": g4f.get_version(),
+            "version": debug.get_version(),
-            "lastet_version": g4f.get_lastet_version(),
+            "lastet_version": debug.get_lastet_version(),
         }
     
     def _gen_title(self):
@@ -68,28 +69,25 @@ class Backend_Api:
             messages[-1]["content"] = get_search_message(messages[-1]["content"])
         model = request.json.get('model')
         model = model if model else g4f.models.default
-        provider = request.json.get('provider').replace('g4f.Provider.', '')
+        provider = request.json.get('provider', '').replace('g4f.Provider.', '')
         provider = provider if provider and provider != "Auto" else None
-        if provider != None:
-            provider = g4f.Provider.ProviderUtils.convert.get(provider)
             
         def try_response():
             try:
-                response = g4f.ChatCompletion.create(
+                yield from g4f.ChatCompletion.create(
                     model=model,
                     provider=provider,
                     messages=messages,
                     stream=True,
                     ignore_stream_and_auth=True
                 )
-                yield from response
             except Exception as e:
                 print(e)
                 yield json.dumps({
                     'code'   : 'G4F_ERROR',
                     '_action': '_ask',
                     'success': False,
-                    'error'  : f'an error occurred {str(e)}'
+                    'error'  : f'{e.__class__.__name__}: {e}'
-                })      
+                })
 
         return self.app.response_class(try_response(), mimetype='text/event-stream')