diff --git a/src/leonardo_api/leonardo_sync.py b/src/leonardo_api/leonardo_sync.py index 246d6d1..b50f6b6 100644 --- a/src/leonardo_api/leonardo_sync.py +++ b/src/leonardo_api/leonardo_sync.py @@ -5,7 +5,7 @@ Copyright (c) 2023. All rights reserved. Created: 29.08.2023 -Last Modified: 24.11.2023 +Last Modified: 01.10.2024 Description: This file contains synchronous implementation for Leonardo.ai API @@ -40,9 +40,9 @@ def __init__(self, auth_token: str, logger: Optional[logging.Logger] = None) -> :param logger: default logger. Default will be initialized if not provided. :type logger: logging.Logger, optional """ - self.___auth_token = auth_token - self.___logger = logger if logger else setup_logger("Leonardo", "leonardo_async.log") - self.___logger.debug("Leonardo init complete") + self.__auth_token = auth_token + self.__logger = logger if logger else setup_logger("Leonardo", "leonardo_async.log") + self.__logger.debug("Leonardo init complete") def ___get_client_session(self, request_type: str = "get", empty: bool = False) -> requests.Session: """ @@ -57,7 +57,7 @@ def ___get_client_session(self, request_type: str = "get", empty: bool = False) """ headers = {} if not empty: - headers = {"Authorization": f"Bearer {self.___auth_token}"} + headers = {"Authorization": f"Bearer {self.__auth_token}"} if request_type.lower() == "get" or request_type.lower() == "delete": headers.update({"content-type": "application/json"}) if request_type.lower() == "post": @@ -77,20 +77,20 @@ def get_user_info(self) -> dict: Exception: If an error occurs while getting user info. """ url = "https://cloud.leonardo.ai/api/rest/v1/me" - self.___logger.debug(f"Requesting user info: GET {url}") + self.__logger.debug("Requesting user info: GET %s", url) session = self.___get_client_session("get") try: response = session.get(url) response.raise_for_status() response_dict = response.json() - self.___logger.debug(f"User info: {response_dict}") + self.__logger.debug("User info: %s", response_dict) session.close() return response_dict except Exception as error: - self.___logger.error(f"Error occurred while getting user info: {str(error)}") + self.__logger.error("Error occurred while getting user info: %s", str(error)) raise error - def post_generations( + def post_generations( # pylint: disable=too-many-positional-arguments self, prompt: str, negative_prompt: Optional[str] = None, @@ -180,16 +180,16 @@ def post_generations( "controlNet": control_net, "controlNetType": control_net_type, } - self.___logger.debug(f"Requesting post generations: POST {url} with payload: {payload}") + self.__logger.debug("Requesting post generations: POST %s with payload: %s", url, payload) session = self.___get_client_session("post") try: response = session.post(url, json=payload) response.raise_for_status() response_dict = response.json() - self.___logger.debug(f"Post generations: {response_dict}") + self.__logger.debug("Post generations: %s", response_dict) return response_dict except Exception as error: - self.___logger.error(f"Error occurred while post generations: {str(error)}") + self.__logger.error("Error occurred while post generations: %s", str(error)) raise error def get_single_generation(self, generation_id: str) -> dict: @@ -205,16 +205,16 @@ def get_single_generation(self, generation_id: str) -> dict: Exception: If an error occurs while getting generation info. """ url = f"https://cloud.leonardo.ai/api/rest/v1/generations/{generation_id}" - self.___logger.debug(f"Requested single generations: GET {url} with generation_id={generation_id}") + self.__logger.debug("Requested single generations: GET %s with generation_id=%s", url, generation_id) session = self.___get_client_session("get") try: response = session.get(url) response.raise_for_status() response_dict = response.json() - self.___logger.debug(f"Single generations: {response_dict}") + self.__logger.debug("Single generations: %s", response_dict) return response_dict except Exception as error: - self.___logger.error(f"Error occurred while get single generations: {str(error)}") + self.__logger.error("Error occurred while get single generations: %s", str(error)) raise error def delete_single_generation(self, generation_id: str) -> requests.Response: @@ -230,15 +230,15 @@ def delete_single_generation(self, generation_id: str) -> requests.Response: Exception: If an error occurs while deleting generation. """ url = f"https://cloud.leonardo.ai/api/rest/v1/generations/{generation_id}" - self.___logger.debug(f"Delete generations with generation_id={generation_id}: DELETE {url}") + self.__logger.debug("Delete generations with generation_id=%s: DELETE %s", generation_id, url) session = self.___get_client_session("delete") try: response = session.delete(url) response.raise_for_status() - self.___logger.debug(f"Generations {generation_id} has been deleted: {response}") + self.__logger.debug("Generations %s has been deleted: %s", generation_id, response) return response except Exception as error: - self.___logger.error(f"Error occurred while delete generation: {str(error)}") + self.__logger.error("Error occurred while delete generation: %s", str(error)) raise error def get_generations_by_user(self, user_id: str, offset: int = 0, limit: int = 10) -> dict: @@ -259,16 +259,16 @@ def get_generations_by_user(self, user_id: str, offset: int = 0, limit: int = 10 """ url = f"https://cloud.leonardo.ai/api/rest/v1/generations/user/{user_id}" params = {"offset": offset, "limit": limit} - self.___logger.debug(f"Requested generations for {user_id} with params {params}: GET {url}") + self.__logger.debug("Requested generations for %s with params %s: GET %s", user_id, params, url) session = self.___get_client_session("get") try: response = session.get(url, params=params) response.raise_for_status() response_dict = response.json() - self.___logger.debug(f"Generations for user {user_id} are: {response}") + self.__logger.debug("Generations for user %s are: %s", user_id, response) return response_dict except Exception as error: - self.___logger.error(f"Error occurred while obtaining user's generations: {str(error)}") + self.__logger.error("Error occurred while obtaining user's generations: %s", str(error)) raise error def upload_init_image(self, file_path: str) -> str: @@ -288,33 +288,33 @@ def upload_init_image(self, file_path: str) -> str: valid_extensions = ["png", "jpg", "jpeg", "webp"] extension = os.path.splitext(file_path)[1].strip(".") if extension not in valid_extensions: - raise ValueError(f"Invalid file extension. Must be one of {valid_extensions}") + raise ValueError("Invalid file extension. Must be one of %s", valid_extensions) url = "https://cloud.leonardo.ai/api/rest/v1/init-image" payload = {"extension": extension} - self.___logger.debug(f"Init image {file_path} upload requested with payload = {payload}: POST {url}") + self.__logger.debug(f"Init image %s upload requested with payload = %s: POST %s", file_path, payload, url) session = self.___get_client_session("post") try: response = session.post(url, json=payload) response.raise_for_status() data = response.json() - self.___logger.debug(f"Init image {file_path} initiated: {data}") + self.__logger.debug("Init image %s initiated: %s", file_path, data) upload_url = data["uploadInitImage"]["url"] fields = json.loads(data["uploadInitImage"]["fields"]) generation_id = data["uploadInitImage"]["id"] session.close() - self.___logger.debug(f"Init image {file_path} uploading as binary: POST {upload_url}") + self.__logger.debug("Init image %s uploading as binary: POST %s", file_path, upload_url) session = self.___get_client_session("post", empty=True) with open(file_path, "rb") as file: file_data = file.read() fields.update({"file": file_data}) response = session.post(upload_url, data=fields) response.raise_for_status() - self.___logger.debug(f"Init image {file_path} has been uploaded with generation_id={generation_id}") + self.__logger.debug("Init image %s has been uploaded with generation_id=%s", file_path, generation_id) return generation_id except Exception as error: - self.___logger.error(f"Error occurred while upload init image: {str(error)}") + self.__logger.error("Error occurred while upload init image: %s", str(error)) raise error def get_single_init_image(self, image_id: str) -> dict: @@ -327,16 +327,16 @@ def get_single_init_image(self, image_id: str) -> dict: :rtype: dict """ url = f"https://cloud.leonardo.ai/api/rest/v1/init-image/{image_id}" - self.___logger.debug(f"Requested single image with image_id={image_id}: GET {url}") + self.__logger.debug(f"Requested single image with image_id=%s: GET %s", image_id, url) session = self.___get_client_session("get") try: response = session.get(url) response.raise_for_status() response_dict = response.json() - self.___logger.debug(f"Single image provided: {response_dict}") + self.__logger.debug("Single image provided: %s", response_dict) return response_dict except Exception as error: - self.___logger.error(f"Error occurred while obtain single init image: {str(error)}") + self.__logger.error("Error occurred while obtain single init image: %s", str(error)) raise error def delete_init_image(self, image_id: str) -> requests.Response: @@ -352,16 +352,16 @@ def delete_init_image(self, image_id: str) -> requests.Response: Exception: If an error occurs while deleting init image. """ url = f"https://cloud.leonardo.ai/api/rest/v1/init-image/{image_id}" - self.___logger.debug(f"Requested to delete single image with image_id={image_id}: DELETE {url}") + self.__logger.debug("Requested to delete single image with image_id=%s: DELETE %s", image_id, url) session = self.___get_client_session("delete") try: response = session.delete(url) response.raise_for_status() - self.___logger.debug(f"Single image deleted: {response}") + self.__logger.debug("Single image deleted: %s", response) session.close() return response except Exception as error: - self.___logger.error(f"Error occurred while deleting init image: {str(error)}") + self.__logger.error("Error occurred while deleting init image: %s", str(error)) raise error def create_upscale(self, image_id: str) -> requests.Response: @@ -378,16 +378,16 @@ def create_upscale(self, image_id: str) -> requests.Response: """ url = "https://cloud.leonardo.ai/api/rest/v1/variations/upscale" payload = {"id": image_id} - self.___logger.debug(f"Requested to upscale image with payload {payload}: POST {url}") + self.__logger.debug("Requested to upscale image with payload %s: POST %s", payload, url) session = self.___get_client_session("post") try: response = session.post(url, json=payload) response.raise_for_status() - self.___logger.debug(f"Upscale created: {response}") + self.__logger.debug("Upscale created: %s", response) session.close() return response except Exception as error: - self.___logger.error(f"Error occurred while up-scaling image: {str(error)}") + self.__logger.error("Error occurred while up-scaling image: %s", str(error)) raise error def get_variation_by_id(self, generation_id: str) -> requests.Response: @@ -403,16 +403,16 @@ def get_variation_by_id(self, generation_id: str) -> requests.Response: Exception: If an error occurs while getting variation. """ url = f"https://cloud.leonardo.ai/api/rest/v1/variations/{generation_id}" - self.___logger.debug(f"Requested to obtain variation by id {generation_id}: GET {url}") + self.__logger.debug("Requested to obtain variation by id %s: GET %s", generation_id, url) session = self.___get_client_session("get") try: response = session.get(url) response.raise_for_status() response_dict = response.json() - self.___logger.debug(f"Get variation by ID: {response_dict}") + self.__logger.debug("Get variation by ID: %s", response_dict) return response_dict except Exception as error: - self.___logger.error(f"Error occurred while get variation by id: {str(error)}") + self.__logger.error("Error occurred while get variation by id: %s", str(error)) raise error def create_dataset(self, name: str, description: Optional[str] = None) -> requests.Response: @@ -431,16 +431,16 @@ def create_dataset(self, name: str, description: Optional[str] = None) -> reques """ url = "https://cloud.leonardo.ai/api/rest/v1/datasets" payload = {"name": name, "description": description} - self.___logger.debug(f"Requested to create dataset with payload {payload}: POST {url}") + self.__logger.debug("Requested to create dataset with payload %s: POST %s", payload, url) session = self.___get_client_session("post") try: response = session.post(url, json=payload) response.raise_for_status() - self.___logger.debug(f"Dataset has been created: {response}") + self.__logger.debug("Dataset has been created: %s", response) session.close() return response except Exception as error: - self.___logger.error(f"Error occurred while create dataset: {str(error)}") + self.__logger.error("Error occurred while create dataset: %s", str(error)) raise error def get_dataset_by_id(self, dataset_id: str) -> dict: @@ -456,16 +456,16 @@ def get_dataset_by_id(self, dataset_id: str) -> dict: Exception: If an error occurs while getting dataset. """ url = f"https://cloud.leonardo.ai/api/rest/v1/datasets/{dataset_id}" - self.___logger.debug(f"Requested to obtain dataset dataset_id={dataset_id}: GET {url}") + self.__logger.debug("Requested to obtain dataset dataset_id=%s: GET %s", dataset_id, url) session = self.___get_client_session("get") try: response = session.get(url) response.raise_for_status() response_dict = response.json() - self.___logger.debug(f"Dataset with dataset_id={dataset_id} provided: {response_dict}") + self.__logger.debug("Dataset with dataset_id=%s provided: %s", dataset_id, response_dict) return response_dict except Exception as error: - self.___logger.error(f"Error occurred while get dataset: {str(error)}") + self.__logger.error("Error occurred while get dataset: %s", str(error)) raise error def delete_dataset_by_id(self, dataset_id: str) -> requests.Response: @@ -481,15 +481,15 @@ def delete_dataset_by_id(self, dataset_id: str) -> requests.Response: Exception: If an error occurs while deleting dataset. """ url = f"https://cloud.leonardo.ai/api/rest/v1/datasets/{dataset_id}" - self.___logger.debug(f"Requested to delete dataset dataset_id={dataset_id}: DELETE {url}") + self.__logger.debug("Requested to delete dataset dataset_id=%s: DELETE %s", dataset_id, url) session = self.___get_client_session("delete") try: response = session.delete(url) response.raise_for_status() - self.___logger.debug(f"Dataset with dataset_id={dataset_id} has been deleted: {response}") + self.__logger.debug("Dataset with dataset_id=%s has been deleted: %s", dataset_id, response) return response except Exception as error: - self.___logger.error(f"Error occurred while delete dataset: {str(error)}") + self.__logger.error("Error occurred while delete dataset: %s", str(error)) raise error def upload_dataset_image(self, dataset_id: str, file_path: str) -> requests.Response: @@ -515,31 +515,34 @@ def upload_dataset_image(self, dataset_id: str, file_path: str) -> requests.Resp url = f"https://cloud.leonardo.ai/api/rest/v1/datasets/{dataset_id}/upload" payload = {"extension": extension} - self.___logger.debug(f"Requested to upload dataset_id={dataset_id} from {file_path}: POST {url}") + self.__logger.debug("Requested to upload dataset_id=%s from %s: POST %s", dataset_id, file_path, url) session = self.___get_client_session("post") try: response = session.post(url, json=payload) response.raise_for_status() data = response.json() - self.___logger.debug( - f"Dataset with dataset_id={dataset_id} started to upload from {file_path}:" f" {response}" + self.__logger.debug( + "Dataset with dataset_id=%s started to upload from %s: %s", + dataset_id, + file_path, + response ) upload_url = data["uploadDatasetImage"]["url"] fields = json.loads(data["uploadDatasetImage"]["fields"]) dataset_id = data["uploadDatasetImage"]["datasetId"] - self.___logger.debug(f"Uploading dataset_id={dataset_id} from {file_path}: POST {url}") + self.__logger.debug("Uploading dataset_id=%s from %s: POST %s", dataset_id, file_path, url) session = self.___get_client_session("post", empty=True) with open(file_path, "rb") as file: file_data = file.read() fields.update({"file": file_data}) response = session.post(upload_url, data=fields) response.raise_for_status() - self.___logger.debug(f"Dataset with dataset_id={dataset_id} uploaded using {file_path}") + self.__logger.debug("Dataset with dataset_id=%s uploaded using %s", dataset_id, file_path) session.close() return response except Exception as error: - self.___logger.error(f"Error occurred uploading dataset: {str(error)}") + self.__logger.error("Error occurred uploading dataset: %s", str(error)) raise error def upload_generated_image_to_dataset(self, dataset_id: str, generated_image_id: str) -> requests.Response: @@ -558,20 +561,26 @@ def upload_generated_image_to_dataset(self, dataset_id: str, generated_image_id: """ url = f"https://cloud.leonardo.ai/api/rest/v1/datasets/{dataset_id}/upload/gen" payload = {"generatedImageId": generated_image_id} - self.___logger.debug( - f"Requested to upload generated_image_id={generated_image_id} " f"to dataset_id={dataset_id}: POST {url}" + self.__logger.debug( + "Requested to upload generated_image_id=%s to dataset_id=%s: POST %s", + generated_image_id, + dataset_id, + url ) session = self.___get_client_session("post") try: response = session.post(url, json=payload) response.raise_for_status() - self.___logger.debug( - f"Image with image_id={generated_image_id} has been uploaded to " f"dataset_id={dataset_id}: {response}" + self.__logger.debug( + "Image with image_id=%s has been uploaded to dataset_id=%s: %s", + generated_image_id, + dataset_id, + response ) session.close() return response except Exception as error: - self.___logger.error(f"Error occurred while upload generated image to dataset: {str(error)}") + self.__logger.error("Error occurred while upload generated image to dataset: %s", str(error)) raise error def train_custom_model( @@ -623,16 +632,16 @@ def train_custom_model( "sd_Version": sd_version, "strength": strength, } - self.___logger.debug(f"Requested to train custom model with payload {payload}: POST {url}") + self.__logger.debug("Requested to train custom model with payload %s: POST %s", payload, url) session = self.___get_client_session("post") try: response = session.post(url, json=payload) response.raise_for_status() - self.___logger.debug(f"Custom modal has been trained: {response}") + self.__logger.debug("Custom modal has been trained: %s", response) session.close() return response except Exception as error: - self.___logger.error(f"Error training custom model: {str(error)}") + self.__logger.error("Error training custom model: %s", str(error)) raise error def get_custom_model_by_id(self, model_id: str) -> dict: @@ -648,17 +657,17 @@ def get_custom_model_by_id(self, model_id: str) -> dict: Exception: If an error occurs while getting custom model. """ url = f"https://cloud.leonardo.ai/api/rest/v1/models/{model_id}" - self.___logger.debug(f"Requested to obtain custom model by model_id={model_id}: GET {url}") + self.__logger.debug("Requested to obtain custom model by model_id=%s: GET %s", model_id, url) session = self.___get_client_session("get") try: response = session.get(url) response.raise_for_status() response_dict = response.json() - self.___logger.debug(f"Custom modal has been trained: {response_dict}") + self.__logger.debug("Custom modal has been trained: %s", response_dict) session.close() return response_dict except Exception as error: - self.___logger.error(f"Error obtaining custom model: {str(error)}") + self.__logger.error("Error obtaining custom model: %s", str(error)) raise error def delete_custom_model_by_id(self, model_id: str) -> requests.Response: @@ -674,15 +683,15 @@ def delete_custom_model_by_id(self, model_id: str) -> requests.Response: Exception: If an error occurs while deleting custom model. """ url = f"https://cloud.leonardo.ai/api/rest/v1/models/{model_id}" - self.___logger.debug(f"Requested to delete custom model by model_id={model_id}: GET {url}") + self.__logger.debug("Requested to delete custom model by model_id=%s: GET %s", model_id, url) session = self.___get_client_session("delete") try: response = session.delete(url) response.raise_for_status() - self.___logger.debug(f"Custom modal has been deleted: {response}") + self.__logger.debug("Custom modal has been deleted: %s", response) return response except Exception as error: - self.___logger.error(f"Error delete custom model: {str(error)}") + self.__logger.error("Error delete custom model: %s", str(error)) raise error def wait_for_image_generation( @@ -725,6 +734,6 @@ def wait_for_image_generation( time.sleep(poll_interval) if timeout_counter >= (timeout / poll_interval): - raise TimeoutError(f"Image has not been generated in {timeout} seconds") + raise TimeoutError("Image has not been generated in %s seconds", timeout) timeout_counter += 1