This commit is contained in:
bennykok
2024-07-16 14:45:28 -07:00
commit 6ea0b09d80
207 changed files with 8190 additions and 0 deletions

View File

@@ -0,0 +1,4 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from .sdk import *
from .sdkconfiguration import *

View File

@@ -0,0 +1,4 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from .sdkhooks import *
from .types import *

View File

@@ -0,0 +1,55 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
import requests
from .types import SDKInitHook, BeforeRequestContext, BeforeRequestHook, AfterSuccessContext, AfterSuccessHook, AfterErrorContext, AfterErrorHook, Hooks
from typing import List, Optional, Tuple
class SDKHooks(Hooks):
def __init__(self):
self.sdk_init_hooks: List[SDKInitHook] = []
self.before_request_hooks: List[BeforeRequestHook] = []
self.after_success_hooks: List[AfterSuccessHook] = []
self.after_error_hooks: List[AfterErrorHook] = []
def register_sdk_init_hook(self, hook: SDKInitHook) -> None:
self.sdk_init_hooks.append(hook)
def register_before_request_hook(self, hook: BeforeRequestHook) -> None:
self.before_request_hooks.append(hook)
def register_after_success_hook(self, hook: AfterSuccessHook) -> None:
self.after_success_hooks.append(hook)
def register_after_error_hook(self, hook: AfterErrorHook) -> None:
self.after_error_hooks.append(hook)
def sdk_init(self, base_url: str, client: requests.Session) -> Tuple[str, requests.Session]:
for hook in self.sdk_init_hooks:
base_url, client = hook.sdk_init(base_url, client)
return base_url, client
def before_request(self, hook_ctx: BeforeRequestContext, request: requests.PreparedRequest) -> requests.PreparedRequest:
for hook in self.before_request_hooks:
out = hook.before_request(hook_ctx, request)
if isinstance(out, Exception):
raise out
request = out
return request
def after_success(self, hook_ctx: AfterSuccessContext, response: requests.Response) -> requests.Response:
for hook in self.after_success_hooks:
out = hook.after_success(hook_ctx, response)
if isinstance(out, Exception):
raise out
response = out
return response
def after_error(self, hook_ctx: AfterErrorContext, response: Optional[requests.Response], error: Optional[Exception]) -> Tuple[Optional[requests.Response], Optional[Exception]]:
for hook in self.after_error_hooks:
result = hook.after_error(hook_ctx, response, error)
if isinstance(result, Exception):
raise result
response, error = result
return response, error

View File

@@ -0,0 +1,74 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
import requests as requests_http
from abc import ABC, abstractmethod
from typing import Any, Callable, List, Optional, Tuple, Union
class HookContext:
operation_id: str
oauth2_scopes: Optional[List[str]] = None
security_source: Optional[Union[Any, Callable[[], Any]]] = None
def __init__(self, operation_id: str, oauth2_scopes: Optional[List[str]], security_source: Optional[Union[Any, Callable[[], Any]]]):
self.operation_id = operation_id
self.oauth2_scopes = oauth2_scopes
self.security_source = security_source
class BeforeRequestContext(HookContext):
def __init__(self, hook_ctx: HookContext):
super().__init__(hook_ctx.operation_id, hook_ctx.oauth2_scopes, hook_ctx.security_source)
class AfterSuccessContext(HookContext):
def __init__(self, hook_ctx: HookContext):
super().__init__(hook_ctx.operation_id, hook_ctx.oauth2_scopes, hook_ctx.security_source)
class AfterErrorContext(HookContext):
def __init__(self, hook_ctx: HookContext):
super().__init__(hook_ctx.operation_id, hook_ctx.oauth2_scopes, hook_ctx.security_source)
class SDKInitHook(ABC):
@abstractmethod
def sdk_init(self, base_url: str, client: requests_http.Session) -> Tuple[str, requests_http.Session]:
pass
class BeforeRequestHook(ABC):
@abstractmethod
def before_request(self, hook_ctx: BeforeRequestContext, request: requests_http.PreparedRequest) -> Union[requests_http.PreparedRequest, Exception]:
pass
class AfterSuccessHook(ABC):
@abstractmethod
def after_success(self, hook_ctx: AfterSuccessContext, response: requests_http.Response) -> Union[requests_http.Response, Exception]:
pass
class AfterErrorHook(ABC):
@abstractmethod
def after_error(self, hook_ctx: AfterErrorContext, response: Optional[requests_http.Response], error: Optional[Exception]) -> Union[Tuple[Optional[requests_http.Response], Optional[Exception]], Exception]:
pass
class Hooks(ABC):
@abstractmethod
def register_sdk_init_hook(self, hook: SDKInitHook):
pass
@abstractmethod
def register_before_request_hook(self, hook: BeforeRequestHook):
pass
@abstractmethod
def register_after_success_hook(self, hook: AfterSuccessHook):
pass
@abstractmethod
def register_after_error_hook(self, hook: AfterErrorHook):
pass

445
src/comfydeploy/comfyui.py Normal file
View File

