"""Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT.""" from .basesdk import BaseSDK from .sdkconfiguration import SDKConfiguration from mistralai import models, utils from mistralai._hooks import HookContext from mistralai.accesses import Accesses from mistralai.documents import Documents from mistralai.types import OptionalNullable, UNSET from mistralai.utils import get_security_from_env from mistralai.utils.unmarshal_json_response import unmarshal_json_response from typing import Any, Mapping, Optional class Libraries(BaseSDK): r"""(beta) Libraries API to create and manage libraries - index your documents to enhance agent capabilities.""" documents: Documents r"""(beta) Libraries API - manage documents in a library.""" accesses: Accesses r"""(beta) Libraries API - manage access to a library.""" def __init__( self, sdk_config: SDKConfiguration, parent_ref: Optional[object] = None ) -> None: BaseSDK.__init__(self, sdk_config, parent_ref=parent_ref) self.sdk_configuration = sdk_config self._init_sdks() def _init_sdks(self): self.documents = Documents(self.sdk_configuration, parent_ref=self.parent_ref) self.accesses = Accesses(self.sdk_configuration, parent_ref=self.parent_ref) def list( self, *, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.ListLibraryOut: r"""List all libraries you have access to. List all libraries that you have created or have been shared with you. :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) req = self._build_request( method="GET", path="/v1/libraries", base_url=base_url, url_variables=url_variables, request=None, request_body_required=False, request_has_path_params=False, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = self.do_request( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="libraries_list_v1", oauth2_scopes=[], security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["4XX", "5XX"], retry_config=retry_config, ) if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.ListLibraryOut, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) raise models.SDKError("Unexpected response received", http_res) async def list_async( self, *, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.ListLibraryOut: r"""List all libraries you have access to. List all libraries that you have created or have been shared with you. :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) req = self._build_request_async( method="GET", path="/v1/libraries", base_url=base_url, url_variables=url_variables, request=None, request_body_required=False, request_has_path_params=False, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = await self.do_request_async( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="libraries_list_v1", oauth2_scopes=[], security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["4XX", "5XX"], retry_config=retry_config, ) if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.ListLibraryOut, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) raise models.SDKError("Unexpected response received", http_res) def create( self, *, name: str, description: OptionalNullable[str] = UNSET, chunk_size: OptionalNullable[int] = UNSET, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.LibraryOut: r"""Create a new Library. Create a new Library, you will be marked as the owner and only you will have the possibility to share it with others. When first created this will only be accessible by you. :param name: :param description: :param chunk_size: :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.LibraryIn( name=name, description=description, chunk_size=chunk_size, ) req = self._build_request( method="POST", path="/v1/libraries", base_url=base_url, url_variables=url_variables, request=request, request_body_required=True, request_has_path_params=False, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, get_serialized_body=lambda: utils.serialize_request_body( request, False, False, "json", models.LibraryIn ), timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = self.do_request( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="libraries_create_v1", oauth2_scopes=[], security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "201", "application/json"): return unmarshal_json_response(models.LibraryOut, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( models.HTTPValidationErrorData, http_res ) raise models.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) raise models.SDKError("Unexpected response received", http_res) async def create_async( self, *, name: str, description: OptionalNullable[str] = UNSET, chunk_size: OptionalNullable[int] = UNSET, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.LibraryOut: r"""Create a new Library. Create a new Library, you will be marked as the owner and only you will have the possibility to share it with others. When first created this will only be accessible by you. :param name: :param description: :param chunk_size: :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.LibraryIn( name=name, description=description, chunk_size=chunk_size, ) req = self._build_request_async( method="POST", path="/v1/libraries", base_url=base_url, url_variables=url_variables, request=request, request_body_required=True, request_has_path_params=False, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, get_serialized_body=lambda: utils.serialize_request_body( request, False, False, "json", models.LibraryIn ), timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = await self.do_request_async( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="libraries_create_v1", oauth2_scopes=[], security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "201", "application/json"): return unmarshal_json_response(models.LibraryOut, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( models.HTTPValidationErrorData, http_res ) raise models.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) raise models.SDKError("Unexpected response received", http_res) def get( self, *, library_id: str, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.LibraryOut: r"""Detailed information about a specific Library. Given a library id, details information about that Library. :param library_id: :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.LibrariesGetV1Request( library_id=library_id, ) req = self._build_request( method="GET", path="/v1/libraries/{library_id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = self.do_request( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="libraries_get_v1", oauth2_scopes=[], security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.LibraryOut, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( models.HTTPValidationErrorData, http_res ) raise models.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) raise models.SDKError("Unexpected response received", http_res) async def get_async( self, *, library_id: str, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.LibraryOut: r"""Detailed information about a specific Library. Given a library id, details information about that Library. :param library_id: :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.LibrariesGetV1Request( library_id=library_id, ) req = self._build_request_async( method="GET", path="/v1/libraries/{library_id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = await self.do_request_async( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="libraries_get_v1", oauth2_scopes=[], security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.LibraryOut, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( models.HTTPValidationErrorData, http_res ) raise models.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) raise models.SDKError("Unexpected response received", http_res) def delete( self, *, library_id: str, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.LibraryOut: r"""Delete a library and all of it's document. Given a library id, deletes it together with all documents that have been uploaded to that library. :param library_id: :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.LibrariesDeleteV1Request( library_id=library_id, ) req = self._build_request( method="DELETE", path="/v1/libraries/{library_id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = self.do_request( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="libraries_delete_v1", oauth2_scopes=[], security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.LibraryOut, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( models.HTTPValidationErrorData, http_res ) raise models.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) raise models.SDKError("Unexpected response received", http_res) async def delete_async( self, *, library_id: str, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.LibraryOut: r"""Delete a library and all of it's document. Given a library id, deletes it together with all documents that have been uploaded to that library. :param library_id: :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.LibrariesDeleteV1Request( library_id=library_id, ) req = self._build_request_async( method="DELETE", path="/v1/libraries/{library_id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = await self.do_request_async( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="libraries_delete_v1", oauth2_scopes=[], security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.LibraryOut, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( models.HTTPValidationErrorData, http_res ) raise models.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) raise models.SDKError("Unexpected response received", http_res) def update( self, *, library_id: str, name: OptionalNullable[str] = UNSET, description: OptionalNullable[str] = UNSET, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.LibraryOut: r"""Update a library. Given a library id, you can update the name and description. :param library_id: :param name: :param description: :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.LibrariesUpdateV1Request( library_id=library_id, library_in_update=models.LibraryInUpdate( name=name, description=description, ), ) req = self._build_request( method="PUT", path="/v1/libraries/{library_id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=True, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, get_serialized_body=lambda: utils.serialize_request_body( request.library_in_update, False, False, "json", models.LibraryInUpdate ), timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = self.do_request( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="libraries_update_v1", oauth2_scopes=[], security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.LibraryOut, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( models.HTTPValidationErrorData, http_res ) raise models.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) raise models.SDKError("Unexpected response received", http_res) async def update_async( self, *, library_id: str, name: OptionalNullable[str] = UNSET, description: OptionalNullable[str] = UNSET, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.LibraryOut: r"""Update a library. Given a library id, you can update the name and description. :param library_id: :param name: :param description: :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.LibrariesUpdateV1Request( library_id=library_id, library_in_update=models.LibraryInUpdate( name=name, description=description, ), ) req = self._build_request_async( method="PUT", path="/v1/libraries/{library_id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=True, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, get_serialized_body=lambda: utils.serialize_request_body( request.library_in_update, False, False, "json", models.LibraryInUpdate ), timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = await self.do_request_async( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="libraries_update_v1", oauth2_scopes=[], security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.LibraryOut, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( models.HTTPValidationErrorData, http_res ) raise models.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise models.SDKError("API error occurred", http_res, http_res_text) raise models.SDKError("Unexpected response received", http_res)