Asynchronous Client

This page contains the API reference for the asynchronous AsyncSeedr client.

seedrcc.async_client.AsyncSeedr

Bases: BaseClient

Asynchronous client for interacting with the Seedr API.

Example

import asyncio
from seedrcc import AsyncSeedr, Token

async def main():
    # Load a previously saved token from a Base64 string
    b64_string = "eydhY2Nlc3NfdG9rZW4nOiAnbmV2ZXIgZ29ubmEgZ2l2ZSB5b3UgdXAnfQ=="
    token = Token.from_base64(b64_string)

    # Initialize the client and make a request
    async with AsyncSeedr(token=token) as client:
        settings = await client.get_settings()
        print(f"Hello, {settings.account.username}")

if __name__ == "__main__":
    asyncio.run(main())

Source code in seedrcc/async_client.py

class AsyncSeedr(BaseClient):
    """Asynchronous client for interacting with the Seedr API.

    Example:
        ```python
        import asyncio
        from seedrcc import AsyncSeedr, Token

        async def main():
            # Load a previously saved token from a Base64 string
            b64_string = "eydhY2Nlc3NfdG9rZW4nOiAnbmV2ZXIgZ29ubmEgZ2l2ZSB5b3UgdXAnfQ=="
            token = Token.from_base64(b64_string)

            # Initialize the client and make a request
            async with AsyncSeedr(token=token) as client:
                settings = await client.get_settings()
                print(f"Hello, {settings.account.username}")

        if __name__ == "__main__":
            asyncio.run(main())
        ```
    """

    _client: httpx.AsyncClient
    _manages_client_lifecycle: bool

    def __init__(
        self,
        token: Token,
        on_token_refresh: Optional[Callable[[Token], None]] = None,
        httpx_client: Optional[httpx.AsyncClient] = None,
        timeout: float = 30.0,
        proxy: Optional[Dict[str, str]] = None,
        **httpx_kwargs: Any,
    ) -> None:
        """Initializes the asynchronous client with an existing token.

        Args:
            token: An authenticated `Token` object.
            on_token_refresh: An optional callback function that is called with the new
                `Token` object when the session is refreshed.
            httpx_client: An optional, pre-configured `httpx.AsyncClient` instance.
            timeout: The timeout for network requests in seconds.
            proxy: An optional dictionary of proxy to use for requests.
            **httpx_kwargs: Optional keyword arguments to pass to the `httpx.AsyncClient` constructor.
                These are ignored if `httpx_client` is provided.
        """
        super().__init__(token, on_token_refresh)
        if httpx_client is not None:
            self._client = httpx_client
            self._manages_client_lifecycle = False
        else:
            httpx_kwargs.setdefault("timeout", timeout)
            httpx_kwargs.setdefault("proxy", proxy)
            self._client = httpx.AsyncClient(**httpx_kwargs)
            self._manages_client_lifecycle = True

    @property
    def token(self) -> Token:
        return super().token

    @staticmethod
    async def get_device_code() -> models.DeviceCode:
        """
        Gets the device and user codes required for authorization.

        This is the first step in the device authentication flow.

        Returns:
            A `DeviceCode` object containing the codes needed for the next step.

        Example:
            ```python
            from seedrcc import AsyncSeedr

            codes = await AsyncSeedr.get_device_code()
            print(f"Go to {codes.verification_url} and enter {codes.user_code}")
            ```
        """
        params = _request_models.GetDeviceCodeParams()
        async with httpx.AsyncClient() as client:
            response = await AsyncSeedr._make_http_request(
                client, "get", _constants.DEVICE_CODE_URL, params=params.to_dict()
            )

            if not response.is_success:
                raise APIError("Failed to get device code.", response=response)

            try:
                response_data = response.json()
            except json.JSONDecodeError as e:
                raise APIError("Invalid JSON response from API.", response=None) from e
        return models.DeviceCode.from_dict(response_data)

    @classmethod
    async def from_password(
        cls: Type["AsyncSeedr"],
        username: str,
        password: str,
        on_token_refresh: Optional[Callable[[Token], None]] = None,
        httpx_client: Optional[httpx.AsyncClient] = None,
        timeout: float = 30.0,
        proxy: Optional[Dict[str, str]] = None,
        **httpx_kwargs: Any,
    ) -> "AsyncSeedr":
        """
        Creates a new client by authenticating with a username and password.

        Args:
            username: The user's Seedr username (email).
            password: The user's Seedr password.
            on_token_refresh: A callback function that is called with the new
                Token object when the session is refreshed.
            httpx_client: An optional, pre-configured `httpx.AsyncClient` instance.
            timeout: The timeout for network requests in seconds.
            proxy: A dictionary of proxy to use for requests.
            **httpx_kwargs: Optional keyword arguments to pass to the `httpx.AsyncClient` constructor.
                These are ignored if `httpx_client` is provided.

        Returns:
            An initialized `AsyncSeedr` client instance.

        Example:
            ```python
            client = await AsyncSeedr.from_password("your_email@example.com", "your_password")
            ```
        """

        async def auth_callable(client: httpx.AsyncClient) -> Dict[str, Any]:
            """Prepare and execute the authentication request."""
            payload = _request_models.PasswordLoginPayload(username=username, password=password)
            return await cls._authenticate_and_get_token_data(
                client,
                "post",
                _constants.TOKEN_URL,
                data=payload.to_dict(),
            )

        return await cls._initialize_client(
            auth_callable,
            lambda r: {},
            on_token_refresh,
            httpx_client,
            timeout=timeout,
            proxy=proxy,
            **httpx_kwargs,
        )

    @classmethod
    async def from_device_code(
        cls: Type["AsyncSeedr"],
        device_code: str,
        on_token_refresh: Optional[Callable[[Token], None]] = None,
        httpx_client: Optional[httpx.AsyncClient] = None,
        timeout: float = 30.0,
        proxy: Optional[Dict[str, str]] = None,
        **httpx_kwargs: Any,
    ) -> "AsyncSeedr":
        """
        Creates a new client by authorizing with a device code.

        This is the second step in the device authentication flow, after getting the
        codes from `AsyncSeedr.get_device_code()`.

        Args:
            device_code: The device code obtained from `get_device_code()`.
            on_token_refresh: A callback function that is called with the new
                Token object when the session is refreshed.
            httpx_client: An optional, pre-configured `httpx.AsyncClient` instance.
            timeout: The timeout for network requests in seconds.
            proxy: A dictionary of proxy to use for requests.
            **httpx_kwargs: Optional keyword arguments to pass to the `httpx.AsyncClient` constructor.
                These are ignored if `httpx_client` is provided.

        Returns:
            An initialized `AsyncSeedr` client instance.

        Example:
            ```python
            client = await AsyncSeedr.from_device_code("your_device_code")
            ```
        """

        async def auth_callable(client: httpx.AsyncClient) -> Dict[str, Any]:
            """Prepare and execute the device authorization request."""
            params = _request_models.DeviceCodeAuthParams(device_code=device_code)
            return await cls._authenticate_and_get_token_data(
                client,
                "get",
                _constants.DEVICE_AUTHORIZE_URL,
                params=params.to_dict(),
            )

        return await cls._initialize_client(
            auth_callable,
            lambda r: {"device_code": device_code},
            on_token_refresh,
            httpx_client,
            timeout=timeout,
            proxy=proxy,
            **httpx_kwargs,
        )

    @classmethod
    async def from_refresh_token(
        cls: Type["AsyncSeedr"],
        refresh_token: str,
        on_token_refresh: Optional[Callable[[Token], None]] = None,
        httpx_client: Optional[httpx.AsyncClient] = None,
        timeout: float = 30.0,
        proxy: Optional[Dict[str, str]] = None,
        **httpx_kwargs: Any,
    ) -> "AsyncSeedr":
        """
        Creates a new client by using an existing refresh token.

        Args:
            refresh_token: A valid refresh token.
            on_token_refresh: A callback function that is called with the new
                Token object when the session is refreshed.
            httpx_client: An optional, pre-configured `httpx.AsyncClient` instance.
            timeout: The timeout for network requests in seconds.
            proxy: A dictionary of proxy to use for requests.
            **httpx_kwargs: Optional keyword arguments to pass to the `httpx.AsyncClient` constructor.
                These are ignored if `httpx_client` is provided.

        Returns:
            An initialized `AsyncSeedr` client instance.

        Example:
            ```python
            client = await AsyncSeedr.from_refresh_token("your_refresh_token")
            ```
        """

        async def auth_callable(client: httpx.AsyncClient) -> Dict[str, Any]:
            """Prepare and execute the token refresh request."""
            payload = _request_models.RefreshTokenPayload(refresh_token=refresh_token)
            return await cls._authenticate_and_get_token_data(
                client,
                "post",
                _constants.TOKEN_URL,
                data=payload.to_dict(),
            )

        return await cls._initialize_client(
            auth_callable,
            lambda r: {"refresh_token": refresh_token},
            on_token_refresh,
            httpx_client,
            timeout=timeout,
            proxy=proxy,
            **httpx_kwargs,
        )

    async def refresh_token(self) -> models.RefreshTokenResult:
        """
        Manually refreshes the access token.

        This is useful if you want to proactively manage the token's lifecycle
        instead of waiting for an automatic refresh on an API call.

        Returns:
            The result of the token refresh operation.

        Example:
            ```python
            try:
                result = await client.refresh_token()
                print(f"Token successfully refreshed. New token expires in {result.expires_in} seconds.")
            except AuthenticationError as e:
                print(f"Failed to refresh token: {e}")
            ```
        """
        return await self._refresh_access_token()

    async def get_settings(self) -> models.UserSettings:
        """
        Get the user settings.

        Returns:
            An object containing the user's account settings.

        Example:
            ```python
            settings = await client.get_settings()
            print(settings.account.username)
            ```
        """
        response_data = await self._api_request("get", "get_settings")
        return models.UserSettings.from_dict(response_data)

    async def get_memory_bandwidth(self) -> models.MemoryBandwidth:
        """
        Get the memory and bandwidth usage.

        Returns:
            An object containing memory and bandwidth details.

        Example:
            ```python
            usage = await client.get_memory_bandwidth()
            print(f"Space used: {usage.space_used}/{usage.space_max}")
            ```
        """
        response_data = await self._api_request("get", "get_memory_bandwidth")
        return models.MemoryBandwidth.from_dict(response_data)

    async def list_contents(self, folder_id: str = "0") -> models.ListContentsResult:
        """
        List the contents of a folder.

        Args:
            folder_id (str, optional): The folder id to list the contents of. Defaults to root folder.

        Returns:
            An object containing the contents of the folder.

        Example:
            ```python
            response = await client.list_contents()
            print(response)
            ```
        """
        payload = _request_models.ListContentsPayload(content_id=folder_id)
        response_data = await self._api_request("post", "list_contents", data=payload.to_dict())
        return models.ListContentsResult.from_dict(response_data)

    async def add_torrent(
        self,
        magnet_link: Optional[str] = None,
        torrent_file: Optional[str] = None,
        wishlist_id: Optional[str] = None,
        folder_id: str = "0",
    ) -> models.AddTorrentResult:
        """
        Add a torrent to the seedr account for downloading.

        Args:
            magnet_link (str, optional): The magnet link of the torrent.
            torrent_file (str, optional): Remote or local path of the torrent file.
            wishlist_id (str, optional): The ID of a wishlist item to add.
            folder_id (str, optional): The folder ID to add the torrent to. Defaults to root ('-1').

        Returns:
            An object containing the result of the add torrent operation.

        Example:
            ```python
            # Add by magnet link
            result = await client.add_torrent(magnet_link="magnet:?xt=urn:btih:...")
            print(result.title)

            # Add from a local .torrent file
            result = await client.add_torrent(torrent_file="/path/to/your/file.torrent")
            print(result.title)
            ```
        """
        payload = _request_models.AddTorrentPayload(
            torrent_magnet=magnet_link,
            wishlist_id=wishlist_id,
            folder_id=folder_id,
        )
        files = {}
        if torrent_file:
            files = await self._read_torrent_file_async(torrent_file)

        response_data = await self._api_request("post", "add_torrent", data=payload.to_dict(), files=files)
        return models.AddTorrentResult.from_dict(response_data)

    async def scan_page(self, url: str) -> models.ScanPageResult:
        """
        Scan a page for torrents and magnet links.

        Args:
            url (str): The URL of the page to scan.

        Returns:
            An object containing the list of torrents found on the page.

        Example:
            ```python
            result = await client.scan_page(url='some_torrent_page_url')
            for torrent in result.torrents:
                print(torrent.title)
            ```
        """
        payload = _request_models.ScanPagePayload(url=url)
        response_data = await self._api_request("post", "scan_page", data=payload.to_dict())
        return models.ScanPageResult.from_dict(response_data)

    async def fetch_file(self, file_id: str) -> models.FetchFileResult:
        """
        Create a link of a file.

        Args:
            file_id (str): The file id to fetch. This is the `folder_file_id` from the `list_contents` method.

        Returns:
            An object containing the file details and download URL.

        Example:
            ```python
            result = await client.fetch_file(file_id='12345')
            print(f"Download URL: {result.url}")
            ```
        """
        payload = _request_models.FetchFilePayload(folder_file_id=file_id)
        response_data = await self._api_request("post", "fetch_file", data=payload.to_dict())
        return models.FetchFileResult.from_dict(response_data)

    async def create_archive(self, folder_id: str) -> models.CreateArchiveResult:
        """
        Create an archive link of a folder.

        Args:
            folder_id (str): The folder id to create the archive of.

        Returns:
            An object containing the result of the archive creation.

        Example:
            ```python
            result = await client.create_archive(folder_id='12345')
            print(f"Archive URL: {result.archive_url}")
            ```
        """
        payload = _request_models.CreateArchivePayload(folder_id=folder_id)
        response_data = await self._api_request("post", "create_empty_archive", data=payload.to_dict())
        return models.CreateArchiveResult.from_dict(response_data)

    async def search_files(self, query: str) -> models.Folder:
        """
        Search for files.

        Args:
            query (str): The query to search for.

        Returns:
            An object containing the search results.

        Example:
            ```python
            results = await client.search_files(query='harry potter')
            for f in results.folders:
                print(f"Found folder: {f.name}")
            ```
        """
        payload = _request_models.SearchFilesPayload(search_query=query)
        response_data = await self._api_request("post", "search_files", data=payload.to_dict())
        return models.Folder.from_dict(response_data)

    async def add_folder(self, name: str) -> models.APIResult:
        """
        Add a folder.

        Args:
            name (str): Folder name to add.

        Returns:
            An object indicating the result of the operation.

        Example:
            ```python
            result = await client.add_folder(name='New Folder')
            if result.result:
                print("Folder created successfully.")
            ```
        """
        payload = _request_models.AddFolderPayload(name=name)
        response_data = await self._api_request("post", "add_folder", data=payload.to_dict())
        return models.APIResult.from_dict(response_data)

    async def rename_file(self, file_id: str, rename_to: str) -> models.APIResult:
        """
        Rename a file.

        Args:
            file_id (str): The file id to rename.
            rename_to (str): The new name of the file.

        Returns:
            An object indicating the result of the operation.

        Example:
            ```python
            result = await client.rename_file(file_id='12345', rename_to='newName')
            if result.result:
                print("File renamed successfully.")
            ```
        """
        payload = _request_models.RenameFilePayload(rename_to=rename_to, file_id=file_id)
        response_data = await self._api_request("post", "rename", data=payload.to_dict())
        return models.APIResult.from_dict(response_data)

    async def rename_folder(self, folder_id: str, rename_to: str) -> models.APIResult:
        """
        Rename a folder.

        Args:
            folder_id (str): The folder id to rename.
            rename_to (str): The new name of the folder.

        Returns:
            An object indicating the result of the operation.

        Example:
            ```python
            result = await client.rename_folder(folder_id='12345', rename_to='newName')
            if result.result:
                print("Folder renamed successfully.")
            ```
        """
        payload = _request_models.RenameFolderPayload(rename_to=rename_to, folder_id=folder_id)
        response_data = await self._api_request("post", "rename", data=payload.to_dict())
        return models.APIResult.from_dict(response_data)

    async def delete_file(self, file_id: str) -> models.APIResult:
        """
        Delete a file.

        Args:
            file_id (str): The file id to delete.

        Returns:
            An object indicating the result of the operation.

        Example:
            ```python
            response = await client.delete_file(file_id='12345')
            print(response)
            ```
        """
        return await self._delete_api_item("file", file_id)

    async def delete_folder(self, folder_id: str) -> models.APIResult:
        """
        Delete a folder.

        Args:
            folder_id (str): The folder id to delete.

        Returns:
            An object indicating the result of the operation.

        Example:
            ```python
            response = await client.delete_folder(folder_id='12345')
            print(response)
            ```
        """
        return await self._delete_api_item("folder", folder_id)

    async def delete_torrent(self, torrent_id: str) -> models.APIResult:
        """
        Delete an active downloading torrent.

        Args:
            torrent_id (str): The torrent id to delete.

        Returns:
            An object indicating the result of the operation.

        Example:
            ```python
            response = await client.delete_torrent(torrent_id='12345')
            print(response)
            ```
        """
        return await self._delete_api_item("torrent", torrent_id)

    async def delete_wishlist(self, wishlist_id: str) -> models.APIResult:
        """
        Delete an item from the wishlist.

        Args:
            wishlist_id (str): The wishlistId of item to delete.

        Returns:
            An object indicating the result of the operation.

        Example:
            ```python
            result = await client.delete_wishlist(wishlist_id='12345')
            ```
        """
        payload = _request_models.RemoveWishlistPayload(id=wishlist_id)
        response_data = await self._api_request("post", "remove_wishlist", data=payload.to_dict())
        return models.APIResult.from_dict(response_data)

    async def get_devices(self) -> List[models.Device]:
        """
        Get the devices connected to the seedr account.

        Returns:
            A list of devices connected to the account.

        Example:
            ```python
            devices = await client.get_devices()
            for device in devices:
                print(device.client_name)
            ```
        """
        response_data = await self._api_request("get", "get_devices")
        devices_data = response_data.get("devices", [])
        return [models.Device.from_dict(d) for d in devices_data]

    async def get_torrent_progress(self, progress_url: str) -> models.TorrentProgress:
        """
        Fetch and parse the progress data for an active torrent download.

        Args:
            progress_url (str): The progress URL from a `Torrent` object's `progress_url` field.

        Returns:
            A `TorrentProgress` object containing download stats.

        Example:
            ```python
            contents = await client.list_contents()
            for torrent in contents.torrents:
                if torrent.progress_url:
                    progress = await client.get_torrent_progress(torrent.progress_url)
                    print(f"{progress.title}: {progress.progress:.1f}%")
            ```
        """
        url = httpx.URL(progress_url).copy_remove_param("callback")
        response = await self._make_http_request(self._client, "get", str(url))
        try:
            data = response.json()
        except json.JSONDecodeError as e:
            raise APIError("Invalid JSON response from progress URL.", response=response) from e
        return models.TorrentProgress.from_dict(data)

    async def change_name(self, name: str, password: str) -> models.APIResult:
        """
        Change the name of the account.

        Args:
            name (str): The new name of the account.
            password (str): The password of the account.

        Returns:
            An object indicating the result of the operation.

        Example:
            ```python
            result = await client.change_name(name='New Name', password='password')
            ```
        """
        payload = _request_models.ChangeNamePayload(fullname=name, password=password)
        response_data = await self._api_request("post", "user_account_modify", data=payload.to_dict())
        return models.APIResult.from_dict(response_data)

    async def change_password(self, old_password: str, new_password: str) -> models.APIResult:
        """
        Change the password of the account.

        Args:
            old_password (str): The old password of the account.
            new_password (str): The new password of the account.

        Returns:
            An object indicating the result of the operation.

        Example:
            ```python
            result = await client.change_password(old_password='old', new_password='new')
            ```
        """
        payload = _request_models.ChangePasswordPayload(
            password=old_password,
            new_password=new_password,
            new_password_repeat=new_password,
        )
        response_data = await self._api_request("post", "user_account_modify", data=payload.to_dict())
        return models.APIResult.from_dict(response_data)

    async def _api_request(
        self, http_method: str, func: str, files: Optional[Dict[str, Any]] = None, **kwargs: Any
    ) -> Dict[str, Any]:
        """Handles the core logic for making authenticated API requests, including token refreshes."""
        url = kwargs.pop("url", _constants.RESOURCE_URL)
        params = kwargs.pop("params", {})
        if "access_token" not in params:
            params["access_token"] = self._token.access_token
        if func:
            params["func"] = func

        response = await self._make_http_request(self._client, http_method, url, params=params, files=files, **kwargs)
        try:
            data = response.json()
        except json.JSONDecodeError as e:
            raise APIError("Invalid JSON response from API.", response=None) from e

        if isinstance(data, dict) and data.get("error") == "expired_token":
            await self._refresh_access_token()
            params["access_token"] = self._token.access_token
            response = await self._make_http_request(
                self._client, http_method, url, params=params, files=files, **kwargs
            )
            try:
                data = response.json()
            except json.JSONDecodeError as e:
                raise APIError("Invalid JSON response from API.", response=None) from e

        if response.is_client_error:
            if response.status_code == 401:
                raise AuthenticationError("Authentication failed.", response=response)
            raise APIError("API request failed.", response=response)

        if isinstance(data, dict) and data.get("result", True) is not True:
            raise APIError("API operation failed.", response=response)

        return data

    async def _refresh_access_token(self) -> models.RefreshTokenResult:
        """Refreshes the access token using the refresh token or device code."""
        if self._token.refresh_token:
            payload = _request_models.RefreshTokenPayload(refresh_token=self._token.refresh_token)
            response = await self._make_http_request(self._client, "post", _constants.TOKEN_URL, data=payload.to_dict())
        elif self._token.device_code:
            params = _request_models.DeviceCodeAuthParams(device_code=self._token.device_code)
            response = await self._make_http_request(
                self._client, "get", _constants.DEVICE_AUTHORIZE_URL, params=params.to_dict()
            )
        else:
            raise AuthenticationError("No refresh token or device code available to refresh the session.")

        if not response.is_success:
            raise AuthenticationError("Failed to refresh token.", response=response)

        try:
            response_data = response.json()
        except json.JSONDecodeError as e:
            raise APIError("Invalid JSON response from API.", response=None) from e

        if "access_token" not in response_data:
            raise AuthenticationError(
                "Token refresh failed. The response did not contain a new access token.",
                response=response,
            )

        self._token = Token(
            access_token=response_data["access_token"],
            refresh_token=self._token.refresh_token,
            device_code=self._token.device_code,
        )
        if self._on_token_refresh:
            if inspect.iscoroutinefunction(self._on_token_refresh):
                await self._on_token_refresh(self._token)
            else:
                await anyio.to_thread.run_sync(self._on_token_refresh, self._token)

        return models.RefreshTokenResult.from_dict(response_data)

    async def _read_torrent_file_async(self, torrent_file: str) -> Dict[str, Any]:
        """Asynchronously reads a torrent file from a local path or a remote URL into memory."""
        if torrent_file.startswith(("http://", "https://")):
            async with httpx.AsyncClient() as client:
                response = await client.get(torrent_file)
                response.raise_for_status()
                return {"torrent_file": response.content}
        else:
            path = anyio.Path(torrent_file)
            content = await path.read_bytes()
            return {"torrent_file": content}

    async def _delete_api_item(self, item_type: Literal["file", "folder", "torrent"], item_id: str) -> models.APIResult:
        """Constructs and sends a request to delete a specific item (file, folder, etc.)."""
        payload = _request_models.DeleteItemPayload(item_type=item_type, item_id=item_id)
        response_data = await self._api_request("post", "delete", data=payload.to_dict())
        return models.APIResult.from_dict(response_data)

    @classmethod
    async def _initialize_client(
        cls: Type["AsyncSeedr"],
        auth_callable: Callable[[httpx.AsyncClient], Coroutine[Any, Any, Dict[str, Any]]],
        token_callable: Callable[[Dict[str, Any]], Dict[str, Any]],
        on_token_refresh: Optional[Callable[[Token], None]],
        httpx_client: Optional[httpx.AsyncClient],
        timeout: float = 30.0,
        proxy: Optional[Dict[str, str]] = None,
        **httpx_kwargs: Any,
    ) -> "AsyncSeedr":
        """A factory helper that orchestrates the authentication process and constructs the client."""
        httpx_kwargs.setdefault("timeout", timeout)
        httpx_kwargs.setdefault("proxy", proxy)
        client = httpx_client or httpx.AsyncClient(**httpx_kwargs)
        success = False
        try:
            response_data = await auth_callable(client)
            token_extras = token_callable(response_data)
            token = Token(
                access_token=response_data["access_token"],
                refresh_token=response_data.get("refresh_token"),
                **token_extras,
            )
            instance = cls(token, on_token_refresh=on_token_refresh, httpx_client=client, **httpx_kwargs)
            success = True
            return instance
        finally:
            if httpx_client is None and not success:
                await client.aclose()

    @classmethod
    async def _authenticate_and_get_token_data(
        cls: Type["AsyncSeedr"],
        client: httpx.AsyncClient,
        method: str,
        url: str,
        **httpx_kwargs: Any,
    ) -> Dict[str, Any]:
        """Handles the common logic for making an asynchronous authentication request."""
        response = await cls._make_http_request(client, method, url, **httpx_kwargs)

        if not response.is_success:
            raise AuthenticationError("Authentication failed.", response=response)

        try:
            data = response.json()
            if isinstance(data, dict) and data.get("error") in ["authorization_pending"]:
                raise AuthenticationError(
                    "Authentication failed.",
                    response=response,
                )
            return data
        except json.JSONDecodeError as e:
            raise APIError("Invalid JSON response from API.", response=None) from e

    @staticmethod
    async def _make_http_request(
        client: httpx.AsyncClient,
        method: str,
        url: str,
        **kwargs: Any,
    ) -> httpx.Response:
        """Performs the raw HTTP request, handles network/server errors, and returns the response."""
        try:
            response = await client.request(method, url, **kwargs)

            if response.is_server_error:
                raise ServerError(response=response)

            return response
        except httpx.RequestError as e:
            raise NetworkError(str(e)) from e

    async def close(self) -> None:
        """Closes the underlying httpx client if it was created by this instance."""
        if self._manages_client_lifecycle:
            await self._client.aclose()

    async def __aenter__(self) -> "AsyncSeedr":
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        await self.close()