@@ -0,0 +1,445 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
import requests as requests_http
from .sdkconfiguration import SDKConfiguration
from comfydeploy import utils
from comfydeploy._hooks import AfterErrorContext, AfterSuccessContext, BeforeRequestContext, HookContext
from comfydeploy.models import components, errors, operations
from typing import List, Optional
class Comfyui:
sdk_configuration: SDKConfiguration
def __init__(self, sdk_config: SDKConfiguration) -> None:
self.sdk_configuration = sdk_config
def get_auth_response_request_id_(self, request_id: str) -> operations.GetAuthResponseRequestIDResponse:
r"""Get an API Key with code
This endpoints is specifically built for ComfyUI workflow upload.
"""
hook_ctx = HookContext(operation_id='get_/auth-response/{request_id}', oauth2_scopes=[], security_source=self.sdk_configuration.security)
request = operations.GetAuthResponseRequestIDRequest(
request_id=request_id,
)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = utils.generate_url(base_url, '/auth-response/{request_id}', request)
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('GET', url, params=query_params, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['401','4XX','500','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.GetAuthResponseRequestIDResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.GetAuthResponseRequestIDResponseBody])
res.two_hundred_application_json_object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 201:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.GetAuthResponseRequestIDComfyuiResponseBody])
res.two_hundred_and_one_application_json_object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 500:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.GetAuthResponseRequestIDResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res
def post_workflow(self, request: Optional[operations.PostWorkflowRequestBody] = None) -> operations.PostWorkflowResponse:
r"""Upload workflow from ComfyUI
This endpoints is specifically built for ComfyUI workflow upload.
"""
hook_ctx = HookContext(operation_id='post_/workflow', oauth2_scopes=[], security_source=self.sdk_configuration.security)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = base_url + '/workflow'
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
req_content_type, data, form = utils.serialize_request_body(request, Optional[operations.PostWorkflowRequestBody], "request", False, True, 'json')
if req_content_type is not None and req_content_type not in ('multipart/form-data', 'multipart/mixed'):
headers['content-type'] = req_content_type
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('POST', url, params=query_params, data=data, files=form, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['401','4XX','500','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.PostWorkflowResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.PostWorkflowResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 500:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.PostWorkflowResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res
def get_workflow_version_version_id_(self, version_id: str) -> operations.GetWorkflowVersionVersionIDResponse:
r"""Get comfyui workflow
Use this to retrieve comfyui workflow by id
"""
hook_ctx = HookContext(operation_id='get_/workflow-version/{version_id}', oauth2_scopes=[], security_source=self.sdk_configuration.security)
request = operations.GetWorkflowVersionVersionIDRequest(
version_id=version_id,
)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = utils.generate_url(base_url, '/workflow-version/{version_id}', request)
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('GET', url, params=query_params, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['401','4XX','500','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.GetWorkflowVersionVersionIDResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.GetWorkflowVersionVersionIDResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 500:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.GetWorkflowVersionVersionIDResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res
def get_workflow_id_(self, id: str) -> operations.GetWorkflowIDResponse:
r"""Get comfyui workflow
Use this to retrieve comfyui workflow by id
"""
hook_ctx = HookContext(operation_id='get_/workflow/{id}', oauth2_scopes=[], security_source=self.sdk_configuration.security)
request = operations.GetWorkflowIDRequest(
id=id,
)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = utils.generate_url(base_url, '/workflow/{id}', request)
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('GET', url, params=query_params, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['401','4XX','500','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.GetWorkflowIDResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.GetWorkflowIDResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 500:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.GetWorkflowIDResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res
def get_deployment_id_inputs(self, id: str) -> operations.GetDeploymentIDInputsResponse:
r"""Get comfyui workflow inputs definition
Use this to retrieve comfyui workflow inputs definition by id
"""
hook_ctx = HookContext(operation_id='get_/deployment/{id}/inputs', oauth2_scopes=[], security_source=self.sdk_configuration.security)
request = operations.GetDeploymentIDInputsRequest(
id=id,
)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = utils.generate_url(base_url, '/deployment/{id}/inputs', request)
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('GET', url, params=query_params, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['401','4XX','500','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.GetDeploymentIDInputsResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[List[operations.ResponseBody]])
res.response_bodies = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 500:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.GetDeploymentIDInputsResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res
def get_deployment(self, environment: Optional[operations.Environment] = None) -> operations.GetDeploymentResponse:
r"""Get all deployed workflows
Get all deployed workflows
"""
hook_ctx = HookContext(operation_id='get_/deployment', oauth2_scopes=[], security_source=self.sdk_configuration.security)
request = operations.GetDeploymentRequest(
environment=environment,
)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = base_url + '/deployment'
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
query_params = { **utils.get_query_params(request), **query_params }
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('GET', url, params=query_params, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['4XX','500','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.GetDeploymentResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[List[operations.GetDeploymentResponseBody]])
res.response_bodies = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 500:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.GetDeploymentResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res

89
src/comfydeploy/files.py Normal file
View File

@@ -0,0 +1,89 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
import requests as requests_http
from .sdkconfiguration import SDKConfiguration
from comfydeploy import utils
from comfydeploy._hooks import AfterErrorContext, AfterSuccessContext, BeforeRequestContext, HookContext
from comfydeploy.models import components, errors, operations
from typing import Optional
class Files:
sdk_configuration: SDKConfiguration
def __init__(self, sdk_config: SDKConfiguration) -> None:
self.sdk_configuration = sdk_config
def get_upload_url(self, type_: operations.Type, file_size: str) -> operations.GetUploadURLResponse:
r"""Upload any files to the storage
Usually when you run something, you want to upload a file, image etc.
"""
hook_ctx = HookContext(operation_id='get_/upload-url', oauth2_scopes=[], security_source=self.sdk_configuration.security)
request = operations.GetUploadURLRequest(
type=type_,
file_size=file_size,
)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = base_url + '/upload-url'
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
query_params = { **utils.get_query_params(request), **query_params }
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('GET', url, params=query_params, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['401','4XX','500','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.GetUploadURLResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.GetUploadURLResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 500:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.GetUploadURLResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res

298
src/comfydeploy/machines.py Normal file
View File

@@ -0,0 +1,298 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
import requests as requests_http
from .sdkconfiguration import SDKConfiguration
from comfydeploy import utils
from comfydeploy._hooks import AfterErrorContext, AfterSuccessContext, BeforeRequestContext, HookContext
from comfydeploy.models import components, errors, operations
from typing import Optional
class Machines:
sdk_configuration: SDKConfiguration
def __init__(self, sdk_config: SDKConfiguration) -> None:
self.sdk_configuration = sdk_config
def post_gpu_event(self, request: Optional[operations.PostGpuEventRequestBody] = None) -> operations.PostGpuEventResponse:
r"""Register a machine event"""
hook_ctx = HookContext(operation_id='post_/gpu_event', oauth2_scopes=[], security_source=self.sdk_configuration.security)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = base_url + '/gpu_event'
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
req_content_type, data, form = utils.serialize_request_body(request, Optional[operations.PostGpuEventRequestBody], "request", False, True, 'json')
if req_content_type is not None and req_content_type not in ('multipart/form-data', 'multipart/mixed'):
headers['content-type'] = req_content_type
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('POST', url, params=query_params, data=data, files=form, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['401','4XX','500','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.PostGpuEventResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.PostGpuEventResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 500:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.PostGpuEventResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res
def get_v1_machines(self, page: Optional[str] = None, page_size: Optional[str] = None, fields: Optional[operations.Fields] = None) -> operations.GetV1MachinesResponse:
r"""Retrieve all machines for a user
Retrieve details of all machines for the authenticated user, with pagination and optional field selection
"""
hook_ctx = HookContext(operation_id='get_/v1/machines', oauth2_scopes=[], security_source=self.sdk_configuration.security)
request = operations.GetV1MachinesRequest(
page=page,
page_size=page_size,
fields=fields,
)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = base_url + '/v1/machines'
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
query_params = { **utils.get_query_params(request), **query_params }
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('GET', url, params=query_params, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['400','401','4XX','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.GetV1MachinesResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.GetV1MachinesResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 400:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.GetV1MachinesResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res
def post_v1_machines(self, request: Optional[operations.PostV1MachinesRequestBody] = None) -> operations.PostV1MachinesResponse:
r"""Create a new machine
Create a new machine with optional default setting
"""
hook_ctx = HookContext(operation_id='post_/v1/machines', oauth2_scopes=[], security_source=self.sdk_configuration.security)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = base_url + '/v1/machines'
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
req_content_type, data, form = utils.serialize_request_body(request, Optional[operations.PostV1MachinesRequestBody], "request", False, True, 'json')
if req_content_type is not None and req_content_type not in ('multipart/form-data', 'multipart/mixed'):
headers['content-type'] = req_content_type
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('POST', url, params=query_params, data=data, files=form, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['400','401','4XX','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.PostV1MachinesResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.PostV1MachinesResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 400:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.PostV1MachinesResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res
def get_v1_machines_machine_id_(self, machine_id: str, ext_urls: Optional[str] = None) -> operations.GetV1MachinesMachineIDResponse:
r"""Retrieve a specific machine by ID
Retrieve details of a specific machine by its ID, with optional workspace details
"""
hook_ctx = HookContext(operation_id='get_/v1/machines/{machine_id}', oauth2_scopes=[], security_source=self.sdk_configuration.security)
request = operations.GetV1MachinesMachineIDRequest(
machine_id=machine_id,
ext_urls=ext_urls,
)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = utils.generate_url(base_url, '/v1/machines/{machine_id}', request)
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
query_params = { **utils.get_query_params(request), **query_params }
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('GET', url, params=query_params, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['400','401','4XX','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.GetV1MachinesMachineIDResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.GetV1MachinesMachineIDResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 400:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.GetV1MachinesMachineIDResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res

View File

@@ -0,0 +1,4 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
# package

View File

@@ -0,0 +1,6 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from .httpmetadata import *
from .security import *
__all__ = ["HTTPMetadata","Security"]

View File

@@ -0,0 +1,17 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
import requests as requests_http
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class HTTPMetadata:
response: requests_http.Response = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
r"""Raw HTTP response; suitable for custom response parsing"""
request: requests_http.Request = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
r"""Raw HTTP request; suitable for debugging"""

View File

@@ -0,0 +1,11 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
@dataclasses.dataclass
class Security:
bearer_auth: str = dataclasses.field(metadata={'security': { 'scheme': True, 'type': 'http', 'sub_type': 'bearer', 'field_name': 'Authorization' }})

View File

@@ -0,0 +1,24 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from .get_auth_response_request_id_ import *
from .get_deployment import *
from .get_deployment_id_inputs import *
from .get_run import *
from .get_upload_url import *
from .get_v1_machines import *
from .get_v1_machines_machine_id_ import *
from .get_v1_workflows import *
from .get_v1_workflows_workflow_id_ import *
from .get_v1_workflows_workflow_id_outputs import *
from .get_websocket_deployment_id_ import *
from .get_workflow_id_ import *
from .get_workflow_version_version_id_ import *
from .post_gpu_event import *
from .post_machine_endpoint import *
from .post_run import *
from .post_v1_machines import *
from .post_v1_workflows import *
from .post_workflow import *
from .sdkerror import *
__all__ = ["GetAuthResponseRequestIDResponseBody","GetDeploymentIDInputsResponseBody","GetDeploymentResponseBody","GetRunResponseBody","GetRunRunResponseBody","GetUploadURLResponseBody","GetV1MachinesMachineIDResponseBody","GetV1MachinesResponseBody","GetV1WorkflowsResponseBody","GetV1WorkflowsWorkflowIDOutputsResponseBody","GetV1WorkflowsWorkflowIDResponseBody","GetWebsocketDeploymentIDResponseBody","GetWorkflowIDResponseBody","GetWorkflowVersionVersionIDResponseBody","PostGpuEventResponseBody","PostMachineEndpointResponseBody","PostRunResponseBody","PostV1MachinesResponseBody","PostV1WorkflowsResponseBody","PostWorkflowResponseBody","SDKError"]

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetAuthResponseRequestIDResponseBody(Exception):
r"""Error when fetching the API Key with code"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetDeploymentResponseBody(Exception):
r"""error"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetDeploymentIDInputsResponseBody(Exception):
r"""Error when uploading the workflow"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,34 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetRunRunResponseBody(Exception):
r"""Error getting output"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetRunResponseBody(Exception):
r"""Workflow not found"""
code: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('code') }})
message: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('message') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetUploadURLResponseBody(Exception):
r"""Error when generating upload url"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1MachinesResponseBody(Exception):
r"""Error in retrieving machines"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1MachinesMachineIDResponseBody(Exception):
r"""Error in retrieving the specific machine"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1WorkflowsResponseBody(Exception):
r"""Error in retrieving workflows"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1WorkflowsWorkflowIDResponseBody(Exception):
r"""Error in retrieving the specific workflow"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1WorkflowsWorkflowIDOutputsResponseBody(Exception):
r"""Error in retrieving the specific workflow"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetWebsocketDeploymentIDResponseBody(Exception):
r"""Error creating run"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetWorkflowIDResponseBody(Exception):
r"""Error when uploading the workflow"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetWorkflowVersionVersionIDResponseBody(Exception):
r"""Error when uploading the workflow"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostGpuEventResponseBody(Exception):
r"""Error creating run"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostMachineEndpointResponseBody(Exception):
r"""Error creating run"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostRunResponseBody(Exception):
r"""Error creating run"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostV1MachinesResponseBody(Exception):
r"""Error in creating machine"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostV1WorkflowsResponseBody(Exception):
r"""Error in creating workflow"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,20 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostWorkflowResponseBody(Exception):
r"""Error when uploading the workflow"""
error: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('error') }})
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
def __str__(self) -> str:
return utils.marshal_json(self, type(self))

View File

@@ -0,0 +1,24 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
import requests as requests_http
class SDKError(Exception):
"""Represents an error returned by the API."""
message: str
status_code: int
body: str
raw_response: requests_http.Response
def __init__(self, message: str, status_code: int, body: str, raw_response: requests_http.Response):
self.message = message
self.status_code = status_code
self.body = body
self.raw_response = raw_response
def __str__(self):
body = ''
if len(self.body) > 0:
body = f'\n{self.body}'
return f'{self.message}: Status {self.status_code}{body}'

View File

@@ -0,0 +1,23 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from .get_auth_response_request_id_ import *
from .get_deployment import *
from .get_deployment_id_inputs import *
from .get_run import *
from .get_upload_url import *
from .get_v1_machines import *
from .get_v1_machines_machine_id_ import *
from .get_v1_workflows import *
from .get_v1_workflows_workflow_id_ import *
from .get_v1_workflows_workflow_id_outputs import *
from .get_websocket_deployment_id_ import *
from .get_workflow_id_ import *
from .get_workflow_version_version_id_ import *
from .post_gpu_event import *
from .post_machine_endpoint import *
from .post_run import *
from .post_v1_machines import *
from .post_v1_workflows import *
from .post_workflow import *
__all__ = ["CustomNodes","Dependencies","Environment","EventType","Fields","Files","GetAuthResponseRequestIDComfyuiResponseBody","GetAuthResponseRequestIDRequest","GetAuthResponseRequestIDResponse","GetAuthResponseRequestIDResponseBody","GetDeploymentIDInputsRequest","GetDeploymentIDInputsResponse","GetDeploymentRequest","GetDeploymentResponse","GetDeploymentResponseBody","GetRunGpu","GetRunRequest","GetRunResponse","GetRunResponseBody","GetRunWorkflowAPI","GetUploadURLRequest","GetUploadURLResponse","GetUploadURLResponseBody","GetV1MachinesMachineIDRequest","GetV1MachinesMachineIDResponse","GetV1MachinesMachineIDResponseBody","GetV1MachinesRequest","GetV1MachinesResponse","GetV1MachinesResponseBody","GetV1WorkflowsRequest","GetV1WorkflowsResponse","GetV1WorkflowsResponseBody","GetV1WorkflowsWorkflowIDOutputsRequest","GetV1WorkflowsWorkflowIDOutputsResponse","GetV1WorkflowsWorkflowIDOutputsResponseBody","GetV1WorkflowsWorkflowIDRequest","GetV1WorkflowsWorkflowIDResponse","GetV1WorkflowsWorkflowIDResponseBody","GetV1WorkflowsWorkflowIDVersions","GetWebsocketDeploymentIDRequest","GetWebsocketDeploymentIDResponse","GetWebsocketDeploymentIDResponseBody","GetWorkflowIDDependencies","GetWorkflowIDRequest","GetWorkflowIDResponse","GetWorkflowIDResponseBody","GetWorkflowIDSnapshot","GetWorkflowIDWorkflow","GetWorkflowIDWorkflowAPI","GetWorkflowVersionVersionIDDependencies","GetWorkflowVersionVersionIDRequest","GetWorkflowVersionVersionIDResponse","GetWorkflowVersionVersionIDResponseBody","GetWorkflowVersionVersionIDSnapshot","GetWorkflowVersionVersionIDWorkflowAPI","GitCustomNodes","Gpu","GpuProvider","GpuType","Inputs","InstallType","MachineType","Machines","Models","Node","One","Origin","Outputs","PostGpuEventRequestBody","PostGpuEventResponse","PostGpuEventResponseBody","PostMachineEndpointRequestBody","PostMachineEndpointResponse","PostMachineEndpointResponseBody","PostRunRequestBody","PostRunResponse","PostRunResponseBody","PostV1MachinesGpu","PostV1MachinesRequestBody","PostV1MachinesResponse","PostV1MachinesResponseBody","PostV1MachinesStatus","PostV1MachinesType","PostV1WorkflowsRequestBody","PostV1WorkflowsResponse","PostV1WorkflowsResponseBody","PostWorkflowRequestBody","PostWorkflowResponse","PostWorkflowResponseBody","QueryParamRunOrigin","ResponseBody","RunLog","RunOrigin","Snapshot","Status","Type","User","Versions","WebhookStatus","Workflow","WorkflowAPI","WorkflowInputs","WorkflowJSON","WsGpuType"]

View File

@@ -0,0 +1,46 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from typing import Optional
@dataclasses.dataclass
class GetAuthResponseRequestIDRequest:
request_id: str = dataclasses.field(metadata={'path_param': { 'field_name': 'request_id', 'style': 'simple', 'explode': False }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetAuthResponseRequestIDComfyuiResponseBody:
r"""The API key is not yet ready"""
message: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('message') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetAuthResponseRequestIDResponseBody:
r"""The returned API Key"""
api_key: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('api_key') }})
name: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('name') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetAuthResponseRequestIDResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
two_hundred_application_json_object: Optional[GetAuthResponseRequestIDResponseBody] = dataclasses.field(default=None)
r"""The returned API Key"""
two_hundred_and_one_application_json_object: Optional[GetAuthResponseRequestIDComfyuiResponseBody] = dataclasses.field(default=None)
r"""The API key is not yet ready"""

View File

@@ -0,0 +1,35 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from dataclasses_json import Undefined, dataclass_json
from enum import Enum
from typing import List, Optional
class Environment(str, Enum):
PRODUCTION = 'production'
STAGING = 'staging'
@dataclasses.dataclass
class GetDeploymentRequest:
environment: Optional[Environment] = dataclasses.field(default=None, metadata={'query_param': { 'field_name': 'environment', 'style': 'form', 'explode': True }})
@dataclasses.dataclass
class GetDeploymentResponseBody:
pass
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetDeploymentResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
response_bodies: Optional[List[GetDeploymentResponseBody]] = dataclasses.field(default=None)
r"""Display all production workflows"""

View File

@@ -0,0 +1,37 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from typing import List, Optional
@dataclasses.dataclass
class GetDeploymentIDInputsRequest:
id: str = dataclasses.field(metadata={'path_param': { 'field_name': 'id', 'style': 'simple', 'explode': False }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class ResponseBody:
class_type: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('class_type') }})
input_id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('input_id') }})
default_value: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('default_value') }})
min_value: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('min_value') }})
max_value: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('max_value') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetDeploymentIDInputsResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
response_bodies: Optional[List[ResponseBody]] = dataclasses.field(default=None)
r"""Retrieve the output"""

View File

@@ -0,0 +1,117 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from enum import Enum
from typing import Optional
@dataclasses.dataclass
class GetRunRequest:
run_id: str = dataclasses.field(metadata={'query_param': { 'field_name': 'run_id', 'style': 'form', 'explode': True }})
@dataclasses.dataclass
class WorkflowInputs:
pass
@dataclasses.dataclass
class GetRunWorkflowAPI:
pass
class Origin(str, Enum):
MANUAL = 'manual'
API = 'api'
PUBLIC_SHARE = 'public-share'
PUBLIC_TEMPLATE = 'public-template'
WORKSPACE = 'workspace'
class Status(str, Enum):
NOT_STARTED = 'not-started'
RUNNING = 'running'
UPLOADING = 'uploading'
SUCCESS = 'success'
FAILED = 'failed'
STARTED = 'started'
QUEUED = 'queued'
TIMEOUT = 'timeout'
CANCELLED = 'cancelled'
class GetRunGpu(str, Enum):
T4 = 'T4'
L4 = 'L4'
A10_G = 'A10G'
A100 = 'A100'
H100 = 'H100'
class MachineType(str, Enum):
CLASSIC = 'classic'
RUNPOD_SERVERLESS = 'runpod-serverless'
MODAL_SERVERLESS = 'modal-serverless'
COMFY_DEPLOY_SERVERLESS = 'comfy-deploy-serverless'
WORKSPACE = 'workspace'
WORKSPACE_V2 = 'workspace-v2'
@dataclasses.dataclass
class RunLog:
pass
class WebhookStatus(str, Enum):
SUCCESS = 'success'
FAILED = 'failed'
NOT_STARTED = 'not-started'
RUNNING = 'running'
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetRunResponseBody:
r"""Retrieve the output"""
id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('id') }})
workflow_version_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_version_id') }})
workflow_inputs: WorkflowInputs = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_inputs') }})
workflow_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_id') }})
workflow_api: GetRunWorkflowAPI = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_api') }})
machine_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('machine_id') }})
origin: Origin = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('origin') }})
status: Status = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('status') }})
ended_at: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('ended_at') }})
created_at: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('created_at') }})
queued_at: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('queued_at') }})
started_at: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('started_at') }})
gpu: Optional[GetRunGpu] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('gpu') }})
machine_version: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('machine_version') }})
machine_type: Optional[MachineType] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('machine_type') }})
modal_function_call_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('modal_function_call_id') }})
user_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('user_id') }})
org_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('org_id') }})
run_log: RunLog = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('run_log') }})
live_status: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('live_status') }})
progress: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('progress') }})
is_realtime: bool = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('is_realtime') }})
webhook: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('webhook') }})
webhook_status: Optional[WebhookStatus] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('webhook_status') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetRunResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[GetRunResponseBody] = dataclasses.field(default=None)
r"""Retrieve the output"""

