LukeParke/plexpy
1
0
Fork 0
mirror of https://github.com/LukeHagar/plexpy.git synced 2025-12-06 12:47:44 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
aeb8aa9fdb7908ac716245ae9cafd56ea08f0429
plexpy/src/plex_api_client/library.py

3925 lines
162 KiB
Python
Raw Blame History

"""Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT."""
from .basesdk import BaseSDK
from plex_api_client import utils
from plex_api_client._hooks import HookContext
from plex_api_client.models import errors, operations
from plex_api_client.types import BaseModel, OptionalNullable, UNSET
from typing import Any, Mapping, Optional, Union, cast
class Library(BaseSDK):
r"""API Calls interacting with Plex Media Server Libraries"""
def get_file_hash(
self,
*,
url: str,
type_: Optional[float] = None,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetFileHashResponse:
r"""Get Hash Value
This resource returns hash values for local files
:param url: This is the path to the local file, must be prefixed by `file://`
:param type: Item type
: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
request = operations.GetFileHashRequest(
url=url,
type=type_,
)
req = self._build_request(
method="GET",
path="/library/hashes",
base_url=base_url,
url_variables=url_variables,
request=request,
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(
operation_id="getFileHash",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "*"):
return operations.GetFileHashResponse(
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetFileHashBadRequestData
)
response_data.raw_response = http_res
raise errors.GetFileHashBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetFileHashUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetFileHashUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_file_hash_async(
self,
*,
url: str,
type_: Optional[float] = None,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetFileHashResponse:
r"""Get Hash Value
This resource returns hash values for local files
:param url: This is the path to the local file, must be prefixed by `file://`
:param type: Item type
: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
request = operations.GetFileHashRequest(
url=url,
type=type_,
)
req = self._build_request_async(
method="GET",
path="/library/hashes",
base_url=base_url,
url_variables=url_variables,
request=request,
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(
operation_id="getFileHash",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "*"):
return operations.GetFileHashResponse(
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetFileHashBadRequestData
)
response_data.raw_response = http_res
raise errors.GetFileHashBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetFileHashUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetFileHashUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_recently_added_library(
self,
*,
request: Union[
operations.GetRecentlyAddedLibraryRequest,
operations.GetRecentlyAddedLibraryRequestTypedDict,
],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetRecentlyAddedLibraryResponse:
r"""Get Recently Added
This endpoint will return the recently added content.
:param request: The request object to send.
: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
if not isinstance(request, BaseModel):
request = utils.unmarshal(
request, operations.GetRecentlyAddedLibraryRequest
)
request = cast(operations.GetRecentlyAddedLibraryRequest, request)
req = self._build_request(
method="GET",
path="/library/recentlyAdded",
base_url=base_url,
url_variables=url_variables,
request=request,
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(
operation_id="get-recently-added-library",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetRecentlyAddedLibraryResponse(
object=utils.unmarshal_json(
http_res.text,
Optional[operations.GetRecentlyAddedLibraryResponseBody],
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetRecentlyAddedLibraryBadRequestData
)
response_data.raw_response = http_res
raise errors.GetRecentlyAddedLibraryBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetRecentlyAddedLibraryUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetRecentlyAddedLibraryUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_recently_added_library_async(
self,
*,
request: Union[
operations.GetRecentlyAddedLibraryRequest,
operations.GetRecentlyAddedLibraryRequestTypedDict,
],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetRecentlyAddedLibraryResponse:
r"""Get Recently Added
This endpoint will return the recently added content.
:param request: The request object to send.
: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
if not isinstance(request, BaseModel):
request = utils.unmarshal(
request, operations.GetRecentlyAddedLibraryRequest
)
request = cast(operations.GetRecentlyAddedLibraryRequest, request)
req = self._build_request_async(
method="GET",
path="/library/recentlyAdded",
base_url=base_url,
url_variables=url_variables,
request=request,
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(
operation_id="get-recently-added-library",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetRecentlyAddedLibraryResponse(
object=utils.unmarshal_json(
http_res.text,
Optional[operations.GetRecentlyAddedLibraryResponseBody],
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetRecentlyAddedLibraryBadRequestData
)
response_data.raw_response = http_res
raise errors.GetRecentlyAddedLibraryBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetRecentlyAddedLibraryUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetRecentlyAddedLibraryUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_all_libraries(
self,
*,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetAllLibrariesResponse:
r"""Get All Libraries
A library section (commonly referred to as just a library) is a collection of media.
Libraries are typed, and depending on their type provide either a flat or a hierarchical view of the media.
For example, a music library has an artist > albums > tracks structure, whereas a movie library is flat.
Libraries have features beyond just being a collection of media; for starters, they include information about supported types, filters and sorts.
This allows a client to provide a rich interface around the media (e.g. allow sorting movies by release year).
: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
req = self._build_request(
method="GET",
path="/library/sections",
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(
operation_id="get-all-libraries",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetAllLibrariesResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetAllLibrariesResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetAllLibrariesBadRequestData
)
response_data.raw_response = http_res
raise errors.GetAllLibrariesBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetAllLibrariesUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetAllLibrariesUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_all_libraries_async(
self,
*,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetAllLibrariesResponse:
r"""Get All Libraries
A library section (commonly referred to as just a library) is a collection of media.
Libraries are typed, and depending on their type provide either a flat or a hierarchical view of the media.
For example, a music library has an artist > albums > tracks structure, whereas a movie library is flat.
Libraries have features beyond just being a collection of media; for starters, they include information about supported types, filters and sorts.
This allows a client to provide a rich interface around the media (e.g. allow sorting movies by release year).
: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
req = self._build_request_async(
method="GET",
path="/library/sections",
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(
operation_id="get-all-libraries",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetAllLibrariesResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetAllLibrariesResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetAllLibrariesBadRequestData
)
response_data.raw_response = http_res
raise errors.GetAllLibrariesBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetAllLibrariesUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetAllLibrariesUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_library_details(
self,
*,
section_key: int,
include_details: Optional[
operations.IncludeDetails
] = operations.IncludeDetails.ZERO,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetLibraryDetailsResponse:
r"""Get Library Details
## Library Details Endpoint
This endpoint provides comprehensive details about the library, focusing on organizational aspects rather than the content itself.
The details include:
### Directories
Organized into three categories:
- **Primary Directories**:
- Used in some clients for quick access to media subsets (e.g., \"All\", \"On Deck\").
- Most can be replicated via media queries.
- Customizable by users.
- **Secondary Directories**:
- Marked with `secondary=\"1\"`.
- Used in older clients for structured navigation.
- **Special Directories**:
- Includes a \"By Folder\" entry for filesystem-based browsing.
- Contains an obsolete `search=\"1\"` entry for on-the-fly search dialog creation.
### Types
Each type in the library comes with a set of filters and sorts, aiding in building dynamic media controls:
- **Type Object Attributes**:
- `key`: Endpoint for the media list of this type.
- `type`: Metadata type (if standard Plex type).
- `title`: Title for this content type (e.g., \"Movies\").
- **Filter Objects**:
- Subset of the media query language.
- Attributes include `filter` (name), `filterType` (data type), `key` (endpoint for value range), and `title`.
- **Sort Objects**:
- Description of sort fields.
- Attributes include `defaultDirection` (asc/desc), `descKey` and `key` (sort parameters), and `title`.
> **Note**: Filters and sorts are optional; without them, no filtering controls are rendered.
:param section_key: The unique key of the Plex library. Note: This is unique in the context of the Plex server.
:param include_details: Whether or not to include details for a section (types, filters, and sorts). Only exists for backwards compatibility, media providers other than the server libraries have it on always.
: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
request = operations.GetLibraryDetailsRequest(
include_details=include_details,
section_key=section_key,
)
req = self._build_request(
method="GET",
path="/library/sections/{sectionKey}",
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(
operation_id="get-library-details",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetLibraryDetailsResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetLibraryDetailsResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetLibraryDetailsBadRequestData
)
response_data.raw_response = http_res
raise errors.GetLibraryDetailsBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetLibraryDetailsUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetLibraryDetailsUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_library_details_async(
self,
*,
section_key: int,
include_details: Optional[
operations.IncludeDetails
] = operations.IncludeDetails.ZERO,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetLibraryDetailsResponse:
r"""Get Library Details
## Library Details Endpoint
This endpoint provides comprehensive details about the library, focusing on organizational aspects rather than the content itself.
The details include:
### Directories
Organized into three categories:
- **Primary Directories**:
- Used in some clients for quick access to media subsets (e.g., \"All\", \"On Deck\").
- Most can be replicated via media queries.
- Customizable by users.
- **Secondary Directories**:
- Marked with `secondary=\"1\"`.
- Used in older clients for structured navigation.
- **Special Directories**:
- Includes a \"By Folder\" entry for filesystem-based browsing.
- Contains an obsolete `search=\"1\"` entry for on-the-fly search dialog creation.
### Types
Each type in the library comes with a set of filters and sorts, aiding in building dynamic media controls:
- **Type Object Attributes**:
- `key`: Endpoint for the media list of this type.
- `type`: Metadata type (if standard Plex type).
- `title`: Title for this content type (e.g., \"Movies\").
- **Filter Objects**:
- Subset of the media query language.
- Attributes include `filter` (name), `filterType` (data type), `key` (endpoint for value range), and `title`.
- **Sort Objects**:
- Description of sort fields.
- Attributes include `defaultDirection` (asc/desc), `descKey` and `key` (sort parameters), and `title`.
> **Note**: Filters and sorts are optional; without them, no filtering controls are rendered.
:param section_key: The unique key of the Plex library. Note: This is unique in the context of the Plex server.
:param include_details: Whether or not to include details for a section (types, filters, and sorts). Only exists for backwards compatibility, media providers other than the server libraries have it on always.
: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
request = operations.GetLibraryDetailsRequest(
include_details=include_details,
section_key=section_key,
)
req = self._build_request_async(
method="GET",
path="/library/sections/{sectionKey}",
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(
operation_id="get-library-details",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetLibraryDetailsResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetLibraryDetailsResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetLibraryDetailsBadRequestData
)
response_data.raw_response = http_res
raise errors.GetLibraryDetailsBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetLibraryDetailsUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetLibraryDetailsUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def delete_library(
self,
*,
section_key: int,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.DeleteLibraryResponse:
r"""Delete Library Section
Delete a library using a specific section id
:param section_key: The unique key of the Plex library. Note: This is unique in the context of the Plex server.
: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
request = operations.DeleteLibraryRequest(
section_key=section_key,
)
req = self._build_request(
method="DELETE",
path="/library/sections/{sectionKey}",
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(
operation_id="deleteLibrary",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "*"):
return operations.DeleteLibraryResponse(
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.DeleteLibraryBadRequestData
)
response_data.raw_response = http_res
raise errors.DeleteLibraryBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.DeleteLibraryUnauthorizedData
)
response_data.raw_response = http_res
raise errors.DeleteLibraryUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def delete_library_async(
self,
*,
section_key: int,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.DeleteLibraryResponse:
r"""Delete Library Section
Delete a library using a specific section id
:param section_key: The unique key of the Plex library. Note: This is unique in the context of the Plex server.
: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
request = operations.DeleteLibraryRequest(
section_key=section_key,
)
req = self._build_request_async(
method="DELETE",
path="/library/sections/{sectionKey}",
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(
operation_id="deleteLibrary",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "*"):
return operations.DeleteLibraryResponse(
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.DeleteLibraryBadRequestData
)
response_data.raw_response = http_res
raise errors.DeleteLibraryBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.DeleteLibraryUnauthorizedData
)
response_data.raw_response = http_res
raise errors.DeleteLibraryUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_library_items(
self,
*,
request: Union[
operations.GetLibraryItemsRequest,
operations.GetLibraryItemsRequestTypedDict,
],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetLibraryItemsResponse:
r"""Get Library Items
Fetches details from a specific section of the library identified by a section key and a tag. The tag parameter accepts the following values:
- `all`: All items in the section.
- `unwatched`: Items that have not been played.
- `newest`: Items that are recently released.
- `recentlyAdded`: Items that are recently added to the library.
- `recentlyViewed`: Items that were recently viewed.
- `onDeck`: Items to continue watching.
- `collection`: Items categorized by collection.
- `edition`: Items categorized by edition.
- `genre`: Items categorized by genre.
- `year`: Items categorized by year of release.
- `decade`: Items categorized by decade.
- `director`: Items categorized by director.
- `actor`: Items categorized by starring actor.
- `country`: Items categorized by country of origin.
- `contentRating`: Items categorized by content rating.
- `rating`: Items categorized by rating.
- `resolution`: Items categorized by resolution.
- `firstCharacter`: Items categorized by the first letter.
- `folder`: Items categorized by folder.
- `albums`: Items categorized by album.
:param request: The request object to send.
: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
if not isinstance(request, BaseModel):
request = utils.unmarshal(request, operations.GetLibraryItemsRequest)
request = cast(operations.GetLibraryItemsRequest, request)
req = self._build_request(
method="GET",
path="/library/sections/{sectionKey}/{tag}",
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(
operation_id="get-library-items",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetLibraryItemsResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetLibraryItemsResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetLibraryItemsBadRequestData
)
response_data.raw_response = http_res
raise errors.GetLibraryItemsBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetLibraryItemsUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetLibraryItemsUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_library_items_async(
self,
*,
request: Union[
operations.GetLibraryItemsRequest,
operations.GetLibraryItemsRequestTypedDict,
],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetLibraryItemsResponse:
r"""Get Library Items
Fetches details from a specific section of the library identified by a section key and a tag. The tag parameter accepts the following values:
- `all`: All items in the section.
- `unwatched`: Items that have not been played.
- `newest`: Items that are recently released.
- `recentlyAdded`: Items that are recently added to the library.
- `recentlyViewed`: Items that were recently viewed.
- `onDeck`: Items to continue watching.
- `collection`: Items categorized by collection.
- `edition`: Items categorized by edition.
- `genre`: Items categorized by genre.
- `year`: Items categorized by year of release.
- `decade`: Items categorized by decade.
- `director`: Items categorized by director.
- `actor`: Items categorized by starring actor.
- `country`: Items categorized by country of origin.
- `contentRating`: Items categorized by content rating.
- `rating`: Items categorized by rating.
- `resolution`: Items categorized by resolution.
- `firstCharacter`: Items categorized by the first letter.
- `folder`: Items categorized by folder.
- `albums`: Items categorized by album.
:param request: The request object to send.
: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
if not isinstance(request, BaseModel):
request = utils.unmarshal(request, operations.GetLibraryItemsRequest)
request = cast(operations.GetLibraryItemsRequest, request)
req = self._build_request_async(
method="GET",
path="/library/sections/{sectionKey}/{tag}",
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(
operation_id="get-library-items",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetLibraryItemsResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetLibraryItemsResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetLibraryItemsBadRequestData
)
response_data.raw_response = http_res
raise errors.GetLibraryItemsBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetLibraryItemsUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetLibraryItemsUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_all_media_library(
self,
*,
request: Union[
operations.GetAllMediaLibraryRequest,
operations.GetAllMediaLibraryRequestTypedDict,
],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetAllMediaLibraryResponse:
r"""Get all media of library
Retrieves a list of all general media data for this library.
:param request: The request object to send.
: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
if not isinstance(request, BaseModel):
request = utils.unmarshal(request, operations.GetAllMediaLibraryRequest)
request = cast(operations.GetAllMediaLibraryRequest, request)
req = self._build_request(
method="GET",
path="/library/sections/{sectionKey}/all",
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(
operation_id="get-all-media-library",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "404", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetAllMediaLibraryResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetAllMediaLibraryResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetAllMediaLibraryBadRequestData
)
response_data.raw_response = http_res
raise errors.GetAllMediaLibraryBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetAllMediaLibraryUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetAllMediaLibraryUnauthorized(data=response_data)
if utils.match_response(http_res, ["404", "4XX"], "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_all_media_library_async(
self,
*,
request: Union[
operations.GetAllMediaLibraryRequest,
operations.GetAllMediaLibraryRequestTypedDict,
],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetAllMediaLibraryResponse:
r"""Get all media of library
Retrieves a list of all general media data for this library.
:param request: The request object to send.
: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
if not isinstance(request, BaseModel):
request = utils.unmarshal(request, operations.GetAllMediaLibraryRequest)
request = cast(operations.GetAllMediaLibraryRequest, request)
req = self._build_request_async(
method="GET",
path="/library/sections/{sectionKey}/all",
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(
operation_id="get-all-media-library",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "404", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetAllMediaLibraryResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetAllMediaLibraryResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetAllMediaLibraryBadRequestData
)
response_data.raw_response = http_res
raise errors.GetAllMediaLibraryBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetAllMediaLibraryUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetAllMediaLibraryUnauthorized(data=response_data)
if utils.match_response(http_res, ["404", "4XX"], "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_refresh_library_metadata(
self,
*,
section_key: int,
force: Optional[operations.Force] = None,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetRefreshLibraryMetadataResponse:
r"""Refresh Metadata Of The Library
This endpoint Refreshes all the Metadata of the library.
:param section_key: The unique key of the Plex library. Note: This is unique in the context of the Plex server.
:param force: Force the refresh even if the library is already being refreshed.
: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
request = operations.GetRefreshLibraryMetadataRequest(
force=force,
section_key=section_key,
)
req = self._build_request(
method="GET",
path="/library/sections/{sectionKey}/refresh",
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(
operation_id="get-refresh-library-metadata",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "*"):
return operations.GetRefreshLibraryMetadataResponse(
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetRefreshLibraryMetadataBadRequestData
)
response_data.raw_response = http_res
raise errors.GetRefreshLibraryMetadataBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetRefreshLibraryMetadataUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetRefreshLibraryMetadataUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_refresh_library_metadata_async(
self,
*,
section_key: int,
force: Optional[operations.Force] = None,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetRefreshLibraryMetadataResponse:
r"""Refresh Metadata Of The Library
This endpoint Refreshes all the Metadata of the library.
:param section_key: The unique key of the Plex library. Note: This is unique in the context of the Plex server.
:param force: Force the refresh even if the library is already being refreshed.
: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
request = operations.GetRefreshLibraryMetadataRequest(
force=force,
section_key=section_key,
)
req = self._build_request_async(
method="GET",
path="/library/sections/{sectionKey}/refresh",
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(
operation_id="get-refresh-library-metadata",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "*"):
return operations.GetRefreshLibraryMetadataResponse(
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetRefreshLibraryMetadataBadRequestData
)
response_data.raw_response = http_res
raise errors.GetRefreshLibraryMetadataBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetRefreshLibraryMetadataUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetRefreshLibraryMetadataUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_search_library(
self,
*,
section_key: int,
type_: operations.GetSearchLibraryQueryParamType,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetSearchLibraryResponse:
r"""Search Library
Search for content within a specific section of the library.
### Types
Each type in the library comes with a set of filters and sorts, aiding in building dynamic media controls:
- **Type Object Attributes**:
- `type`: Metadata type (if standard Plex type).
- `title`: Title for this content type (e.g., \"Movies\").
- **Filter Objects**:
- Subset of the media query language.
- Attributes include `filter` (name), `filterType` (data type), `key` (endpoint for value range), and `title`.
- **Sort Objects**:
- Description of sort fields.
- Attributes include `defaultDirection` (asc/desc), `descKey` and `key` (sort parameters), and `title`.
> **Note**: Filters and sorts are optional; without them, no filtering controls are rendered.
:param section_key: The unique key of the Plex library. Note: This is unique in the context of the Plex server.
:param type: The type of media to retrieve or filter by. 1 = movie 2 = show 3 = season 4 = episode E.g. A movie library will not return anything with type 3 as there are no seasons for movie libraries
: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
request = operations.GetSearchLibraryRequest(
section_key=section_key,
type=type_,
)
req = self._build_request(
method="GET",
path="/library/sections/{sectionKey}/search",
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(
operation_id="get-search-library",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetSearchLibraryResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetSearchLibraryResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetSearchLibraryBadRequestData
)
response_data.raw_response = http_res
raise errors.GetSearchLibraryBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetSearchLibraryUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetSearchLibraryUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_search_library_async(
self,
*,
section_key: int,
type_: operations.GetSearchLibraryQueryParamType,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetSearchLibraryResponse:
r"""Search Library
Search for content within a specific section of the library.
### Types
Each type in the library comes with a set of filters and sorts, aiding in building dynamic media controls:
- **Type Object Attributes**:
- `type`: Metadata type (if standard Plex type).
- `title`: Title for this content type (e.g., \"Movies\").
- **Filter Objects**:
- Subset of the media query language.
- Attributes include `filter` (name), `filterType` (data type), `key` (endpoint for value range), and `title`.
- **Sort Objects**:
- Description of sort fields.
- Attributes include `defaultDirection` (asc/desc), `descKey` and `key` (sort parameters), and `title`.
> **Note**: Filters and sorts are optional; without them, no filtering controls are rendered.
:param section_key: The unique key of the Plex library. Note: This is unique in the context of the Plex server.
:param type: The type of media to retrieve or filter by. 1 = movie 2 = show 3 = season 4 = episode E.g. A movie library will not return anything with type 3 as there are no seasons for movie libraries
: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
request = operations.GetSearchLibraryRequest(
section_key=section_key,
type=type_,
)
req = self._build_request_async(
method="GET",
path="/library/sections/{sectionKey}/search",
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(
operation_id="get-search-library",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetSearchLibraryResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetSearchLibraryResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetSearchLibraryBadRequestData
)
response_data.raw_response = http_res
raise errors.GetSearchLibraryBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetSearchLibraryUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetSearchLibraryUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_genres_library(
self,
*,
section_key: int,
type_: operations.GetGenresLibraryQueryParamType,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetGenresLibraryResponse:
r"""Get Genres of library media
Retrieves a list of all the genres that are found for the media in this library.
:param section_key: The unique key of the Plex library. Note: This is unique in the context of the Plex server.
:param type: The type of media to retrieve or filter by. 1 = movie 2 = show 3 = season 4 = episode E.g. A movie library will not return anything with type 3 as there are no seasons for movie libraries
: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
request = operations.GetGenresLibraryRequest(
section_key=section_key,
type=type_,
)
req = self._build_request(
method="GET",
path="/library/sections/{sectionKey}/genre",
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(
operation_id="get-genres-library",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "404", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetGenresLibraryResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetGenresLibraryResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetGenresLibraryBadRequestData
)
response_data.raw_response = http_res
raise errors.GetGenresLibraryBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetGenresLibraryUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetGenresLibraryUnauthorized(data=response_data)
if utils.match_response(http_res, ["404", "4XX"], "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_genres_library_async(
self,
*,
section_key: int,
type_: operations.GetGenresLibraryQueryParamType,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetGenresLibraryResponse:
r"""Get Genres of library media
Retrieves a list of all the genres that are found for the media in this library.
:param section_key: The unique key of the Plex library. Note: This is unique in the context of the Plex server.
:param type: The type of media to retrieve or filter by. 1 = movie 2 = show 3 = season 4 = episode E.g. A movie library will not return anything with type 3 as there are no seasons for movie libraries
: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
request = operations.GetGenresLibraryRequest(
section_key=section_key,
type=type_,
)
req = self._build_request_async(
method="GET",
path="/library/sections/{sectionKey}/genre",
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(
operation_id="get-genres-library",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "404", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetGenresLibraryResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetGenresLibraryResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetGenresLibraryBadRequestData
)
response_data.raw_response = http_res
raise errors.GetGenresLibraryBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetGenresLibraryUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetGenresLibraryUnauthorized(data=response_data)
if utils.match_response(http_res, ["404", "4XX"], "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_countries_library(
self,
*,
section_key: int,
type_: operations.GetCountriesLibraryQueryParamType,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetCountriesLibraryResponse:
r"""Get Countries of library media
Retrieves a list of all the countries that are found for the media in this library.
:param section_key: The unique key of the Plex library. Note: This is unique in the context of the Plex server.
:param type: The type of media to retrieve or filter by. 1 = movie 2 = show 3 = season 4 = episode E.g. A movie library will not return anything with type 3 as there are no seasons for movie libraries
: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
request = operations.GetCountriesLibraryRequest(
section_key=section_key,
type=type_,
)
req = self._build_request(
method="GET",
path="/library/sections/{sectionKey}/country",
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(
operation_id="get-countries-library",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "404", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetCountriesLibraryResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetCountriesLibraryResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetCountriesLibraryBadRequestData
)
response_data.raw_response = http_res
raise errors.GetCountriesLibraryBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetCountriesLibraryUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetCountriesLibraryUnauthorized(data=response_data)
if utils.match_response(http_res, ["404", "4XX"], "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_countries_library_async(
self,
*,
section_key: int,
type_: operations.GetCountriesLibraryQueryParamType,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetCountriesLibraryResponse:
r"""Get Countries of library media
Retrieves a list of all the countries that are found for the media in this library.
:param section_key: The unique key of the Plex library. Note: This is unique in the context of the Plex server.
:param type: The type of media to retrieve or filter by. 1 = movie 2 = show 3 = season 4 = episode E.g. A movie library will not return anything with type 3 as there are no seasons for movie libraries
: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
request = operations.GetCountriesLibraryRequest(
section_key=section_key,
type=type_,
)
req = self._build_request_async(
method="GET",
path="/library/sections/{sectionKey}/country",
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(
operation_id="get-countries-library",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "404", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetCountriesLibraryResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetCountriesLibraryResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetCountriesLibraryBadRequestData
)
response_data.raw_response = http_res
raise errors.GetCountriesLibraryBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetCountriesLibraryUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetCountriesLibraryUnauthorized(data=response_data)
if utils.match_response(http_res, ["404", "4XX"], "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_actors_library(
self,
*,
section_key: int,
type_: operations.GetActorsLibraryQueryParamType,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetActorsLibraryResponse:
r"""Get Actors of library media
Retrieves a list of all the actors that are found for the media in this library.
:param section_key: The unique key of the Plex library. Note: This is unique in the context of the Plex server.
:param type: The type of media to retrieve or filter by. 1 = movie 2 = show 3 = season 4 = episode E.g. A movie library will not return anything with type 3 as there are no seasons for movie libraries
: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
request = operations.GetActorsLibraryRequest(
section_key=section_key,
type=type_,
)
req = self._build_request(
method="GET",
path="/library/sections/{sectionKey}/actor",
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(
operation_id="get-actors-library",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "404", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetActorsLibraryResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetActorsLibraryResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetActorsLibraryBadRequestData
)
response_data.raw_response = http_res
raise errors.GetActorsLibraryBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetActorsLibraryUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetActorsLibraryUnauthorized(data=response_data)
if utils.match_response(http_res, ["404", "4XX"], "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_actors_library_async(
self,
*,
section_key: int,
type_: operations.GetActorsLibraryQueryParamType,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetActorsLibraryResponse:
r"""Get Actors of library media
Retrieves a list of all the actors that are found for the media in this library.
:param section_key: The unique key of the Plex library. Note: This is unique in the context of the Plex server.
:param type: The type of media to retrieve or filter by. 1 = movie 2 = show 3 = season 4 = episode E.g. A movie library will not return anything with type 3 as there are no seasons for movie libraries
: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
request = operations.GetActorsLibraryRequest(
section_key=section_key,
type=type_,
)
req = self._build_request_async(
method="GET",
path="/library/sections/{sectionKey}/actor",
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(
operation_id="get-actors-library",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "404", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetActorsLibraryResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetActorsLibraryResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetActorsLibraryBadRequestData
)
response_data.raw_response = http_res
raise errors.GetActorsLibraryBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetActorsLibraryUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetActorsLibraryUnauthorized(data=response_data)
if utils.match_response(http_res, ["404", "4XX"], "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_search_all_libraries(
self,
*,
request: Union[
operations.GetSearchAllLibrariesRequest,
operations.GetSearchAllLibrariesRequestTypedDict,
],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetSearchAllLibrariesResponse:
r"""Search All Libraries
Search the provided query across all library sections, or a single section, and return matches as hubs, split up by type.
:param request: The request object to send.
: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
if not isinstance(request, BaseModel):
request = utils.unmarshal(request, operations.GetSearchAllLibrariesRequest)
request = cast(operations.GetSearchAllLibrariesRequest, request)
req = self._build_request(
method="GET",
path="/library/search",
base_url=base_url,
url_variables=url_variables,
request=request,
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(
operation_id="get-search-all-libraries",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetSearchAllLibrariesResponse(
object=utils.unmarshal_json(
http_res.text,
Optional[operations.GetSearchAllLibrariesResponseBody],
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetSearchAllLibrariesBadRequestData
)
response_data.raw_response = http_res
raise errors.GetSearchAllLibrariesBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetSearchAllLibrariesUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetSearchAllLibrariesUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_search_all_libraries_async(
self,
*,
request: Union[
operations.GetSearchAllLibrariesRequest,
operations.GetSearchAllLibrariesRequestTypedDict,
],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetSearchAllLibrariesResponse:
r"""Search All Libraries
Search the provided query across all library sections, or a single section, and return matches as hubs, split up by type.
:param request: The request object to send.
: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
if not isinstance(request, BaseModel):
request = utils.unmarshal(request, operations.GetSearchAllLibrariesRequest)
request = cast(operations.GetSearchAllLibrariesRequest, request)
req = self._build_request_async(
method="GET",
path="/library/search",
base_url=base_url,
url_variables=url_variables,
request=request,
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(
operation_id="get-search-all-libraries",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetSearchAllLibrariesResponse(
object=utils.unmarshal_json(
http_res.text,
Optional[operations.GetSearchAllLibrariesResponseBody],
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetSearchAllLibrariesBadRequestData
)
response_data.raw_response = http_res
raise errors.GetSearchAllLibrariesBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetSearchAllLibrariesUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetSearchAllLibrariesUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_media_meta_data(
self,
*,
request: Union[
operations.GetMediaMetaDataRequest,
operations.GetMediaMetaDataRequestTypedDict,
],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetMediaMetaDataResponse:
r"""Get Media Metadata
This endpoint will return all the (meta)data of a library item specified with by the ratingKey.
:param request: The request object to send.
: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
if not isinstance(request, BaseModel):
request = utils.unmarshal(request, operations.GetMediaMetaDataRequest)
request = cast(operations.GetMediaMetaDataRequest, request)
req = self._build_request(
method="GET",
path="/library/metadata/{ratingKey}",
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(
operation_id="get-media-meta-data",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "404", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetMediaMetaDataResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetMediaMetaDataResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetMediaMetaDataBadRequestData
)
response_data.raw_response = http_res
raise errors.GetMediaMetaDataBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetMediaMetaDataUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetMediaMetaDataUnauthorized(data=response_data)
if utils.match_response(http_res, ["404", "4XX"], "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_media_meta_data_async(
self,
*,
request: Union[
operations.GetMediaMetaDataRequest,
operations.GetMediaMetaDataRequestTypedDict,
],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetMediaMetaDataResponse:
r"""Get Media Metadata
This endpoint will return all the (meta)data of a library item specified with by the ratingKey.
:param request: The request object to send.
: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
if not isinstance(request, BaseModel):
request = utils.unmarshal(request, operations.GetMediaMetaDataRequest)
request = cast(operations.GetMediaMetaDataRequest, request)
req = self._build_request_async(
method="GET",
path="/library/metadata/{ratingKey}",
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(
operation_id="get-media-meta-data",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "404", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetMediaMetaDataResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetMediaMetaDataResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetMediaMetaDataBadRequestData
)
response_data.raw_response = http_res
raise errors.GetMediaMetaDataBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetMediaMetaDataUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetMediaMetaDataUnauthorized(data=response_data)
if utils.match_response(http_res, ["404", "4XX"], "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_metadata_children(
self,
*,
rating_key: float,
include_elements: Optional[str] = None,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetMetadataChildrenResponse:
r"""Get Items Children
This endpoint will return the children of of a library item specified with the ratingKey.
:param rating_key: the id of the library item to return the children of.
:param include_elements: Adds additional elements to the response. Supported types are (Stream)
: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
request = operations.GetMetadataChildrenRequest(
rating_key=rating_key,
include_elements=include_elements,
)
req = self._build_request(
method="GET",
path="/library/metadata/{ratingKey}/children",
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(
operation_id="getMetadataChildren",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetMetadataChildrenResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetMetadataChildrenResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetMetadataChildrenBadRequestData
)
response_data.raw_response = http_res
raise errors.GetMetadataChildrenBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetMetadataChildrenUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetMetadataChildrenUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_metadata_children_async(
self,
*,
rating_key: float,
include_elements: Optional[str] = None,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetMetadataChildrenResponse:
r"""Get Items Children
This endpoint will return the children of of a library item specified with the ratingKey.
:param rating_key: the id of the library item to return the children of.
:param include_elements: Adds additional elements to the response. Supported types are (Stream)
: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
request = operations.GetMetadataChildrenRequest(
rating_key=rating_key,
include_elements=include_elements,
)
req = self._build_request_async(
method="GET",
path="/library/metadata/{ratingKey}/children",
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(
operation_id="getMetadataChildren",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetMetadataChildrenResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetMetadataChildrenResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetMetadataChildrenBadRequestData
)
response_data.raw_response = http_res
raise errors.GetMetadataChildrenBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetMetadataChildrenUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetMetadataChildrenUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_top_watched_content(
self,
*,
type_: operations.GetTopWatchedContentQueryParamType,
include_guids: Optional[int] = None,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetTopWatchedContentResponse:
r"""Get Top Watched Content
This endpoint will return the top watched content from libraries of a certain type
:param type: The type of media to retrieve or filter by. 1 = movie 2 = show 3 = season 4 = episode E.g. A movie library will not return anything with type 3 as there are no seasons for movie libraries
:param include_guids: Adds the Guids object to the response
: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
request = operations.GetTopWatchedContentRequest(
include_guids=include_guids,
type=type_,
)
req = self._build_request(
method="GET",
path="/library/all/top",
base_url=base_url,
url_variables=url_variables,
request=request,
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(
operation_id="getTopWatchedContent",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetTopWatchedContentResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetTopWatchedContentResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetTopWatchedContentBadRequestData
)
response_data.raw_response = http_res
raise errors.GetTopWatchedContentBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetTopWatchedContentUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetTopWatchedContentUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_top_watched_content_async(
self,
*,
type_: operations.GetTopWatchedContentQueryParamType,
include_guids: Optional[int] = None,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetTopWatchedContentResponse:
r"""Get Top Watched Content
This endpoint will return the top watched content from libraries of a certain type
:param type: The type of media to retrieve or filter by. 1 = movie 2 = show 3 = season 4 = episode E.g. A movie library will not return anything with type 3 as there are no seasons for movie libraries
:param include_guids: Adds the Guids object to the response
: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
request = operations.GetTopWatchedContentRequest(
include_guids=include_guids,
type=type_,
)
req = self._build_request_async(
method="GET",
path="/library/all/top",
base_url=base_url,
url_variables=url_variables,
request=request,
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(
operation_id="getTopWatchedContent",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetTopWatchedContentResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetTopWatchedContentResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetTopWatchedContentBadRequestData
)
response_data.raw_response = http_res
raise errors.GetTopWatchedContentBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetTopWatchedContentUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetTopWatchedContentUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
def get_on_deck(
self,
*,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetOnDeckResponse:
r"""Get On Deck
This endpoint will return the on deck content.
: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
req = self._build_request(
method="GET",
path="/library/onDeck",
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(
operation_id="getOnDeck",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetOnDeckResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetOnDeckResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetOnDeckBadRequestData
)
response_data.raw_response = http_res
raise errors.GetOnDeckBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetOnDeckUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetOnDeckUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = utils.stream_to_text(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
async def get_on_deck_async(
self,
*,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> operations.GetOnDeckResponse:
r"""Get On Deck
This endpoint will return the on deck content.
: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
req = self._build_request_async(
method="GET",
path="/library/onDeck",
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(
operation_id="getOnDeck",
oauth2_scopes=[],
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["400", "401", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return operations.GetOnDeckResponse(
object=utils.unmarshal_json(
http_res.text, Optional[operations.GetOnDeckResponseBody]
),
status_code=http_res.status_code,
content_type=http_res.headers.get("Content-Type") or "",
raw_response=http_res,
)
if utils.match_response(http_res, "400", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetOnDeckBadRequestData
)
response_data.raw_response = http_res
raise errors.GetOnDeckBadRequest(data=response_data)
if utils.match_response(http_res, "401", "application/json"):
response_data = utils.unmarshal_json(
http_res.text, errors.GetOnDeckUnauthorizedData
)
response_data.raw_response = http_res
raise errors.GetOnDeckUnauthorized(data=response_data)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
"API error occurred", http_res.status_code, http_res_text, http_res
)
content_type = http_res.headers.get("Content-Type")
http_res_text = await utils.stream_to_text_async(http_res)
raise errors.SDKError(
f"Unexpected response received (code: {http_res.status_code}, type: {content_type})",
http_res.status_code,
http_res_text,
http_res,
)
Reference in New Issue View Git Blame Copy Permalink