__init__(token, on_token_refresh=None, httpx_client=None, timeout=30.0, proxy=None, **httpx_kwargs)

Initializes the asynchronous client with an existing token.

Parameters:

Name	Type	Description	Default
token	Token	An authenticated Token object.	required
on_token_refresh	Optional[Callable[[Token], None]]	An optional callback function that is called with the new Token object when the session is refreshed.	None
httpx_client	Optional[AsyncClient]	An optional, pre-configured httpx.AsyncClient instance.	None
timeout	float	The timeout for network requests in seconds.	30.0
proxy	Optional[Dict[str, str]]	An optional dictionary of proxy to use for requests.	None
**httpx_kwargs	Any	Optional keyword arguments to pass to the httpx.AsyncClient constructor. These are ignored if httpx_client is provided.	{}

Source code in seedrcc/async_client.py

def __init__(
    self,
    token: Token,
    on_token_refresh: Optional[Callable[[Token], None]] = None,
    httpx_client: Optional[httpx.AsyncClient] = None,
    timeout: float = 30.0,
    proxy: Optional[Dict[str, str]] = None,
    **httpx_kwargs: Any,
) -> None:
    """Initializes the asynchronous client with an existing token.

    Args:
        token: An authenticated `Token` object.
        on_token_refresh: An optional callback function that is called with the new
            `Token` object when the session is refreshed.
        httpx_client: An optional, pre-configured `httpx.AsyncClient` instance.
        timeout: The timeout for network requests in seconds.
        proxy: An optional dictionary of proxy to use for requests.
        **httpx_kwargs: Optional keyword arguments to pass to the `httpx.AsyncClient` constructor.
            These are ignored if `httpx_client` is provided.
    """
    super().__init__(token, on_token_refresh)
    if httpx_client is not None:
        self._client = httpx_client
        self._manages_client_lifecycle = False
    else:
        httpx_kwargs.setdefault("timeout", timeout)
        httpx_kwargs.setdefault("proxy", proxy)
        self._client = httpx.AsyncClient(**httpx_kwargs)
        self._manages_client_lifecycle = True