View File

@@ -0,0 +1,48 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from enum import Enum
from typing import Optional
class Type(str, Enum):
IMAGE_PNG = 'image/png'
IMAGE_JPG = 'image/jpg'
IMAGE_JPEG = 'image/jpeg'
IMAGE_WEBP = 'image/webp'
VIDEO_MP4 = 'video/mp4'
VIDEO_WEBM = 'video/webm'
APPLICATION_OCTET_STREAM = 'application/octet-stream'
@dataclasses.dataclass
class GetUploadURLRequest:
type: Type = dataclasses.field(metadata={'query_param': { 'field_name': 'type', 'style': 'form', 'explode': True }})
file_size: str = dataclasses.field(metadata={'query_param': { 'field_name': 'file_size', 'style': 'form', 'explode': True }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetUploadURLResponseBody:
r"""Retrieve the output"""
upload_url: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('upload_url') }})
file_id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('file_id') }})
download_url: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('download_url') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetUploadURLResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[GetUploadURLResponseBody] = dataclasses.field(default=None)
r"""Retrieve the output"""

View File

@@ -0,0 +1,59 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from enum import Enum
from typing import Any, List, Optional
class Fields(str, Enum):
FULL = 'full'
MINIMAL = 'minimal'
@dataclasses.dataclass
class GetV1MachinesRequest:
page: Optional[str] = dataclasses.field(default='1', metadata={'query_param': { 'field_name': 'page', 'style': 'form', 'explode': True }})
page_size: Optional[str] = dataclasses.field(default='12', metadata={'query_param': { 'field_name': 'pageSize', 'style': 'form', 'explode': True }})
fields: Optional[Fields] = dataclasses.field(default=Fields.MINIMAL, metadata={'query_param': { 'field_name': 'fields', 'style': 'form', 'explode': True }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class Machines:
UNSET='__SPEAKEASY_UNSET__'
id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('id') }})
endpoints: List[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('endpoints') }})
name: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('name'), 'exclude': lambda f: f is None }})
type: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('type'), 'exclude': lambda f: f is None }})
status: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('status'), 'exclude': lambda f: f is None }})
created_at: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('created_at'), 'exclude': lambda f: f is None }})
updated_at: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('updated_at'), 'exclude': lambda f: f is None }})
object_info: Optional[Any] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('object_info'), 'exclude': lambda f: f is Machines.UNSET }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1MachinesResponseBody:
r"""Machines retrieved successfully"""
machines: List[Machines] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('machines') }})
total: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('total') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1MachinesResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[GetV1MachinesResponseBody] = dataclasses.field(default=None)
r"""Machines retrieved successfully"""

