mirror of
https://github.com/LukeHagar/comfy-deploy-python.git
synced 2025-12-06 12:27:45 +00:00
90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
|
|
|
|
import requests as requests_http
|
|
from .comfyui import Comfyui
|
|
from .files import Files
|
|
from .machines import Machines
|
|
from .run import Run
|
|
from .sdkconfiguration import SDKConfiguration
|
|
from .utils.retries import RetryConfig
|
|
from .workflows import Workflows
|
|
from comfydeploy import utils
|
|
from comfydeploy._hooks import SDKHooks
|
|
from comfydeploy.models import components
|
|
from typing import Callable, Dict, Optional, Union
|
|
|
|
class ComfyDeploy:
|
|
r"""Comfy Deploy API: Interact with Comfy Deploy programmatically to trigger run and retrieve output"""
|
|
run: Run
|
|
files: Files
|
|
workflows: Workflows
|
|
comfyui: Comfyui
|
|
machines: Machines
|
|
|
|
sdk_configuration: SDKConfiguration
|
|
|
|
def __init__(self,
|
|
bearer_auth: Union[str, Callable[[], str]],
|
|
server_idx: Optional[int] = None,
|
|
server_url: Optional[str] = None,
|
|
url_params: Optional[Dict[str, str]] = None,
|
|
client: Optional[requests_http.Session] = None,
|
|
retry_config: Optional[RetryConfig] = None
|
|
) -> None:
|
|
"""Instantiates the SDK configuring it with the provided parameters.
|
|
|
|
:param bearer_auth: The bearer_auth required for authentication
|
|
:type bearer_auth: Union[str, Callable[[], str]]
|
|
:param server_idx: The index of the server to use for all operations
|
|
:type server_idx: int
|
|
:param server_url: The server URL to use for all operations
|
|
:type server_url: str
|
|
:param url_params: Parameters to optionally template the server URL with
|
|
:type url_params: Dict[str, str]
|
|
:param client: The requests.Session HTTP client to use for all operations
|
|
:type client: requests_http.Session
|
|
:param retry_config: The utils.RetryConfig to use globally
|
|
:type retry_config: RetryConfig
|
|
"""
|
|
if client is None:
|
|
client = requests_http.Session()
|
|
|
|
if callable(bearer_auth):
|
|
def security():
|
|
return components.Security(bearer_auth = bearer_auth())
|
|
else:
|
|
security = components.Security(bearer_auth = bearer_auth)
|
|
|
|
if server_url is not None:
|
|
if url_params is not None:
|
|
server_url = utils.template_url(server_url, url_params)
|
|
|
|
|
|
self.sdk_configuration = SDKConfiguration(
|
|
client,
|
|
security,
|
|
server_url,
|
|
server_idx,
|
|
retry_config=retry_config
|
|
)
|
|
|
|
hooks = SDKHooks()
|
|
|
|
current_server_url, *_ = self.sdk_configuration.get_server_details()
|
|
server_url, self.sdk_configuration.client = hooks.sdk_init(current_server_url, self.sdk_configuration.client)
|
|
if current_server_url != server_url:
|
|
self.sdk_configuration.server_url = server_url
|
|
|
|
# pylint: disable=protected-access
|
|
self.sdk_configuration.__dict__['_hooks'] = hooks
|
|
|
|
self._init_sdks()
|
|
|
|
|
|
def _init_sdks(self):
|
|
self.run = Run(self.sdk_configuration)
|
|
self.files = Files(self.sdk_configuration)
|
|
self.workflows = Workflows(self.sdk_configuration)
|
|
self.comfyui = Comfyui(self.sdk_configuration)
|
|
self.machines = Machines(self.sdk_configuration)
|