get_device_code() async staticmethod

Gets the device and user codes required for authorization.

This is the first step in the device authentication flow.

Returns:

Type	Description
DeviceCode	A DeviceCode object containing the codes needed for the next step.

Example

from seedrcc import AsyncSeedr

codes = await AsyncSeedr.get_device_code()
print(f"Go to {codes.verification_url} and enter {codes.user_code}")

Source code in seedrcc/async_client.py

@staticmethod
async def get_device_code() -> models.DeviceCode:
    """
    Gets the device and user codes required for authorization.

    This is the first step in the device authentication flow.

    Returns:
        A `DeviceCode` object containing the codes needed for the next step.

    Example:
        ```python
        from seedrcc import AsyncSeedr

        codes = await AsyncSeedr.get_device_code()
        print(f"Go to {codes.verification_url} and enter {codes.user_code}")
        ```
    """
    params = _request_models.GetDeviceCodeParams()
    async with httpx.AsyncClient() as client:
        response = await AsyncSeedr._make_http_request(
            client, "get", _constants.DEVICE_CODE_URL, params=params.to_dict()
        )

        if not response.is_success:
            raise APIError("Failed to get device code.", response=response)

        try:
            response_data = response.json()
        except json.JSONDecodeError as e:
            raise APIError("Invalid JSON response from API.", response=None) from e
    return models.DeviceCode.from_dict(response_data)