View File

@@ -0,0 +1,41 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from typing import Optional
@dataclasses.dataclass
class GetV1MachinesMachineIDRequest:
machine_id: str = dataclasses.field(metadata={'path_param': { 'field_name': 'machine_id', 'style': 'simple', 'explode': False }})
ext_urls: Optional[str] = dataclasses.field(default='false', metadata={'query_param': { 'field_name': 'extUrls', 'style': 'form', 'explode': True }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1MachinesMachineIDResponseBody:
r"""Specific machine retrieved successfully"""
id: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('id'), 'exclude': lambda f: f is None }})
name: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('name'), 'exclude': lambda f: f is None }})
type: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('type'), 'exclude': lambda f: f is None }})
status: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('status'), 'exclude': lambda f: f is None }})
created_at: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('created_at'), 'exclude': lambda f: f is None }})
updated_at: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('updated_at'), 'exclude': lambda f: f is None }})
message: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('message'), 'exclude': lambda f: f is None }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1MachinesMachineIDResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[GetV1MachinesMachineIDResponseBody] = dataclasses.field(default=None)
r"""Specific machine retrieved successfully"""

View File

@@ -0,0 +1,60 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from typing import Any, List, Optional
@dataclasses.dataclass
class GetV1WorkflowsRequest:
page: Optional[str] = dataclasses.field(default='1', metadata={'query_param': { 'field_name': 'page', 'style': 'form', 'explode': True }})
page_size: Optional[str] = dataclasses.field(default='12', metadata={'query_param': { 'field_name': 'pageSize', 'style': 'form', 'explode': True }})
search: Optional[str] = dataclasses.field(default=None, metadata={'query_param': { 'field_name': 'search', 'style': 'form', 'explode': True }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class User:
name: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('name') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class Versions:
id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('id') }})
version: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('version') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1WorkflowsResponseBody:
id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('id') }})
updated_at: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('updated_at') }})
name: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('name') }})
selected_machine_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('selected_machine_id') }})
count: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('count') }})
user: User = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('user') }})
versions: List[Versions] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('versions') }})
deployments: List[Any] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('deployments') }})
runs: List[Any] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('runs') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1WorkflowsResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
response_bodies: Optional[List[GetV1WorkflowsResponseBody]] = dataclasses.field(default=None)
r"""Workflows retrieved successfully"""

View File

@@ -0,0 +1,59 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from typing import Any, List, Optional
@dataclasses.dataclass
class GetV1WorkflowsWorkflowIDRequest:
workflow_id: str = dataclasses.field(metadata={'path_param': { 'field_name': 'workflow_id', 'style': 'simple', 'explode': False }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1WorkflowsWorkflowIDVersions:
UNSET='__SPEAKEASY_UNSET__'
workflow_id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_id') }})
id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('id') }})
comment: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('comment') }})
version: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('version') }})
snapshot: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('snapshot') }})
dependencies: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('dependencies') }})
created_at: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('created_at') }})
updated_at: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('updated_at') }})
workflow: Optional[Any] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow'), 'exclude': lambda f: f is GetV1WorkflowsWorkflowIDVersions.UNSET }})
workflow_api: Optional[Any] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_api'), 'exclude': lambda f: f is GetV1WorkflowsWorkflowIDVersions.UNSET }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1WorkflowsWorkflowIDResponseBody:
r"""Specific workflow retrieved successfully"""
name: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('name') }})
user_id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('user_id') }})
id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('id') }})
selected_machine_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('selected_machine_id') }})
created_at: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('created_at') }})
updated_at: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('updated_at') }})
versions: List[GetV1WorkflowsWorkflowIDVersions] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('versions') }})
org_id: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('org_id'), 'exclude': lambda f: f is None }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1WorkflowsWorkflowIDResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[GetV1WorkflowsWorkflowIDResponseBody] = dataclasses.field(default=None)
r"""Specific workflow retrieved successfully"""

View File

@@ -0,0 +1,58 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from enum import Enum
from typing import List, Optional
class QueryParamRunOrigin(str, Enum):
MANUAL = 'manual'
API = 'api'
PUBLIC_SHARE = 'public-share'
PUBLIC_TEMPLATE = 'public-template'
WORKSPACE = 'workspace'
@dataclasses.dataclass
class GetV1WorkflowsWorkflowIDOutputsRequest:
workflow_id: str = dataclasses.field(metadata={'path_param': { 'field_name': 'workflow_id', 'style': 'simple', 'explode': False }})
page: Optional[str] = dataclasses.field(default='1', metadata={'query_param': { 'field_name': 'page', 'style': 'form', 'explode': True }})
page_size: Optional[str] = dataclasses.field(default='12', metadata={'query_param': { 'field_name': 'pageSize', 'style': 'form', 'explode': True }})
search: Optional[str] = dataclasses.field(default=None, metadata={'query_param': { 'field_name': 'search', 'style': 'form', 'explode': True }})
run_origin: Optional[QueryParamRunOrigin] = dataclasses.field(default=None, metadata={'query_param': { 'field_name': 'runOrigin', 'style': 'form', 'explode': True }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class Outputs:
file_ur_ls: List[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('fileURLs') }})
run_id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('runID') }})
duration: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('duration') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1WorkflowsWorkflowIDOutputsResponseBody:
r"""Specific workflow retrieved successfully"""
outputs: List[Outputs] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('outputs') }})
total: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('total') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetV1WorkflowsWorkflowIDOutputsResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[GetV1WorkflowsWorkflowIDOutputsResponseBody] = dataclasses.field(default=None)
r"""Specific workflow retrieved successfully"""

View File

@@ -0,0 +1,34 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from typing import Optional
@dataclasses.dataclass
class GetWebsocketDeploymentIDRequest:
deployment_id: str = dataclasses.field(metadata={'path_param': { 'field_name': 'deployment_id', 'style': 'simple', 'explode': False }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetWebsocketDeploymentIDResponseBody:
r"""Workflow queued"""
ws_connection_url: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('ws_connection_url') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetWebsocketDeploymentIDResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[GetWebsocketDeploymentIDResponseBody] = dataclasses.field(default=None)
r"""Workflow queued"""

View File

@@ -0,0 +1,64 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from typing import Optional
@dataclasses.dataclass
class GetWorkflowIDRequest:
id: str = dataclasses.field(metadata={'path_param': { 'field_name': 'id', 'style': 'simple', 'explode': False }})
@dataclasses.dataclass
class GetWorkflowIDWorkflow:
pass
@dataclasses.dataclass
class GetWorkflowIDWorkflowAPI:
pass
@dataclasses.dataclass
class GetWorkflowIDSnapshot:
pass
@dataclasses.dataclass
class GetWorkflowIDDependencies:
pass
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetWorkflowIDResponseBody:
r"""Retrieve the output"""
workflow_id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_id') }})
id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('id') }})
workflow: GetWorkflowIDWorkflow = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow') }})
workflow_api: GetWorkflowIDWorkflowAPI = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_api') }})
user_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('user_id') }})
comment: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('comment') }})
version: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('version') }})
snapshot: GetWorkflowIDSnapshot = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('snapshot') }})
dependencies: GetWorkflowIDDependencies = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('dependencies') }})
created_at: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('created_at') }})
updated_at: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('updated_at') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetWorkflowIDResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[GetWorkflowIDResponseBody] = dataclasses.field(default=None)
r"""Retrieve the output"""

View File

@@ -0,0 +1,64 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from typing import Optional
@dataclasses.dataclass
class GetWorkflowVersionVersionIDRequest:
version_id: str = dataclasses.field(metadata={'path_param': { 'field_name': 'version_id', 'style': 'simple', 'explode': False }})
@dataclasses.dataclass
class Workflow:
pass
@dataclasses.dataclass
class GetWorkflowVersionVersionIDWorkflowAPI:
pass
@dataclasses.dataclass
class GetWorkflowVersionVersionIDSnapshot:
pass
@dataclasses.dataclass
class GetWorkflowVersionVersionIDDependencies:
pass
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetWorkflowVersionVersionIDResponseBody:
r"""Retrieve the output"""
workflow_id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_id') }})
id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('id') }})
workflow: Workflow = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow') }})
workflow_api: GetWorkflowVersionVersionIDWorkflowAPI = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_api') }})
user_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('user_id') }})
comment: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('comment') }})
version: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('version') }})
snapshot: GetWorkflowVersionVersionIDSnapshot = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('snapshot') }})
dependencies: GetWorkflowVersionVersionIDDependencies = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('dependencies') }})
created_at: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('created_at') }})
updated_at: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('updated_at') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GetWorkflowVersionVersionIDResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[GetWorkflowVersionVersionIDResponseBody] = dataclasses.field(default=None)
r"""Retrieve the output"""

View File