from_password(username, password, on_token_refresh=None, httpx_client=None, timeout=30.0, proxy=None, **httpx_kwargs) async classmethod

Creates a new client by authenticating with a username and password.

Parameters:

Name	Type	Description	Default
username	str	The user's Seedr username (email).	required
password	str	The user's Seedr password.	required
on_token_refresh	Optional[Callable[[Token], None]]	A callback function that is called with the new Token object when the session is refreshed.	None
httpx_client	Optional[AsyncClient]	An optional, pre-configured httpx.AsyncClient instance.	None
timeout	float	The timeout for network requests in seconds.	30.0
proxy	Optional[Dict[str, str]]	A dictionary of proxy to use for requests.	None
**httpx_kwargs	Any	Optional keyword arguments to pass to the httpx.AsyncClient constructor. These are ignored if httpx_client is provided.	{}

Returns:

Type	Description
AsyncSeedr	An initialized AsyncSeedr client instance.

Example

client = await AsyncSeedr.from_password("your_email@example.com", "your_password")

Source code in seedrcc/async_client.py

@classmethod
async def from_password(
    cls: Type["AsyncSeedr"],
    username: str,
    password: str,
    on_token_refresh: Optional[Callable[[Token], None]] = None,
    httpx_client: Optional[httpx.AsyncClient] = None,
    timeout: float = 30.0,
    proxy: Optional[Dict[str, str]] = None,
    **httpx_kwargs: Any,
) -> "AsyncSeedr":
    """
    Creates a new client by authenticating with a username and password.

    Args:
        username: The user's Seedr username (email).
        password: The user's Seedr password.
        on_token_refresh: A callback function that is called with the new
            Token object when the session is refreshed.
        httpx_client: An optional, pre-configured `httpx.AsyncClient` instance.
        timeout: The timeout for network requests in seconds.
        proxy: A dictionary of proxy to use for requests.
        **httpx_kwargs: Optional keyword arguments to pass to the `httpx.AsyncClient` constructor.
            These are ignored if `httpx_client` is provided.

    Returns:
        An initialized `AsyncSeedr` client instance.

    Example:
        ```python
        client = await AsyncSeedr.from_password("your_email@example.com", "your_password")
        ```
    """

    async def auth_callable(client: httpx.AsyncClient) -> Dict[str, Any]:
        """Prepare and execute the authentication request."""
        payload = _request_models.PasswordLoginPayload(username=username, password=password)
        return await cls._authenticate_and_get_token_data(
            client,
            "post",
            _constants.TOKEN_URL,
            data=payload.to_dict(),
        )

    return await cls._initialize_client(
        auth_callable,
        lambda r: {},
        on_token_refresh,
        httpx_client,
        timeout=timeout,
        proxy=proxy,
        **httpx_kwargs,
    )

from_device_code(device_code, on_token_refresh=None, httpx_client=None, timeout=30.0, proxy=None, **httpx_kwargs) async classmethod

Creates a new client by authorizing with a device code.

This is the second step in the device authentication flow, after getting the codes from AsyncSeedr.get_device_code().

Parameters:

Name	Type	Description	Default
device_code	str	The device code obtained from get_device_code().	required
on_token_refresh	Optional[Callable[[Token], None]]	A callback function that is called with the new Token object when the session is refreshed.	None
httpx_client	Optional[AsyncClient]	An optional, pre-configured httpx.AsyncClient instance.	None
timeout	float	The timeout for network requests in seconds.	30.0
proxy	Optional[Dict[str, str]]	A dictionary of proxy to use for requests.	None
**httpx_kwargs	Any	Optional keyword arguments to pass to the httpx.AsyncClient constructor. These are ignored if httpx_client is provided.	{}

Returns:

Type	Description
AsyncSeedr	An initialized AsyncSeedr client instance.

Example

client = await AsyncSeedr.from_device_code("your_device_code")

Source code in seedrcc/async_client.py

@classmethod
async def from_device_code(
    cls: Type["AsyncSeedr"],
    device_code: str,
    on_token_refresh: Optional[Callable[[Token], None]] = None,
    httpx_client: Optional[httpx.AsyncClient] = None,
    timeout: float = 30.0,
    proxy: Optional[Dict[str, str]] = None,
    **httpx_kwargs: Any,
) -> "AsyncSeedr":
    """
    Creates a new client by authorizing with a device code.

    This is the second step in the device authentication flow, after getting the
    codes from `AsyncSeedr.get_device_code()`.

    Args:
        device_code: The device code obtained from `get_device_code()`.
        on_token_refresh: A callback function that is called with the new
            Token object when the session is refreshed.
        httpx_client: An optional, pre-configured `httpx.AsyncClient` instance.
        timeout: The timeout for network requests in seconds.
        proxy: A dictionary of proxy to use for requests.
        **httpx_kwargs: Optional keyword arguments to pass to the `httpx.AsyncClient` constructor.
            These are ignored if `httpx_client` is provided.

    Returns:
        An initialized `AsyncSeedr` client instance.

    Example:
        ```python
        client = await AsyncSeedr.from_device_code("your_device_code")
        ```
    """

    async def auth_callable(client: httpx.AsyncClient) -> Dict[str, Any]:
        """Prepare and execute the device authorization request."""
        params = _request_models.DeviceCodeAuthParams(device_code=device_code)
        return await cls._authenticate_and_get_token_data(
            client,
            "get",
            _constants.DEVICE_AUTHORIZE_URL,
            params=params.to_dict(),
        )

    return await cls._initialize_client(
        auth_callable,
        lambda r: {"device_code": device_code},
        on_token_refresh,
        httpx_client,
        timeout=timeout,
        proxy=proxy,
        **httpx_kwargs,
    )

from_refresh_token(refresh_token, on_token_refresh=None, httpx_client=None, timeout=30.0, proxy=None, **httpx_kwargs) async classmethod

Creates a new client by using an existing refresh token.

Parameters:

Name	Type	Description	Default
refresh_token	str	A valid refresh token.	required
on_token_refresh	Optional[Callable[[Token], None]]	A callback function that is called with the new Token object when the session is refreshed.	None
httpx_client	Optional[AsyncClient]	An optional, pre-configured httpx.AsyncClient instance.	None
timeout	float	The timeout for network requests in seconds.	30.0
proxy	Optional[Dict[str, str]]	A dictionary of proxy to use for requests.	None
**httpx_kwargs	Any	Optional keyword arguments to pass to the httpx.AsyncClient constructor. These are ignored if httpx_client is provided.	{}

Returns:

Type	Description
AsyncSeedr	An initialized AsyncSeedr client instance.

Example

client = await AsyncSeedr.from_refresh_token("your_refresh_token")

Source code in seedrcc/async_client.py

@classmethod
async def from_refresh_token(
    cls: Type["AsyncSeedr"],
    refresh_token: str,
    on_token_refresh: Optional[Callable[[Token], None]] = None,
    httpx_client: Optional[httpx.AsyncClient] = None,
    timeout: float = 30.0,
    proxy: Optional[Dict[str, str]] = None,
    **httpx_kwargs: Any,
) -> "AsyncSeedr":
    """
    Creates a new client by using an existing refresh token.

    Args:
        refresh_token: A valid refresh token.
        on_token_refresh: A callback function that is called with the new
            Token object when the session is refreshed.
        httpx_client: An optional, pre-configured `httpx.AsyncClient` instance.
        timeout: The timeout for network requests in seconds.
        proxy: A dictionary of proxy to use for requests.
        **httpx_kwargs: Optional keyword arguments to pass to the `httpx.AsyncClient` constructor.
            These are ignored if `httpx_client` is provided.

    Returns:
        An initialized `AsyncSeedr` client instance.

    Example:
        ```python
        client = await AsyncSeedr.from_refresh_token("your_refresh_token")
        ```
    """

    async def auth_callable(client: httpx.AsyncClient) -> Dict[str, Any]:
        """Prepare and execute the token refresh request."""
        payload = _request_models.RefreshTokenPayload(refresh_token=refresh_token)
        return await cls._authenticate_and_get_token_data(
            client,
            "post",
            _constants.TOKEN_URL,
            data=payload.to_dict(),
        )

    return await cls._initialize_client(
        auth_callable,
        lambda r: {"refresh_token": refresh_token},
        on_token_refresh,
        httpx_client,
        timeout=timeout,
        proxy=proxy,
        **httpx_kwargs,
    )

refresh_token() async

Manually refreshes the access token.

This is useful if you want to proactively manage the token's lifecycle instead of waiting for an automatic refresh on an API call.

Returns:

Type	Description
RefreshTokenResult	The result of the token refresh operation.

Example

try:
    result = await client.refresh_token()
    print(f"Token successfully refreshed. New token expires in {result.expires_in} seconds.")
except AuthenticationError as e:
    print(f"Failed to refresh token: {e}")

Source code in seedrcc/async_client.py

async def refresh_token(self) -> models.RefreshTokenResult:
    """
    Manually refreshes the access token.

    This is useful if you want to proactively manage the token's lifecycle
    instead of waiting for an automatic refresh on an API call.

    Returns:
        The result of the token refresh operation.

    Example:
        ```python
        try:
            result = await client.refresh_token()
            print(f"Token successfully refreshed. New token expires in {result.expires_in} seconds.")
        except AuthenticationError as e:
            print(f"Failed to refresh token: {e}")
        ```
    """
    return await self._refresh_access_token()

get_settings() async

Get the user settings.

Returns:

Type	Description
UserSettings	An object containing the user's account settings.

Example

settings = await client.get_settings()
print(settings.account.username)

Source code in seedrcc/async_client.py

async def get_settings(self) -> models.UserSettings:
    """
    Get the user settings.

    Returns:
        An object containing the user's account settings.

    Example:
        ```python
        settings = await client.get_settings()
        print(settings.account.username)
        ```
    """
    response_data = await self._api_request("get", "get_settings")
    return models.UserSettings.from_dict(response_data)

get_memory_bandwidth() async

Get the memory and bandwidth usage.

Returns:

Type	Description
MemoryBandwidth	An object containing memory and bandwidth details.

Example

usage = await client.get_memory_bandwidth()
print(f"Space used: {usage.space_used}/{usage.space_max}")

Source code in seedrcc/async_client.py

async def get_memory_bandwidth(self) -> models.MemoryBandwidth:
    """
    Get the memory and bandwidth usage.

    Returns:
        An object containing memory and bandwidth details.

    Example:
        ```python
        usage = await client.get_memory_bandwidth()
        print(f"Space used: {usage.space_used}/{usage.space_max}")
        ```
    """
    response_data = await self._api_request("get", "get_memory_bandwidth")
    return models.MemoryBandwidth.from_dict(response_data)

list_contents(folder_id='0') async

List the contents of a folder.

Parameters:

Name	Type	Description	Default
folder_id	str	The folder id to list the contents of. Defaults to root folder.	'0'

Returns:

Type	Description
ListContentsResult	An object containing the contents of the folder.

Example

response = await client.list_contents()
print(response)

Source code in seedrcc/async_client.py

async def list_contents(self, folder_id: str = "0") -> models.ListContentsResult:
    """
    List the contents of a folder.

    Args:
        folder_id (str, optional): The folder id to list the contents of. Defaults to root folder.

    Returns:
        An object containing the contents of the folder.

    Example:
        ```python
        response = await client.list_contents()
        print(response)
        ```
    """
    payload = _request_models.ListContentsPayload(content_id=folder_id)
    response_data = await self._api_request("post", "list_contents", data=payload.to_dict())
    return models.ListContentsResult.from_dict(response_data)

add_torrent(magnet_link=None, torrent_file=None, wishlist_id=None, folder_id='0') async

Add a torrent to the seedr account for downloading.

Parameters:

Name	Type	Description	Default
magnet_link	str	The magnet link of the torrent.	None
torrent_file	str	Remote or local path of the torrent file.	None
wishlist_id	str	The ID of a wishlist item to add.	None
folder_id	str	The folder ID to add the torrent to. Defaults to root ('-1').	'0'

Returns:

Type	Description
AddTorrentResult	An object containing the result of the add torrent operation.

Example

# Add by magnet link
result = await client.add_torrent(magnet_link="magnet:?xt=urn:btih:...")
print(result.title)

# Add from a local .torrent file
result = await client.add_torrent(torrent_file="/path/to/your/file.torrent")
print(result.title)

Source code in seedrcc/async_client.py

async def add_torrent(
    self,
    magnet_link: Optional[str] = None,
    torrent_file: Optional[str] = None,
    wishlist_id: Optional[str] = None,
    folder_id: str = "0",
) -> models.AddTorrentResult:
    """
    Add a torrent to the seedr account for downloading.

    Args:
        magnet_link (str, optional): The magnet link of the torrent.
        torrent_file (str, optional): Remote or local path of the torrent file.
        wishlist_id (str, optional): The ID of a wishlist item to add.
        folder_id (str, optional): The folder ID to add the torrent to. Defaults to root ('-1').

    Returns:
        An object containing the result of the add torrent operation.

    Example:
        ```python
        # Add by magnet link
        result = await client.add_torrent(magnet_link="magnet:?xt=urn:btih:...")
        print(result.title)

        # Add from a local .torrent file
        result = await client.add_torrent(torrent_file="/path/to/your/file.torrent")
        print(result.title)
        ```
    """
    payload = _request_models.AddTorrentPayload(
        torrent_magnet=magnet_link,
        wishlist_id=wishlist_id,
        folder_id=folder_id,
    )
    files = {}
    if torrent_file:
        files = await self._read_torrent_file_async(torrent_file)

    response_data = await self._api_request("post", "add_torrent", data=payload.to_dict(), files=files)
    return models.AddTorrentResult.from_dict(response_data)