@@ -0,0 +1,67 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from enum import Enum
from typing import Optional
class GpuType(str, Enum):
T4 = 'T4'
L4 = 'L4'
A10_G = 'A10G'
A100 = 'A100'
H100 = 'H100'
class WsGpuType(str, Enum):
FOUR_THOUSAND_AND_NINETY = '4090'
class EventType(str, Enum):
GPU_START = 'gpu_start'
GPU_END = 'gpu_end'
class GpuProvider(str, Enum):
MODAL = 'modal'
RUNPOD = 'runpod'
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostGpuEventRequestBody:
UNSET='__SPEAKEASY_UNSET__'
machine_id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('machine_id') }})
timestamp: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('timestamp') }})
event_type: EventType = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('eventType') }})
gpu_provider: GpuProvider = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('gpu_provider') }})
gpu_type: Optional[GpuType] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('gpuType'), 'exclude': lambda f: f is PostGpuEventRequestBody.UNSET }})
ws_gpu_type: Optional[WsGpuType] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('wsGpuType'), 'exclude': lambda f: f is PostGpuEventRequestBody.UNSET }})
event_id: Optional[str] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('event_id'), 'exclude': lambda f: f is PostGpuEventRequestBody.UNSET }})
is_workspace: Optional[bool] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('is_workspace'), 'exclude': lambda f: f is PostGpuEventRequestBody.UNSET }})
user_or_org_id: Optional[str] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('user_or_org_id'), 'exclude': lambda f: f is PostGpuEventRequestBody.UNSET }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostGpuEventResponseBody:
r"""Machine event registered"""
event_id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('event_id') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostGpuEventResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[PostGpuEventResponseBody] = dataclasses.field(default=None)
r"""Machine event registered"""

View File

@@ -0,0 +1,37 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from typing import Optional
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostMachineEndpointRequestBody:
machine_id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('machine_id') }})
type: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('type') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostMachineEndpointResponseBody:
r"""Create short lived machine endpoint"""
url: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('url') }})
gpu: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('gpu') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostMachineEndpointResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[PostMachineEndpointResponseBody] = dataclasses.field(default=None)
r"""Create short lived machine endpoint"""

View File

@@ -0,0 +1,73 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from enum import Enum
from typing import Any, Dict, Optional, Union
class Gpu(str, Enum):
T4 = 'T4'
L4 = 'L4'
A10_G = 'A10G'
A100 = 'A100'
H100 = 'H100'
class RunOrigin(str, Enum):
MANUAL = 'manual'
API = 'api'
PUBLIC_SHARE = 'public-share'
PUBLIC_TEMPLATE = 'public-template'
WORKSPACE = 'workspace'
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostRunRequestBody:
r"""Run options"""
UNSET='__SPEAKEASY_UNSET__'
deployment_id: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('deployment_id'), 'exclude': lambda f: f is None }})
r"""Deployment ID to run"""
workflow_api: Optional[Any] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_api'), 'exclude': lambda f: f is PostRunRequestBody.UNSET }})
r"""Workflow API JSON to run"""
workflow_id: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_id'), 'exclude': lambda f: f is None }})
r"""Workflow ID to run"""
machine_id: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('machine_id'), 'exclude': lambda f: f is None }})
gpu: Optional[Gpu] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('gpu'), 'exclude': lambda f: f is None }})
concurrency_limit: Optional[float] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('concurrency_limit'), 'exclude': lambda f: f is None }})
private_volume_name: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('private_volume_name'), 'exclude': lambda f: f is None }})
timeout: Optional[float] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('timeout'), 'exclude': lambda f: f is None }})
run_origin: Optional[RunOrigin] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('run_origin'), 'exclude': lambda f: f is None }})
inputs: Optional[Dict[str, Inputs]] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('inputs'), 'exclude': lambda f: f is None }})
r"""External inputs to the workflow"""
webhook: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('webhook'), 'exclude': lambda f: f is None }})
r"""Webhook URL to receive workflow updates"""
stream: Optional[bool] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('stream'), 'exclude': lambda f: f is None }})
r"""Whether to return a streaming url"""
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostRunResponseBody:
r"""Workflow queued"""
run_id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('run_id') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostRunResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[PostRunResponseBody] = dataclasses.field(default=None)
r"""Workflow queued"""
Inputs = Union[str, float]

View File

@@ -0,0 +1,85 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from enum import Enum
from typing import Any, List, Optional
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostV1MachinesRequestBody:
default_machine: Optional[bool] = dataclasses.field(default=True, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('default_machine'), 'exclude': lambda f: f is None }})
class PostV1MachinesType(str, Enum):
COMFY_DEPLOY_SERVERLESS = 'comfy-deploy-serverless'
class PostV1MachinesStatus(str, Enum):
BUILDING = 'building'
class PostV1MachinesGpu(str, Enum):
A10_G = 'A10G'
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostV1MachinesResponseBody:
r"""Machine created successfully"""
UNSET='__SPEAKEASY_UNSET__'
id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('id') }})
user_id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('user_id') }})
name: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('name') }})
org_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('org_id') }})
endpoint: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('endpoint') }})
created_at: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('created_at') }})
updated_at: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('updated_at') }})
disabled: bool = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('disabled') }})
type: PostV1MachinesType = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('type') }})
status: PostV1MachinesStatus = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('status') }})
machine_version: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('machine_version') }})
machine_builder_version: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('machine_builder_version') }})
models: List[Any] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('models') }})
gpu: PostV1MachinesGpu = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('gpu') }})
pod_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('pod_id') }})
base_docker_image: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('base_docker_image') }})
allow_concurrent_inputs: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('allow_concurrent_inputs') }})
concurrency_limit: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('concurrency_limit') }})
legacy_mode: bool = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('legacy_mode') }})
ws_timeout: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('ws_timeout') }})
run_timeout: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('run_timeout') }})
idle_timeout: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('idle_timeout') }})
build_machine_instance_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('build_machine_instance_id') }})
build_log: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('build_log') }})
modal_app_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('modal_app_id') }})
target_workflow_id: Optional[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('target_workflow_id') }})
install_custom_node_with_gpu: bool = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('install_custom_node_with_gpu') }})
deleted: bool = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('deleted') }})
keep_warm: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('keep_warm') }})
allow_background_volume_commits: bool = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('allow_background_volume_commits') }})
gpu_workspace: bool = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('gpu_workspace') }})
snapshot: Optional[Any] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('snapshot'), 'exclude': lambda f: f is PostV1MachinesResponseBody.UNSET }})
ws_gpu: Optional[Any] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('ws_gpu'), 'exclude': lambda f: f is PostV1MachinesResponseBody.UNSET }})
object_info: Optional[Any] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('object_info'), 'exclude': lambda f: f is PostV1MachinesResponseBody.UNSET }})
dependencies: Optional[Any] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('dependencies'), 'exclude': lambda f: f is PostV1MachinesResponseBody.UNSET }})
extra_docker_commands: Optional[Any] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('extra_docker_commands'), 'exclude': lambda f: f is PostV1MachinesResponseBody.UNSET }})
docker_command_steps: Optional[Any] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('docker_command_steps'), 'exclude': lambda f: f is PostV1MachinesResponseBody.UNSET }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostV1MachinesResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[PostV1MachinesResponseBody] = dataclasses.field(default=None)
r"""Machine created successfully"""

View File

@@ -0,0 +1,44 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from typing import Optional
@dataclasses.dataclass
class WorkflowJSON:
pass
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostV1WorkflowsRequestBody:
workflow_name: Optional[str] = dataclasses.field(default='new workflow', metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_name'), 'exclude': lambda f: f is None }})
workflow_json: Optional[WorkflowJSON] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_json'), 'exclude': lambda f: f is None }})
create_machine: Optional[bool] = dataclasses.field(default=True, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('create_machine'), 'exclude': lambda f: f is None }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostV1WorkflowsResponseBody:
r"""Workflow created successfully"""
workflow_id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_id') }})
workflow_version: float = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_version') }})
machine_id: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('machine_id'), 'exclude': lambda f: f is None }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostV1WorkflowsResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[PostV1WorkflowsResponseBody] = dataclasses.field(default=None)
r"""Workflow created successfully"""

View File

@@ -0,0 +1,135 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from ...models.components import httpmetadata as components_httpmetadata
from comfydeploy import utils
from dataclasses_json import Undefined, dataclass_json
from enum import Enum
from typing import Any, Dict, List, Optional, Union
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class WorkflowAPI:
inputs: Dict[str, Any] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('inputs') }})
class_type: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('class_type'), 'exclude': lambda f: f is None }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class GitCustomNodes:
hash: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('hash') }})
disabled: bool = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('disabled') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class Snapshot:
comfyui: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('comfyui') }})
git_custom_nodes: Dict[str, GitCustomNodes] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('git_custom_nodes') }})
file_custom_nodes: List[Any] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('file_custom_nodes') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class Node:
inputs: Dict[str, Any] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('inputs') }})
class_type: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('class_type'), 'exclude': lambda f: f is None }})
class One(str, Enum):
COPY = 'copy'
UNZIP = 'unzip'
GIT_CLONE = 'git-clone'
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class CustomNodes:
name: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('name') }})
url: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('url') }})
node: Optional[List[Node]] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('node'), 'exclude': lambda f: f is None }})
hash: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('hash'), 'exclude': lambda f: f is None }})
files: Optional[List[str]] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('files'), 'exclude': lambda f: f is None }})
install_type: Optional[InstallType] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('install_type'), 'exclude': lambda f: f is None }})
warning: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('warning'), 'exclude': lambda f: f is None }})
pip: Optional[List[str]] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('pip'), 'exclude': lambda f: f is None }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class Models:
name: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('name') }})
hash: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('hash'), 'exclude': lambda f: f is None }})
url: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('url'), 'exclude': lambda f: f is None }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class Files:
name: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('name') }})
hash: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('hash'), 'exclude': lambda f: f is None }})
url: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('url'), 'exclude': lambda f: f is None }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class Dependencies:
comfyui: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('comfyui') }})
missing_nodes: List[str] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('missing_nodes') }})
custom_nodes: Dict[str, CustomNodes] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('custom_nodes') }})
models: Dict[str, List[Models]] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('models') }})
files: Dict[str, List[Files]] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('files') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostWorkflowRequestBody:
UNSET='__SPEAKEASY_UNSET__'
workflow_api: Dict[str, WorkflowAPI] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_api') }})
snapshot: Snapshot = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('snapshot') }})
workflow_id: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_id'), 'exclude': lambda f: f is None }})
workflow_name: Optional[str] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_name'), 'exclude': lambda f: f is None }})
workflow: Optional[Any] = dataclasses.field(default=UNSET, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow'), 'exclude': lambda f: f is PostWorkflowRequestBody.UNSET }})
dependencies: Optional[Dependencies] = dataclasses.field(default=None, metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('dependencies'), 'exclude': lambda f: f is None }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostWorkflowResponseBody:
r"""Retrieve the output"""
workflow_id: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('workflow_id') }})
version: str = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('version') }})
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class PostWorkflowResponse:
http_meta: components_httpmetadata.HTTPMetadata = dataclasses.field(metadata={'dataclasses_json': { 'exclude': lambda f: True }})
object: Optional[PostWorkflowResponseBody] = dataclasses.field(default=None)
r"""Retrieve the output"""
InstallType = Union[One, str]

164
src/comfydeploy/run.py Normal file
View File