scan_page(url) async

Scan a page for torrents and magnet links.

Parameters:

Name	Type	Description	Default
url	str	The URL of the page to scan.	required

Returns:

Type	Description
ScanPageResult	An object containing the list of torrents found on the page.

Example

result = await client.scan_page(url='some_torrent_page_url')
for torrent in result.torrents:
    print(torrent.title)

Source code in seedrcc/async_client.py

async def scan_page(self, url: str) -> models.ScanPageResult:
    """
    Scan a page for torrents and magnet links.

    Args:
        url (str): The URL of the page to scan.

    Returns:
        An object containing the list of torrents found on the page.

    Example:
        ```python
        result = await client.scan_page(url='some_torrent_page_url')
        for torrent in result.torrents:
            print(torrent.title)
        ```
    """
    payload = _request_models.ScanPagePayload(url=url)
    response_data = await self._api_request("post", "scan_page", data=payload.to_dict())
    return models.ScanPageResult.from_dict(response_data)

fetch_file(file_id) async

Create a link of a file.

Parameters:

Name	Type	Description	Default
file_id	str	The file id to fetch. This is the folder_file_id from the list_contents method.	required

Returns:

Type	Description
FetchFileResult	An object containing the file details and download URL.

Example

result = await client.fetch_file(file_id='12345')
print(f"Download URL: {result.url}")

Source code in seedrcc/async_client.py

async def fetch_file(self, file_id: str) -> models.FetchFileResult:
    """
    Create a link of a file.

    Args:
        file_id (str): The file id to fetch. This is the `folder_file_id` from the `list_contents` method.

    Returns:
        An object containing the file details and download URL.

    Example:
        ```python
        result = await client.fetch_file(file_id='12345')
        print(f"Download URL: {result.url}")
        ```
    """
    payload = _request_models.FetchFilePayload(folder_file_id=file_id)
    response_data = await self._api_request("post", "fetch_file", data=payload.to_dict())
    return models.FetchFileResult.from_dict(response_data)

create_archive(folder_id) async

Create an archive link of a folder.

Parameters:

Name	Type	Description	Default
folder_id	str	The folder id to create the archive of.	required

Returns:

Type	Description
CreateArchiveResult	An object containing the result of the archive creation.

Example

result = await client.create_archive(folder_id='12345')
print(f"Archive URL: {result.archive_url}")

Source code in seedrcc/async_client.py

async def create_archive(self, folder_id: str) -> models.CreateArchiveResult:
    """
    Create an archive link of a folder.

    Args:
        folder_id (str): The folder id to create the archive of.

    Returns:
        An object containing the result of the archive creation.

    Example:
        ```python
        result = await client.create_archive(folder_id='12345')
        print(f"Archive URL: {result.archive_url}")
        ```
    """
    payload = _request_models.CreateArchivePayload(folder_id=folder_id)
    response_data = await self._api_request("post", "create_empty_archive", data=payload.to_dict())
    return models.CreateArchiveResult.from_dict(response_data)

search_files(query) async

Search for files.

Parameters:

Name	Type	Description	Default
query	str	The query to search for.	required

Returns:

Type	Description
Folder	An object containing the search results.

Example

results = await client.search_files(query='harry potter')
for f in results.folders:
    print(f"Found folder: {f.name}")

Source code in seedrcc/async_client.py

async def search_files(self, query: str) -> models.Folder:
    """
    Search for files.

    Args:
        query (str): The query to search for.

    Returns:
        An object containing the search results.

    Example:
        ```python
        results = await client.search_files(query='harry potter')
        for f in results.folders:
            print(f"Found folder: {f.name}")
        ```
    """
    payload = _request_models.SearchFilesPayload(search_query=query)
    response_data = await self._api_request("post", "search_files", data=payload.to_dict())
    return models.Folder.from_dict(response_data)

add_folder(name) async

Add a folder.

Parameters:

Name	Type	Description	Default
name	str	Folder name to add.	required

Returns:

Type	Description
APIResult	An object indicating the result of the operation.

Example

result = await client.add_folder(name='New Folder')
if result.result:
    print("Folder created successfully.")

Source code in seedrcc/async_client.py

async def add_folder(self, name: str) -> models.APIResult:
    """
    Add a folder.

    Args:
        name (str): Folder name to add.

    Returns:
        An object indicating the result of the operation.

    Example:
        ```python
        result = await client.add_folder(name='New Folder')
        if result.result:
            print("Folder created successfully.")
        ```
    """
    payload = _request_models.AddFolderPayload(name=name)
    response_data = await self._api_request("post", "add_folder", data=payload.to_dict())
    return models.APIResult.from_dict(response_data)

rename_file(file_id, rename_to) async

Rename a file.

Parameters:

Name	Type	Description	Default
file_id	str	The file id to rename.	required
rename_to	str	The new name of the file.	required

Returns:

Type	Description
APIResult	An object indicating the result of the operation.

Example

result = await client.rename_file(file_id='12345', rename_to='newName')
if result.result:
    print("File renamed successfully.")

Source code in seedrcc/async_client.py

async def rename_file(self, file_id: str, rename_to: str) -> models.APIResult:
    """
    Rename a file.

    Args:
        file_id (str): The file id to rename.
        rename_to (str): The new name of the file.

    Returns:
        An object indicating the result of the operation.

    Example:
        ```python
        result = await client.rename_file(file_id='12345', rename_to='newName')
        if result.result:
            print("File renamed successfully.")
        ```
    """
    payload = _request_models.RenameFilePayload(rename_to=rename_to, file_id=file_id)
    response_data = await self._api_request("post", "rename", data=payload.to_dict())
    return models.APIResult.from_dict(response_data)

rename_folder(folder_id, rename_to) async

Rename a folder.

Parameters:

Name	Type	Description	Default
folder_id	str	The folder id to rename.	required
rename_to	str	The new name of the folder.	required

Returns:

Type	Description
APIResult	An object indicating the result of the operation.

Example

result = await client.rename_folder(folder_id='12345', rename_to='newName')
if result.result:
    print("Folder renamed successfully.")

Source code in seedrcc/async_client.py

async def rename_folder(self, folder_id: str, rename_to: str) -> models.APIResult:
    """
    Rename a folder.

    Args:
        folder_id (str): The folder id to rename.
        rename_to (str): The new name of the folder.

    Returns:
        An object indicating the result of the operation.

    Example:
        ```python
        result = await client.rename_folder(folder_id='12345', rename_to='newName')
        if result.result:
            print("Folder renamed successfully.")
        ```
    """
    payload = _request_models.RenameFolderPayload(rename_to=rename_to, folder_id=folder_id)
    response_data = await self._api_request("post", "rename", data=payload.to_dict())
    return models.APIResult.from_dict(response_data)