@@ -0,0 +1,164 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
import requests as requests_http
from .sdkconfiguration import SDKConfiguration
from comfydeploy import utils
from comfydeploy._hooks import AfterErrorContext, AfterSuccessContext, BeforeRequestContext, HookContext
from comfydeploy.models import components, errors, operations
from typing import Optional
class Run:
sdk_configuration: SDKConfiguration
def __init__(self, sdk_config: SDKConfiguration) -> None:
self.sdk_configuration = sdk_config
def get(self, run_id: str) -> operations.GetRunResponse:
r"""Get workflow run output
Call this to get a run's output, usually in conjunction with polling method
"""
hook_ctx = HookContext(operation_id='get_/run', oauth2_scopes=[], security_source=self.sdk_configuration.security)
request = operations.GetRunRequest(
run_id=run_id,
)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = base_url + '/run'
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
query_params = { **utils.get_query_params(request), **query_params }
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('GET', url, params=query_params, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['400','401','4XX','500','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.GetRunResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.GetRunResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 400:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.GetRunResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 500:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.GetRunRunResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res
def create(self, request: Optional[operations.PostRunRequestBody] = None) -> operations.PostRunResponse:
r"""Run a workflow via deployment_id"""
hook_ctx = HookContext(operation_id='post_/run', oauth2_scopes=[], security_source=self.sdk_configuration.security)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = base_url + '/run'
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
req_content_type, data, form = utils.serialize_request_body(request, Optional[operations.PostRunRequestBody], "request", False, True, 'json')
if req_content_type is not None and req_content_type not in ('multipart/form-data', 'multipart/mixed'):
headers['content-type'] = req_content_type
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('POST', url, params=query_params, data=data, files=form, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['401','4XX','500','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.PostRunResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.PostRunResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 500:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.PostRunResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res

89
src/comfydeploy/sdk.py Normal file
View File

@@ -0,0 +1,89 @@
"""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)

View File

@@ -0,0 +1,44 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
import requests as requests_http
from ._hooks import SDKHooks
from .utils import utils
from .utils.retries import RetryConfig
from comfydeploy.models import components
from dataclasses import dataclass
from typing import Callable, Dict, Optional, Tuple, Union
SERVERS = [
'https:///api',
]
"""Contains the list of servers available to the SDK"""
@dataclass
class SDKConfiguration:
client: requests_http.Session
security: Union[components.Security,Callable[[], components.Security]] = None
server_url: Optional[str] = ''
server_idx: Optional[int] = 0
language: str = 'python'
openapi_doc_version: str = '0.0.1'
sdk_version: str = '0.0.1'
gen_version: str = '2.372.3'
user_agent: str = 'speakeasy-sdk/python 0.0.1 2.372.3 0.0.1 comfydeploy'
retry_config: Optional[RetryConfig] = None
def __post_init__(self):
self._hooks = SDKHooks()
def get_server_details(self) -> Tuple[str, Dict[str, str]]:
if self.server_url is not None and self.server_url != '':
return utils.remove_suffix(self.server_url, '/'), {}
if self.server_idx is None:
self.server_idx = 0
return SERVERS[self.server_idx], {}
def get_hooks(self) -> SDKHooks:
return self._hooks

View File

@@ -0,0 +1,4 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from .retries import *
from .utils import *

View File

@@ -0,0 +1,119 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
import random
import time
from typing import List
import requests
class BackoffStrategy:
initial_interval: int
max_interval: int
exponent: float
max_elapsed_time: int
def __init__(self, initial_interval: int, max_interval: int, exponent: float, max_elapsed_time: int):
self.initial_interval = initial_interval
self.max_interval = max_interval
self.exponent = exponent
self.max_elapsed_time = max_elapsed_time
class RetryConfig:
strategy: str
backoff: BackoffStrategy
retry_connection_errors: bool
def __init__(self, strategy: str, backoff: BackoffStrategy, retry_connection_errors: bool):
self.strategy = strategy
self.backoff = backoff
self.retry_connection_errors = retry_connection_errors
class Retries:
config: RetryConfig
status_codes: List[str]
def __init__(self, config: RetryConfig, status_codes: List[str]):
self.config = config
self.status_codes = status_codes
class TemporaryError(Exception):
response: requests.Response
def __init__(self, response: requests.Response):
self.response = response
class PermanentError(Exception):
inner: Exception
def __init__(self, inner: Exception):
self.inner = inner
def retry(func, retries: Retries):
if retries.config.strategy == 'backoff':
def do_request():
res: requests.Response
try:
res = func()
for code in retries.status_codes:
if "X" in code.upper():
code_range = int(code[0])
status_major = res.status_code / 100
if status_major >= code_range and status_major < code_range + 1:
raise TemporaryError(res)
else:
parsed_code = int(code)
if res.status_code == parsed_code:
raise TemporaryError(res)
except requests.exceptions.ConnectionError as exception:
if retries.config.retry_connection_errors:
raise
raise PermanentError(exception) from exception
except requests.exceptions.Timeout as exception:
if retries.config.retry_connection_errors:
raise
raise PermanentError(exception) from exception
except TemporaryError:
raise
except Exception as exception:
raise PermanentError(exception) from exception
return res
return retry_with_backoff(do_request, retries.config.backoff.initial_interval, retries.config.backoff.max_interval, retries.config.backoff.exponent, retries.config.backoff.max_elapsed_time)
return func()
def retry_with_backoff(func, initial_interval=500, max_interval=60000, exponent=1.5, max_elapsed_time=3600000):
start = round(time.time()*1000)
retries = 0
while True:
try:
return func()
except PermanentError as exception:
raise exception.inner
except Exception as exception: # pylint: disable=broad-exception-caught
now = round(time.time()*1000)
if now - start > max_elapsed_time:
if isinstance(exception, TemporaryError):
return exception.response
raise
sleep = ((initial_interval/1000) *
exponent**retries + random.uniform(0, 1))
sleep = min(sleep, max_interval / 1000)
time.sleep(sleep)
retries += 1

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,431 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
import requests as requests_http
from .sdkconfiguration import SDKConfiguration
from comfydeploy import utils
from comfydeploy._hooks import AfterErrorContext, AfterSuccessContext, BeforeRequestContext, HookContext
from comfydeploy.models import components, errors, operations
from typing import List, Optional
class Workflows:
sdk_configuration: SDKConfiguration
def __init__(self, sdk_config: SDKConfiguration) -> None:
self.sdk_configuration = sdk_config
def get_websocket_deployment_id_(self, deployment_id: str) -> operations.GetWebsocketDeploymentIDResponse:
r"""Get a websocket url for a specific deployment"""
hook_ctx = HookContext(operation_id='get_/websocket/{deployment_id}', oauth2_scopes=[], security_source=self.sdk_configuration.security)
request = operations.GetWebsocketDeploymentIDRequest(
deployment_id=deployment_id,
)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = utils.generate_url(base_url, '/websocket/{deployment_id}', request)
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('GET', url, params=query_params, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['401','4XX','500','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.GetWebsocketDeploymentIDResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.GetWebsocketDeploymentIDResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 500:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.GetWebsocketDeploymentIDResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res
def post_machine_endpoint(self, request: Optional[operations.PostMachineEndpointRequestBody] = None) -> operations.PostMachineEndpointResponse:
r"""Create an endpoint for a machine"""
hook_ctx = HookContext(operation_id='post_/machine-endpoint', oauth2_scopes=[], security_source=self.sdk_configuration.security)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = base_url + '/machine-endpoint'
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
req_content_type, data, form = utils.serialize_request_body(request, Optional[operations.PostMachineEndpointRequestBody], "request", False, True, 'json')
if req_content_type is not None and req_content_type not in ('multipart/form-data', 'multipart/mixed'):
headers['content-type'] = req_content_type
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('POST', url, params=query_params, data=data, files=form, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['401','4XX','500','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.PostMachineEndpointResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.PostMachineEndpointResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 500:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.PostMachineEndpointResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res
def get_v1_workflows(self, page: Optional[str] = None, page_size: Optional[str] = None, search: Optional[str] = None) -> operations.GetV1WorkflowsResponse:
r"""Retrieve workflows
Retrieve workflows based on optional query parameters
"""
hook_ctx = HookContext(operation_id='get_/v1/workflows', oauth2_scopes=[], security_source=self.sdk_configuration.security)
request = operations.GetV1WorkflowsRequest(
page=page,
page_size=page_size,
search=search,
)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = base_url + '/v1/workflows'
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
query_params = { **utils.get_query_params(request), **query_params }
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('GET', url, params=query_params, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['400','401','4XX','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.GetV1WorkflowsResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[List[operations.GetV1WorkflowsResponseBody]])
res.response_bodies = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 400:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.GetV1WorkflowsResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res
def post_v1_workflows(self, request: Optional[operations.PostV1WorkflowsRequestBody] = None) -> operations.PostV1WorkflowsResponse:
r"""Create a new workflow
Create a new workflow by analyzing the provided workflow JSON
"""
hook_ctx = HookContext(operation_id='post_/v1/workflows', oauth2_scopes=[], security_source=self.sdk_configuration.security)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = base_url + '/v1/workflows'
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
req_content_type, data, form = utils.serialize_request_body(request, Optional[operations.PostV1WorkflowsRequestBody], "request", False, True, 'json')
if req_content_type is not None and req_content_type not in ('multipart/form-data', 'multipart/mixed'):
headers['content-type'] = req_content_type
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('POST', url, params=query_params, data=data, files=form, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['400','401','4XX','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.PostV1WorkflowsResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.PostV1WorkflowsResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 400:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.PostV1WorkflowsResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res
def get_v1_workflows_workflow_id_(self, workflow_id: str) -> operations.GetV1WorkflowsWorkflowIDResponse:
r"""Retrieve a specific workflow by ID
Retrieve the latest version of a specific workflow by its ID
"""
hook_ctx = HookContext(operation_id='get_/v1/workflows/{workflow_id}', oauth2_scopes=[], security_source=self.sdk_configuration.security)
request = operations.GetV1WorkflowsWorkflowIDRequest(
workflow_id=workflow_id,
)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = utils.generate_url(base_url, '/v1/workflows/{workflow_id}', request)
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('GET', url, params=query_params, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['400','401','4XX','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.GetV1WorkflowsWorkflowIDResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.GetV1WorkflowsWorkflowIDResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 400:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.GetV1WorkflowsWorkflowIDResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res
def get_v1_workflows_workflow_id_outputs(self, request: operations.GetV1WorkflowsWorkflowIDOutputsRequest) -> operations.GetV1WorkflowsWorkflowIDOutputsResponse:
r"""Retrieve the most recent outputs for a workflow
Retrieve the latest version of a specific workflow by its ID
"""
hook_ctx = HookContext(operation_id='get_/v1/workflows/{workflow_id}/outputs', oauth2_scopes=[], security_source=self.sdk_configuration.security)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = utils.generate_url(base_url, '/v1/workflows/{workflow_id}/outputs', request)
if callable(self.sdk_configuration.security):
headers, query_params = utils.get_security(self.sdk_configuration.security())
else:
headers, query_params = utils.get_security(self.sdk_configuration.security)
query_params = { **utils.get_query_params(request), **query_params }
headers['Accept'] = 'application/json'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = client.prepare_request(requests_http.Request('GET', url, params=query_params, headers=headers))
req = self.sdk_configuration.get_hooks().before_request(BeforeRequestContext(hook_ctx), req)
http_res = client.send(req)
except Exception as e:
_, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), None, e)
if e is not None:
raise e
if utils.match_status_codes(['400','401','4XX','5XX'], http_res.status_code):
result, e = self.sdk_configuration.get_hooks().after_error(AfterErrorContext(hook_ctx), http_res, None)
if e is not None:
raise e
if result is not None:
http_res = result
else:
http_res = self.sdk_configuration.get_hooks().after_success(AfterSuccessContext(hook_ctx), http_res)
res = operations.GetV1WorkflowsWorkflowIDOutputsResponse(http_meta=components.HTTPMetadata(request=req, response=http_res))
if http_res.status_code == 200:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, Optional[operations.GetV1WorkflowsWorkflowIDOutputsResponseBody])
res.object = out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 400:
# pylint: disable=no-else-return
if utils.match_content_type(http_res.headers.get('Content-Type') or '', 'application/json'):
out = utils.unmarshal_json(http_res.text, errors.GetV1WorkflowsWorkflowIDOutputsResponseBody, infer_missing=True)
out.http_meta = components.HTTPMetadata(request=req, response=http_res)
raise out
else:
content_type = http_res.headers.get('Content-Type')
raise errors.SDKError(f'unknown content-type received: {content_type}', http_res.status_code, http_res.text, http_res)
elif http_res.status_code == 401 or http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res