delete_file(file_id) async

Delete a file.

Parameters:

Name	Type	Description	Default
file_id	str	The file id to delete.	required

Returns:

Type	Description
APIResult	An object indicating the result of the operation.

Example

response = await client.delete_file(file_id='12345')
print(response)

Source code in seedrcc/async_client.py

async def delete_file(self, file_id: str) -> models.APIResult:
    """
    Delete a file.

    Args:
        file_id (str): The file id to delete.

    Returns:
        An object indicating the result of the operation.

    Example:
        ```python
        response = await client.delete_file(file_id='12345')
        print(response)
        ```
    """
    return await self._delete_api_item("file", file_id)

delete_folder(folder_id) async

Delete a folder.

Parameters:

Name	Type	Description	Default
folder_id	str	The folder id to delete.	required

Returns:

Type	Description
APIResult	An object indicating the result of the operation.

Example

response = await client.delete_folder(folder_id='12345')
print(response)

Source code in seedrcc/async_client.py

async def delete_folder(self, folder_id: str) -> models.APIResult:
    """
    Delete a folder.

    Args:
        folder_id (str): The folder id to delete.

    Returns:
        An object indicating the result of the operation.

    Example:
        ```python
        response = await client.delete_folder(folder_id='12345')
        print(response)
        ```
    """
    return await self._delete_api_item("folder", folder_id)

delete_torrent(torrent_id) async

Delete an active downloading torrent.

Parameters:

Name	Type	Description	Default
torrent_id	str	The torrent id to delete.	required

Returns:

Type	Description
APIResult	An object indicating the result of the operation.

Example

response = await client.delete_torrent(torrent_id='12345')
print(response)

Source code in seedrcc/async_client.py

async def delete_torrent(self, torrent_id: str) -> models.APIResult:
    """
    Delete an active downloading torrent.

    Args:
        torrent_id (str): The torrent id to delete.

    Returns:
        An object indicating the result of the operation.

    Example:
        ```python
        response = await client.delete_torrent(torrent_id='12345')
        print(response)
        ```
    """
    return await self._delete_api_item("torrent", torrent_id)

delete_wishlist(wishlist_id) async

Delete an item from the wishlist.

Parameters:

Name	Type	Description	Default
wishlist_id	str	The wishlistId of item to delete.	required

Returns:

Type	Description
APIResult	An object indicating the result of the operation.

Example

result = await client.delete_wishlist(wishlist_id='12345')

Source code in seedrcc/async_client.py

async def delete_wishlist(self, wishlist_id: str) -> models.APIResult:
    """
    Delete an item from the wishlist.

    Args:
        wishlist_id (str): The wishlistId of item to delete.

    Returns:
        An object indicating the result of the operation.

    Example:
        ```python
        result = await client.delete_wishlist(wishlist_id='12345')
        ```
    """
    payload = _request_models.RemoveWishlistPayload(id=wishlist_id)
    response_data = await self._api_request("post", "remove_wishlist", data=payload.to_dict())
    return models.APIResult.from_dict(response_data)

get_devices() async

Get the devices connected to the seedr account.

Returns:

Type	Description
List[Device]	A list of devices connected to the account.

Example

devices = await client.get_devices()
for device in devices:
    print(device.client_name)

Source code in seedrcc/async_client.py

async def get_devices(self) -> List[models.Device]:
    """
    Get the devices connected to the seedr account.

    Returns:
        A list of devices connected to the account.

    Example:
        ```python
        devices = await client.get_devices()
        for device in devices:
            print(device.client_name)
        ```
    """
    response_data = await self._api_request("get", "get_devices")
    devices_data = response_data.get("devices", [])
    return [models.Device.from_dict(d) for d in devices_data]

get_torrent_progress(progress_url) async

Fetch and parse the progress data for an active torrent download.

Parameters:

Name	Type	Description	Default
progress_url	str	The progress URL from a Torrent object's progress_url field.	required

Returns:

Type	Description
TorrentProgress	A TorrentProgress object containing download stats.

Example

contents = await client.list_contents()
for torrent in contents.torrents:
    if torrent.progress_url:
        progress = await client.get_torrent_progress(torrent.progress_url)
        print(f"{progress.title}: {progress.progress:.1f}%")

Source code in seedrcc/async_client.py

async def get_torrent_progress(self, progress_url: str) -> models.TorrentProgress:
    """
    Fetch and parse the progress data for an active torrent download.

    Args:
        progress_url (str): The progress URL from a `Torrent` object's `progress_url` field.

    Returns:
        A `TorrentProgress` object containing download stats.

    Example:
        ```python
        contents = await client.list_contents()
        for torrent in contents.torrents:
            if torrent.progress_url:
                progress = await client.get_torrent_progress(torrent.progress_url)
                print(f"{progress.title}: {progress.progress:.1f}%")
        ```
    """
    url = httpx.URL(progress_url).copy_remove_param("callback")
    response = await self._make_http_request(self._client, "get", str(url))
    try:
        data = response.json()
    except json.JSONDecodeError as e:
        raise APIError("Invalid JSON response from progress URL.", response=response) from e
    return models.TorrentProgress.from_dict(data)

change_name(name, password) async

Change the name of the account.

Parameters:

Name	Type	Description	Default
name	str	The new name of the account.	required
password	str	The password of the account.	required

Returns:

Type	Description
APIResult	An object indicating the result of the operation.

Example

result = await client.change_name(name='New Name', password='password')

Source code in seedrcc/async_client.py

async def change_name(self, name: str, password: str) -> models.APIResult:
    """
    Change the name of the account.

    Args:
        name (str): The new name of the account.
        password (str): The password of the account.

    Returns:
        An object indicating the result of the operation.

    Example:
        ```python
        result = await client.change_name(name='New Name', password='password')
        ```
    """
    payload = _request_models.ChangeNamePayload(fullname=name, password=password)
    response_data = await self._api_request("post", "user_account_modify", data=payload.to_dict())
    return models.APIResult.from_dict(response_data)

change_password(old_password, new_password) async

Change the password of the account.

Parameters:

Name	Type	Description	Default
old_password	str	The old password of the account.	required
new_password	str	The new password of the account.	required

Returns:

Type	Description
APIResult	An object indicating the result of the operation.

Example

result = await client.change_password(old_password='old', new_password='new')

Source code in seedrcc/async_client.py

async def change_password(self, old_password: str, new_password: str) -> models.APIResult:
    """
    Change the password of the account.

    Args:
        old_password (str): The old password of the account.
        new_password (str): The new password of the account.

    Returns:
        An object indicating the result of the operation.

    Example:
        ```python
        result = await client.change_password(old_password='old', new_password='new')
        ```
    """
    payload = _request_models.ChangePasswordPayload(
        password=old_password,
        new_password=new_password,
        new_password_repeat=new_password,
    )
    response_data = await self._api_request("post", "user_account_modify", data=payload.to_dict())
    return models.APIResult.from_dict(response_data)

close() async

Closes the underlying httpx client if it was created by this instance.

Source code in seedrcc/async_client.py

async def close(self) -> None:
    """Closes the underlying httpx client if it was created by this instance."""
    if self._manages_client_lifecycle:
        await self._client.aclose()