From ac34a2a767a9d5a5ad683cadfd588bdc32ecf294 Mon Sep 17 00:00:00 2001 From: antazoey Date: Mon, 3 Feb 2025 17:12:32 -0600 Subject: [PATCH 1/4] refactor: wip --- audius/cli/__init__.py | 180 ++++++++++++++++++++++------------------- 1 file changed, 97 insertions(+), 83 deletions(-) diff --git a/audius/cli/__init__.py b/audius/cli/__init__.py index 8e41e17..f4c05cb 100644 --- a/audius/cli/__init__.py +++ b/audius/cli/__init__.py @@ -1,83 +1,97 @@ -import difflib -import re -from typing import Any - -import click - -from audius.cli.config import config -from audius.cli.options import player_option -from audius.cli.playlists import playlists -from audius.cli.tips import tips -from audius.cli.tracks import tracks -from audius.cli.users import users -from audius.cli.utils import sdk -from audius.client_factory import get_hosts -from audius.exceptions import AudiusException -from audius.sdk import Audius - - -class AudiusCLI(click.Group): - def invoke(self, ctx) -> Any: - try: - return super().invoke(ctx) - except click.UsageError as err: - self._suggest_cmd(err) - except AudiusException as err: - raise click.ClickException(f"({type(err).__name__}) {err}") from err - - @staticmethod - def _suggest_cmd(usage_error): - if usage_error.message is None: - raise usage_error - - match = re.match("No such command '(.*)'.", usage_error.message) - if not match: - raise usage_error - - bad_arg = match.groups()[0] - suggested_commands = difflib.get_close_matches( - bad_arg, list(usage_error.ctx.command.commands.keys()), cutoff=0.6 - ) - if suggested_commands: - if bad_arg not in suggested_commands: - usage_error.message = ( - f"No such command '{bad_arg}'. Did you mean {' or '.join(suggested_commands)}?" - ) - - raise usage_error - - -def create_cli(sdk_cls=Audius): - @click.group(cls=AudiusCLI) - def cli(): - "Audius CLI" - - @cli.command() - def hosts(): - """ - List available hosts. - """ - - gen = (f"{x}\n" for x in get_hosts()) - click.echo_via_pager(gen) - - @cli.command() - @sdk.audius() - @click.argument("track_id", required=False) - @player_option() - def play(sdk, track_id, player): - """ - Play something from Audius. - """ - sdk.tracks.play(track_id, player=player) - - cli.add_command(users(sdk_cls)) - cli.add_command(playlists(sdk_cls)) - cli.add_command(tracks(sdk_cls)) - cli.add_command(tips(sdk_cls)) - cli.add_command(config(sdk_cls)) - - return cli - - -audius = create_cli() +from cyclopts import App + +audius = App() + + +@audius.command +def hosts(): + print("hosts") + + +if __name__ == "__main__": + app() + + +# import difflib +# import re +# from typing import Any +# +# import click +# +# from audius.cli.config import config +# from audius.cli.options import player_option +# from audius.cli.playlists import playlists +# from audius.cli.tips import tips +# from audius.cli.tracks import tracks +# from audius.cli.users import users +# from audius.cli.utils import sdk +# from audius.client_factory import get_hosts +# from audius.exceptions import AudiusException +# from audius.sdk import Audius +# +# +# class AudiusCLI(click.Group): +# def invoke(self, ctx) -> Any: +# try: +# return super().invoke(ctx) +# except click.UsageError as err: +# self._suggest_cmd(err) +# except AudiusException as err: +# raise click.ClickException(f"({type(err).__name__}) {err}") from err +# +# @staticmethod +# def _suggest_cmd(usage_error): +# if usage_error.message is None: +# raise usage_error +# +# match = re.match("No such command '(.*)'.", usage_error.message) +# if not match: +# raise usage_error +# +# bad_arg = match.groups()[0] +# suggested_commands = difflib.get_close_matches( +# bad_arg, list(usage_error.ctx.command.commands.keys()), cutoff=0.6 +# ) +# if suggested_commands: +# if bad_arg not in suggested_commands: +# usage_error.message = ( +# f"No such command '{bad_arg}'. Did you mean {' or '.join(suggested_commands)}?" +# ) +# +# raise usage_error +# +# +# def create_cli(sdk_cls=Audius): +# @click.group(cls=AudiusCLI) +# def cli(): +# "Audius CLI" +# +# @cli.command() +# def hosts(): +# """ +# List available hosts. +# """ +# +# gen = (f"{x}\n" for x in get_hosts()) +# click.echo_via_pager(gen) +# +# @cli.command() +# @sdk.audius() +# @click.argument("track_id", required=False) +# @player_option() +# def play(sdk, track_id, player): +# """ +# Play something from Audius. +# """ +# sdk.tracks.play(track_id, player=player) +# +# cli.add_command(users(sdk_cls)) +# cli.add_command(playlists(sdk_cls)) +# cli.add_command(tracks(sdk_cls)) +# cli.add_command(tips(sdk_cls)) +# cli.add_command(config(sdk_cls)) +# +# return cli +# +# +# audius = create_cli() From 7795ad6dc7ddd1ccb2f2c3fa63d7a040501f0ceb Mon Sep 17 00:00:00 2001 From: antazoey Date: Mon, 3 Feb 2025 19:25:09 -0600 Subject: [PATCH 2/4] refactor: use cyclopts --- audius/cli/__init__.py | 118 ++++++---------------- audius/cli/config.py | 33 +++---- audius/cli/options.py | 12 --- audius/cli/playlists.py | 121 +++++++++++------------ audius/cli/tips.py | 110 ++++++++++----------- audius/cli/tracks.py | 180 ++++++++++++++++----------------- audius/cli/users.py | 202 +++++++++++++++++++------------------- audius/cli/utils.py | 22 +---- audius/client_factory.py | 3 +- audius/config.py | 15 +-- audius/const.py | 5 + audius/player/__init__.py | 8 +- audius/player/base.py | 4 +- audius/playlists.py | 4 +- audius/tips.py | 4 +- audius/tracks.py | 13 +-- audius/users.py | 10 +- setup.py | 3 +- tests/test_cli.py | 38 ++++--- 19 files changed, 412 insertions(+), 493 deletions(-) delete mode 100644 audius/cli/options.py create mode 100644 audius/const.py diff --git a/audius/cli/__init__.py b/audius/cli/__init__.py index f4c05cb..b4d952b 100644 --- a/audius/cli/__init__.py +++ b/audius/cli/__init__.py @@ -1,97 +1,39 @@ +from typing import Optional + from cyclopts import App +from audius import Audius +from audius.cli.config import config +from audius.cli.playlists import playlists +from audius.cli.tips import tips +from audius.cli.tracks import tracks +from audius.cli.users import users +from audius.cli.utils import print +from audius.client_factory import get_hosts +from audius.types import PlayerType + audius = App() +audius.command(config) +audius.command(playlists) +audius.command(tips) +audius.command(tracks) +audius.command(users) @audius.command def hosts(): - print("hosts") - + """ + List available hosts + """ + all_hosts = list(get_hosts()) + for host in all_hosts: + print(host) -if __name__ == "__main__": - app() - -# import difflib -# import re -# from typing import Any -# -# import click -# -# from audius.cli.config import config -# from audius.cli.options import player_option -# from audius.cli.playlists import playlists -# from audius.cli.tips import tips -# from audius.cli.tracks import tracks -# from audius.cli.users import users -# from audius.cli.utils import sdk -# from audius.client_factory import get_hosts -# from audius.exceptions import AudiusException -# from audius.sdk import Audius -# -# -# class AudiusCLI(click.Group): -# def invoke(self, ctx) -> Any: -# try: -# return super().invoke(ctx) -# except click.UsageError as err: -# self._suggest_cmd(err) -# except AudiusException as err: -# raise click.ClickException(f"({type(err).__name__}) {err}") from err -# -# @staticmethod -# def _suggest_cmd(usage_error): -# if usage_error.message is None: -# raise usage_error -# -# match = re.match("No such command '(.*)'.", usage_error.message) -# if not match: -# raise usage_error -# -# bad_arg = match.groups()[0] -# suggested_commands = difflib.get_close_matches( -# bad_arg, list(usage_error.ctx.command.commands.keys()), cutoff=0.6 -# ) -# if suggested_commands: -# if bad_arg not in suggested_commands: -# usage_error.message = ( -# f"No such command '{bad_arg}'. Did you mean {' or '.join(suggested_commands)}?" -# ) -# -# raise usage_error -# -# -# def create_cli(sdk_cls=Audius): -# @click.group(cls=AudiusCLI) -# def cli(): -# "Audius CLI" -# -# @cli.command() -# def hosts(): -# """ -# List available hosts. -# """ -# -# gen = (f"{x}\n" for x in get_hosts()) -# click.echo_via_pager(gen) -# -# @cli.command() -# @sdk.audius() -# @click.argument("track_id", required=False) -# @player_option() -# def play(sdk, track_id, player): -# """ -# Play something from Audius. -# """ -# sdk.tracks.play(track_id, player=player) -# -# cli.add_command(users(sdk_cls)) -# cli.add_command(playlists(sdk_cls)) -# cli.add_command(tracks(sdk_cls)) -# cli.add_command(tips(sdk_cls)) -# cli.add_command(config(sdk_cls)) -# -# return cli -# -# -# audius = create_cli() +@audius.command +def play(track_id: Optional[str] = None, player: Optional[PlayerType] = None): + """ + Play something from Audius + """ + sdk = Audius() + sdk.tracks.play(track_id, player=player) diff --git a/audius/cli/config.py b/audius/cli/config.py index 0c44504..079fd4c 100644 --- a/audius/cli/config.py +++ b/audius/cli/config.py @@ -1,26 +1,15 @@ -from typing import Type +from cyclopts import App -import click +from audius.cli.utils import print +from audius.sdk import Audius -from audius.cli.utils import sdk +config = App(name="config", help="Manage config") -def config(sdk_cls: Type): - sdk.py = sdk_cls - - @click.group(name="config") - def cli(): - """ - Show configuration. - """ - - @cli.command() - @sdk.audius() - def app_name(sdk): - """ - Show the app name. - """ - - click.echo(sdk.config.app_name) - - return cli +@config.command +def app_name(): + """ + Show the application name + """ + sdk = Audius() + print(sdk.config.app_name) diff --git a/audius/cli/options.py b/audius/cli/options.py deleted file mode 100644 index 572138b..0000000 --- a/audius/cli/options.py +++ /dev/null @@ -1,12 +0,0 @@ -import click - -from audius.types import PlayerType - - -def player_option(): - return click.option( - "--player", - help="The player to use.", - type=click.Choice([x.value for x in PlayerType.__members__.values()], case_sensitive=False), - callback=lambda _, _2, val: PlayerType(val) if val else None, - ) diff --git a/audius/cli/playlists.py b/audius/cli/playlists.py index e83354a..028936c 100644 --- a/audius/cli/playlists.py +++ b/audius/cli/playlists.py @@ -1,61 +1,60 @@ -from typing import Type - -import click - -from audius.cli.utils import sdk - - -def playlists(sdk_cls: Type): - sdk.py = sdk_cls - - @click.group(name="playlists") - def cli(): - """ - Manage playlists. - """ - - @cli.command() - @sdk.audius() - def trending(sdk): - """ - Page through trending playlists. - """ - - trending = list(sdk.playlists.trending()) - gen = (f"{i + 1}: {x['playlist_name']} (id={x['id']})\n" for i, x in enumerate(trending)) - click.echo_via_pager(gen) - - @cli.command() - @sdk.audius() - @click.argument("playlist_id") - def get(sdk, playlist_id): - """ - Get a playlist. - """ - - playlist = sdk.playlists.get(playlist_id) - _echo_playlist(playlist) - - @cli.command() - @sdk.audius() - @click.argument("query") - def search(sdk, query): - """ - Search through playlists. - """ - - result = sdk.playlists.search(query=query) - for idx, playlist in enumerate(result): - _echo_playlist(playlist) - - if idx < len(result) - 1: - click.echo() - - return cli - - -def _echo_playlist(playlist: dict): - click.echo(f"Playlist: {playlist['playlist_name']} (id={playlist['id']})") - click.echo(f"Description: {playlist['description']}") - click.echo(f"User: {playlist['user']['name']} (id={playlist['user']['id']})") - click.echo(f"Total plays: {playlist['total_play_count']}") +from cyclopts import App + +from audius.cli.utils import print +from audius.sdk import Audius + +playlists = App(name="playlists", help="View playlists") + + +@playlists.command +def trending(top: int = 10): + """ + Show top trending playlists + + Args: + top (int): The number of playlists to show + """ + sdk = Audius() + count = 0 + for idx, trending_playlist in enumerate(sdk.playlists.trending()): + print(f"{idx + 1}: {trending_playlist['playlist_name']} (id={trending_playlist['id']})") + count += 1 + if count >= top: + break + + +@playlists.command +def get(playlist_id: str): + """ + Get a playlist + + Args: + playlist_id (str): The playlist ID + """ + sdk = Audius() + playlist = sdk.playlists.get(playlist_id) + print_playlist(playlist) + + +@playlists.command() +def search(query: str = ""): + """ + Search through playlists + + Args: + query (str): The search query + """ + sdk = Audius() + result = sdk.playlists.search(query=query) + for idx, playlist in enumerate(result): + print_playlist(playlist) + + if idx < len(result) - 1: + print() + + +def print_playlist(playlist: dict): + print(f"Playlist: {playlist['playlist_name']} (id={playlist['id']})") + print(f"Description: {playlist['description']}") + print(f"User: {playlist['user']['name']} (id={playlist['user']['id']})") + print(f"Total plays: {playlist['total_play_count']}") diff --git a/audius/cli/tips.py b/audius/cli/tips.py index 327a9db..f5626bf 100644 --- a/audius/cli/tips.py +++ b/audius/cli/tips.py @@ -1,61 +1,51 @@ -from typing import Type - -import click - -from audius.cli.utils import sdk - - -def tips(sdk_cls: Type): - sdk.py = sdk_cls - - @click.group(name="tips") - def cli(): - """ - Checkout artist tips. - """ - - @cli.command() - @sdk.audius() - @click.option("--offset", help="Number of tips to skip (for pagination).", type=int) - @click.option("--limit", help="Number of tips to fetch.", type=int) - @click.option("--user-id", help="The user making the request.") - @click.option( - "--receiver-min-followers", help="Exclude recipients without min followers.", type=int +from typing import Optional + +from cyclopts import App + +from audius.cli.utils import print +from audius.sdk import Audius + +tips = App(name="tips", help="View tips") + + +@tips.command +def get( + offset: Optional[int] = None, + limit: Optional[int] = None, + user_id: Optional[str] = None, + receiver_min_followers: Optional[int] = None, + receiver_is_verified: bool = False, + current_user_follows: Optional[str] = None, + unique_by: Optional[str] = None, +) -> None: + """ + Get tips + + Args: + offset (int | None): Number of tips to skip + limit (int | None): Number of tips to fetch + user_id (str | None): User ID + receiver_min_followers (int | None): Exclude recipients without min followers + receiver_is_verified (bool): Exclude non-verified recipients + current_user_follows (str | None): Query by who recipient follows + unique_by (str | None): Require involvement in the given capacity + """ + sdk = Audius() + result = sdk.tips.get( + offset=offset, + limit=limit, + user_id=user_id, + receiver_min_followers=receiver_min_followers, + receiver_is_verified=receiver_is_verified, + current_user_follows=current_user_follows, + unique_by=unique_by, ) - @click.option("--receiver-is-verified", help="Exclude non-verified recipients.", is_flag=True) - @click.option("--current-user-follows", help="Query by who recipient follows.") - @click.option("--unique-by", help="Require involvement in the given capacity.") - def get( - sdk, - offset, - limit, - user_id, - receiver_min_followers, - receiver_is_verified, - current_user_follows, - unique_by, - ): - """ - Get tips. - """ - - result = sdk.tips.get( - offset=offset, - limit=limit, - user_id=user_id, - receiver_min_followers=receiver_min_followers, - receiver_is_verified=receiver_is_verified, - current_user_follows=current_user_follows, - unique_by=unique_by, - ) - if not result: - click.echo("No tips found.") - - else: - for idx, tip in enumerate(result): - click.echo( - f"'{tip['sender']['name']}' tipped '{tip['amount']}' " - f"to '{tip['receiver']['name']}' on '{tip['created_at']}'." - ) - - return cli + if not result: + print("No tips found.") + + else: + for idx, tip in enumerate(result): + print( + f"'{tip['sender']['name']}' tipped '{tip['amount']}' " + f"to '{tip['receiver']['name']}' on '{tip['created_at']}'." + ) diff --git a/audius/cli/tracks.py b/audius/cli/tracks.py index 06a9f69..cfef9f6 100644 --- a/audius/cli/tracks.py +++ b/audius/cli/tracks.py @@ -1,90 +1,92 @@ from pathlib import Path -from typing import Type - -import click - -from audius.cli.options import player_option -from audius.cli.utils import sdk - -DEFAULT_BUFFER_SIZE = 1024 * 1024 - - -def tracks(sdk_cls: Type): - sdk.py = sdk_cls - - @click.group(name="tracks") - def cli(): - """ - Browse and listen to tracks. - """ - - @cli.command() - @sdk.audius() - def trending(sdk): - """ - Page through trending tracks. - """ - - trending = list(sdk.tracks.trending()) - gen = (f"{i + 1}: {x['track_name']} (id={x['id']})\n" for i, x in enumerate(trending)) - click.echo_via_pager(gen) - - @cli.command() - @sdk.audius() - @click.argument("track_id") - def get(sdk, track_id): - """ - Get a track. - """ - - track = sdk.tracks.get(track_id) - _echo_track(track) - - @cli.command() - @sdk.audius() - @click.argument("query") - def search(sdk, query): - """ - Search through tracks. - """ - - result = sdk.tracks.search(query=query) - for idx, track in enumerate(result): - _echo_track(track) - - if idx < len(result) - 1: - click.echo() - - @cli.command() - @sdk.audius() - @click.argument("track_id", required=False) - @player_option() - def play(sdk, track_id, player): - """ - Play a track. - """ - sdk.tracks.play(track_id, player=player) - - @cli.command() - @sdk.audius() - @click.argument("track_id") - @click.argument("out_path", type=Path) - @click.option( - "--buffer-size", help="The buffer size when downloading.", default=DEFAULT_BUFFER_SIZE - ) - def download(sdk, track_id, out_path, buffer_size): - """ - Download a track. - """ - - sdk.tracks.download(track_id, out_path, chunk_size=buffer_size) - - return cli - - -def _echo_track(track: dict): - click.echo(f"Track {track['title']} (id={track['id']})") - click.echo(f"Description: {track['description']}") - click.echo(f"Artist: {track['user']['name']}") - click.echo(f"Genre: {track['genre']}, Mood: {track['mood']}") - click.echo(f"Duration: {track['duration']}") +from typing import Optional + +from cyclopts import App + +from audius.cli.utils import print +from audius.const import DEFAULT_BUFFER_SIZE +from audius.sdk import Audius +from audius.types import PlayerType + +tracks = App(name="tracks", help="View and play tracks") + + +@tracks.command +def trending(top: int = 10): + """ + Show trending tracks + + Args: + top (int): The number to show + """ + sdk = Audius() + count = 0 + for track in sdk.tracks.trending(): + print(f"{track['title']} (id={track['id']})") + count += 1 + if count >= top: + break + + +@tracks.command +def get(track_id: str): + """ + Get a track + + Args: + track_id (str): The track ID + """ + sdk = Audius() + track = sdk.tracks.get(track_id) + print_track(track) + + +@tracks.command +def search(query: str = ""): + """ + Search tracks + + Args: + query (str): The search query + """ + sdk = Audius() + result = sdk.tracks.search(query=query) + for idx, track in enumerate(result): + print_track(track) + if idx < len(result) - 1: + print() + + +@tracks.command(name="play") +def play_track(track_id: Optional[str] = None, player: Optional[PlayerType] = None): + """ + Play a track + + Args: + track_id (str | None): The track ID, defaults to random + player (str | None): The player name, such as 'afplay' or 'vlc' + """ + sdk = Audius() + sdk.tracks.play(track_id, player=player) + + +@tracks.command +def download(track_id: str, out_path: Path, buffer_size: int = DEFAULT_BUFFER_SIZE): + """ + Download a track + + Args: + track_id (str): The track ID + out_path (Path): The output path + buffer_size (int): The buffer size, defaults to 1,048,576 + """ + sdk = Audius() + sdk.tracks.download(track_id, out_path, chunk_size=buffer_size) + + +def print_track(track: dict): + print(f"Track {track['title']} (id={track['id']})") + print(f"Description: {track['description']}") + print(f"Artist: {track['user']['name']}") + print(f"Genre: {track['genre']}, Mood: {track['mood']}") + print(f"Duration: {track['duration']}") diff --git a/audius/cli/users.py b/audius/cli/users.py index 6842b61..02c164c 100644 --- a/audius/cli/users.py +++ b/audius/cli/users.py @@ -1,100 +1,102 @@ -from typing import Type - -import click - -from audius.cli.utils import sdk - - -def users(sdk_cls: Type): - sdk.py = sdk_cls - - @click.group(name="users") - def users(): - """ - Commands for users. - """ - - @users.command() - @sdk.audius() - def top(sdk): - """ - Page through the top users. - """ - - top = list(sdk.users.top()) - gen = (f"{i + 1}: {x['name']} (id={x['id']})\n" for i, x in enumerate(top)) - click.echo_via_pager(gen) - - @users.command() - @sdk.audius() - @click.argument("user_id") - def get(sdk, user_id): - """ - Get a user. - """ - - user = sdk.users.get(user_id) - _echo_user(user) - - @users.command() - @sdk.audius() - @click.argument("query") - def search(sdk, query): - """ - Search for users. - """ - - result = sdk.users.search(query=query) - for idx, user in enumerate(result): - _echo_user(user) - - if idx < len(result) - 1: - click.echo() - - @users.command() - @sdk.audius() - @click.argument("user_id") - def wallets(sdk, user_id): - """ - List user wallet addresses. - """ - - result = sdk.users.get_connected_wallets(user_id) - if result["erc_wallets"]: - click.echo("ERC Wallets:") - for addr in result["erc_wallets"]: - click.echo(f"\t{addr}") - - else: - click.echo("No ERC wallets") - - if result["spl_wallets"]: - # Newline sep. - click.echo() - click.echo("SPL Wallets:") - for addr in result["spl_wallets"]: - click.echo(addr) - - @users.command() - @sdk.audius() - @click.argument("user_id") - def tracks(sdk, user_id): - """ - Get a user's tracks. - """ - - tracks = sdk.users.get_tracks(user_id) - for idx, track in enumerate(tracks): - click.echo(f"Track: {track['title']} (id={track['id']})") - - if idx < len(tracks) - 1: - click.echo() - - return users - - -def _echo_user(user: dict): - click.echo(f"Artist: {user['name']} (id={user['id']})") - click.echo(f"Bio: {user['bio']}") - click.echo(f"Followers: {user['follower_count']}, Followees: {user['followee_count']}") - click.echo(f"ERC Wallet: {user['erc_wallet']}, SPL Wallet: {user['spl_wallet']}") +from cyclopts import App + +from audius.cli.utils import print +from audius.sdk import Audius + +users = App(name="users", help="View users") + + +@users.command(name="top") +def show_top(top: int = 10): + """ + Show top users + + Args: + top (int): Number of users to show + """ + sdk = Audius() + count = 0 + for user in sdk.users.top(): + print(user) + count += 1 + if count >= top: + break + + +@users.command +def get(user_id: str): + """ + Get a user + + Args: + user_id (str): User ID + """ + sdk = Audius() + user = sdk.users.get(user_id) + print_user(user) + + +@users.command +def search(query: str = ""): + """ + Search for users + + Args: + query (str): Search query + """ + sdk = Audius() + result = sdk.users.search(query=query) + for idx, user in enumerate(result): + print_user(user) + + if idx < len(result) - 1: + print() + + +@users.command +def wallets(user_id: str): + """ + List user wallet addresses + + Args: + user_id (str): User ID + """ + sdk = Audius() + result = sdk.users.get_connected_wallets(user_id) + if result["erc_wallets"]: + print("ERC Wallets:") + for addr in result["erc_wallets"]: + print(f"\t{addr}") + + else: + print("No ERC wallets") + + if result["spl_wallets"]: + # Newline sep. + print("\nSPL Wallets:") + for addr in result["spl_wallets"]: + print(addr) + + +@users.command +def tracks(user_id: str): + """ + Show user tracks + + Args: + user_id (str): User ID + """ + sdk = Audius() + user_tracks = sdk.users.get_tracks(user_id) + for idx, track in enumerate(user_tracks): + print(f"Track: {track['title']} (id={track['id']})") + + if idx < len(user_tracks) - 1: + print() + + +def print_user(user: dict): + print(f"Artist: {user['name']} (id={user['id']})") + print(f"Bio: {user['bio']}") + print(f"Followers: {user['follower_count']}, Followees: {user['followee_count']}") + print(f"ERC Wallet: {user['erc_wallet']}, SPL Wallet: {user['spl_wallet']}") diff --git a/audius/cli/utils.py b/audius/cli/utils.py index c032e86..348ce0e 100644 --- a/audius/cli/utils.py +++ b/audius/cli/utils.py @@ -1,19 +1,7 @@ -import click +try: + from rich import print +except ImportError: + print = print -from audius.sdk import Audius - -class sdk: - py = Audius - - @classmethod - def audius(cls): - """ - A click command decorator giving you access to the SDK. - """ - - def decorator(f): - f = click.make_pass_decorator(cls.py, ensure=True)(f) - return f - - return decorator +__all__ = ["print"] diff --git a/audius/client_factory.py b/audius/client_factory.py index dd22326..a6c925b 100644 --- a/audius/client_factory.py +++ b/audius/client_factory.py @@ -1,12 +1,11 @@ from random import randint -from typing import List import requests from audius.client import Client -def get_hosts() -> List[str]: +def get_hosts() -> list[str]: response = requests.get("https://api.audius.co") response.raise_for_status() return response.json().get("data", []) diff --git a/audius/config.py b/audius/config.py index c77df8b..8db6aae 100644 --- a/audius/config.py +++ b/audius/config.py @@ -1,13 +1,14 @@ import os -from typing import Dict, Optional, Union +from typing import Optional, Union +from audius.const import ( + AUDIUS_APP_NAME_ENV_VAR, + AUDIUS_HOST_NAME_ENV_VAR, + AUDIUS_PLAYER_ENV_VAR, + DEFAULT_APP_NAME, +) from audius.player import PlayerType -DEFAULT_APP_NAME = "audius-py" -AUDIUS_APP_NAME_ENV_VAR = "AUDIUS_APP_NAME" -AUDIUS_HOST_NAME_ENV_VAR = "AUDIUS_HOST_NAME" -AUDIUS_PLAYER_ENV_VAR = "AUDIUS_PLAYER" - class Config: """ @@ -18,7 +19,7 @@ def __init__( self, app_name: str = DEFAULT_APP_NAME, host: Optional[str] = None, - aliases: Optional[Dict[str, str]] = None, + aliases: Optional[dict[str, str]] = None, player: Optional[Union[PlayerType, str]] = None, ): self.app_name = app_name diff --git a/audius/const.py b/audius/const.py new file mode 100644 index 0000000..2cf5cb9 --- /dev/null +++ b/audius/const.py @@ -0,0 +1,5 @@ +DEFAULT_BUFFER_SIZE = 1048576 # 1024 * 1024 +DEFAULT_APP_NAME = "audius-py" +AUDIUS_APP_NAME_ENV_VAR = "AUDIUS_APP_NAME" +AUDIUS_HOST_NAME_ENV_VAR = "AUDIUS_HOST_NAME" +AUDIUS_PLAYER_ENV_VAR = "AUDIUS_PLAYER" diff --git a/audius/player/__init__.py b/audius/player/__init__.py index 300df94..b9daaf8 100644 --- a/audius/player/__init__.py +++ b/audius/player/__init__.py @@ -1,4 +1,4 @@ -from typing import TYPE_CHECKING, Dict, Optional, Type +from typing import TYPE_CHECKING, Optional from audius.client import API from audius.exceptions import MissingPlayerError @@ -14,17 +14,17 @@ class Player(API): def __init__(self, sdk: "Audius") -> None: super().__init__(sdk) - self._player_classes: Dict[PlayerType, Type] = { + self._player_classes: dict[PlayerType, type] = { PlayerType.AFPLAY: AFPlayer, PlayerType.VLC: VLCPlayer, } - self._player_map: Dict[PlayerType, BasePlayer] = {} + self._player_map: dict[PlayerType, BasePlayer] = {} def play(self, url: str, player_type: Optional[PlayerType] = None): player = self.get_player(player_type=player_type) player.play(url) - def display_now_playing(self, track: Dict, player_type: Optional[PlayerType] = None): + def display_now_playing(self, track: dict, player_type: Optional[PlayerType] = None): player = self.get_player(player_type=player_type) player.display_now_playing(track) diff --git a/audius/player/base.py b/audius/player/base.py index 60cfc87..1417b88 100644 --- a/audius/player/base.py +++ b/audius/player/base.py @@ -1,5 +1,5 @@ from abc import abstractmethod -from typing import TYPE_CHECKING, Dict +from typing import TYPE_CHECKING import click @@ -28,7 +28,7 @@ def play(self, url: str): Player-subclasses must implement this method. """ - def display_now_playing(self, track: Dict): + def display_now_playing(self, track: dict): click.echo( f"({self._type.value.lower().capitalize()}) " f"Now playing '{track['title']}' by {track['user']['name']}" diff --git a/audius/playlists.py b/audius/playlists.py index 5e7ada0..f343815 100644 --- a/audius/playlists.py +++ b/audius/playlists.py @@ -1,4 +1,4 @@ -from typing import Dict, Iterator, List, Optional +from collections.abc import Iterator from requests.exceptions import HTTPError @@ -22,6 +22,6 @@ def get(self, playlist_id: str): return result.get("data", {}) - def search(self, query: Optional[str] = None) -> List[Dict]: + def search(self, query: str = "") -> list[dict]: result = self.client.get("playlists/search", params={"query": query}) return result.get("data", []) diff --git a/audius/tips.py b/audius/tips.py index 5f37883..41a6322 100644 --- a/audius/tips.py +++ b/audius/tips.py @@ -1,4 +1,4 @@ -from typing import Dict, Optional, Union +from typing import Optional, Union from audius.client import API @@ -33,7 +33,7 @@ def get( involved in the given capacity. """ - params: Dict[str, Union[int, str, bool]] = {} + params: dict[str, Union[int, str, bool]] = {} if offset is not None: params["offset"] = offset if limit is not None: diff --git a/audius/tracks.py b/audius/tracks.py index a0cc4ea..aba85d1 100644 --- a/audius/tracks.py +++ b/audius/tracks.py @@ -1,7 +1,8 @@ import os +from collections.abc import Iterable, Iterator from pathlib import Path from random import randint -from typing import IO, Dict, Iterable, Iterator, List, Optional, Tuple, Union +from typing import IO, Optional, Union import click from requests import Response @@ -21,12 +22,12 @@ def update_to(self, b=1, bsize=1, tsize=None): def _write_response( - output_paths: List[FileDestination], + output_paths: list[FileDestination], response: Response, progress_bar: Optional[DownloadProgressBar] = None, chunk_size: int = 1, ): - output_files: List[Tuple[IO, bool]] = [] + output_files: list[tuple[IO, bool]] = [] for output_path in output_paths: if isinstance(output_path, Path): _file = open(str(output_path), "wb") @@ -53,8 +54,8 @@ def _write_response( def _validate_output_paths( output_paths: Union[FileDestination, Iterable[FileDestination]], -) -> List[FileDestination]: - output_path_ls: List[FileDestination] +) -> list[FileDestination]: + output_path_ls: list[FileDestination] if not isinstance(output_paths, (list, tuple)): output_path_ls = [output_paths] # type: ignore else: @@ -83,7 +84,7 @@ def get(self, track_id: str): return result.get("data", {}) - def search(self, query: Optional[str] = None) -> List[Dict]: + def search(self, query: str = "") -> list[dict]: result = self.client.get("tracks/search", params={"query": query}) return result.get("data", []) diff --git a/audius/users.py b/audius/users.py index 4437bfe..4429a85 100644 --- a/audius/users.py +++ b/audius/users.py @@ -1,4 +1,4 @@ -from typing import Dict, Iterator, List, Optional +from collections.abc import Iterator from requests.exceptions import HTTPError @@ -10,7 +10,7 @@ class Users(API): def top(self) -> Iterator[dict]: yield from self.client.get("users/top").get("data", []) - def get(self, user_id: str) -> Dict: + def get(self, user_id: str) -> dict: user_id = self._handle_id(user_id) try: result = self.client.get(f"users/{user_id}") @@ -22,16 +22,16 @@ def get(self, user_id: str) -> Dict: return result.get("data", {}) - def search(self, query: Optional[str] = None) -> List[Dict]: + def search(self, query: str = "") -> list[dict]: result = self.client.get("users/search", params={"query": query}) return result.get("data", []) - def get_connected_wallets(self, user_id: str) -> Dict: + def get_connected_wallets(self, user_id: str) -> dict: user_id = self._handle_id(user_id) result = self.client.get(f"users/{user_id}/connected_wallets") return result.get("data", {}) - def get_tracks(self, user_id: str) -> List[Dict]: + def get_tracks(self, user_id: str) -> list[dict]: user_id = self._handle_id(user_id) result = self.client.get(f"users/{user_id}/tracks") return result.get("data", []) diff --git a/setup.py b/setup.py index 156cf08..2c52e16 100644 --- a/setup.py +++ b/setup.py @@ -58,8 +58,7 @@ include_package_data=True, install_requires=[ "requests>=2.32.3,<3", - "click>=8.1.8,<9", - "tqdm>=4.67.1,<5", + "" "tqdm>=4.67.1,<5", "afplay-py>=0.2.0,<0.3", ], python_requires=">=3.9,<4", diff --git a/tests/test_cli.py b/tests/test_cli.py index 4c6c3fe..c901006 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,24 +1,38 @@ -import pytest -from click.testing import CliRunner +from subprocess import PIPE, run -from audius.cli import audius +import pytest @pytest.fixture -def runner(): - return CliRunner() +def run_cmd(): + class Result: + def __init__(self, process_result): + self.process_result = process_result + @property + def exit_code(self) -> int: + return self.process_result.returncode -@pytest.fixture -def cli(): - return audius + @property + def output(self) -> str: + return self.process_result.stdout.decode("utf8") + + def fn(*args): + cmd = ["audius", *args] + process_result = run(cmd, stdout=PIPE, stderr=PIPE) + return Result(process_result) + + return fn -def test_cli(runner, cli): - result = runner.invoke(cli, "--help") +def test_cli(run_cmd): + result = run_cmd() assert result.exit_code == 0 + assert "Commands" in result.output + assert "hosts" in result.output + assert "List available hosts" in result.output -def test_app_name(runner, cli): - result = runner.invoke(cli, ["config", "app-name"]) +def test_app_name(run_cmd): + result = run_cmd("config", "app-name") assert result.output == "audius-py\n" From ddb825b3b6cff8da8c9ec4fc62b66a5bfc991947 Mon Sep 17 00:00:00 2001 From: antazoey Date: Mon, 3 Feb 2025 19:30:40 -0600 Subject: [PATCH 3/4] chore: type checking fl8 --- audius/cli/__init__.py | 9 ++++++--- audius/cli/tracks.py | 8 +++++--- audius/player/__init__.py | 6 +++--- audius/player/base.py | 8 +++----- audius/tracks.py | 28 +++++++++++++++------------- setup.cfg | 1 + setup.py | 4 +++- 7 files changed, 36 insertions(+), 28 deletions(-) diff --git a/audius/cli/__init__.py b/audius/cli/__init__.py index b4d952b..84a2edd 100644 --- a/audius/cli/__init__.py +++ b/audius/cli/__init__.py @@ -1,4 +1,4 @@ -from typing import Optional +from typing import TYPE_CHECKING, Optional from cyclopts import App @@ -10,7 +10,10 @@ from audius.cli.users import users from audius.cli.utils import print from audius.client_factory import get_hosts -from audius.types import PlayerType + +if TYPE_CHECKING: + from audius.types import PlayerType + audius = App() audius.command(config) @@ -31,7 +34,7 @@ def hosts(): @audius.command -def play(track_id: Optional[str] = None, player: Optional[PlayerType] = None): +def play(track_id: Optional[str] = None, player: Optional["PlayerType"] = None): """ Play something from Audius """ diff --git a/audius/cli/tracks.py b/audius/cli/tracks.py index cfef9f6..56785e6 100644 --- a/audius/cli/tracks.py +++ b/audius/cli/tracks.py @@ -1,12 +1,14 @@ from pathlib import Path -from typing import Optional +from typing import TYPE_CHECKING, Optional from cyclopts import App from audius.cli.utils import print from audius.const import DEFAULT_BUFFER_SIZE from audius.sdk import Audius -from audius.types import PlayerType + +if TYPE_CHECKING: + from audius.types import PlayerType tracks = App(name="tracks", help="View and play tracks") @@ -58,7 +60,7 @@ def search(query: str = ""): @tracks.command(name="play") -def play_track(track_id: Optional[str] = None, player: Optional[PlayerType] = None): +def play_track(track_id: Optional[str] = None, player: Optional["PlayerType"] = None): """ Play a track diff --git a/audius/player/__init__.py b/audius/player/__init__.py index b9daaf8..5ba050e 100644 --- a/audius/player/__init__.py +++ b/audius/player/__init__.py @@ -3,11 +3,11 @@ from audius.client import API from audius.exceptions import MissingPlayerError from audius.player.af import AFPlayer -from audius.player.base import BasePlayer from audius.player.vlc import VLCPlayer from audius.types import PlayerType if TYPE_CHECKING: + from audius.player.base import BasePlayer from audius.sdk import Audius @@ -18,7 +18,7 @@ def __init__(self, sdk: "Audius") -> None: PlayerType.AFPLAY: AFPlayer, PlayerType.VLC: VLCPlayer, } - self._player_map: dict[PlayerType, BasePlayer] = {} + self._player_map: dict[PlayerType, "BasePlayer"] = {} def play(self, url: str, player_type: Optional[PlayerType] = None): player = self.get_player(player_type=player_type) @@ -28,7 +28,7 @@ def display_now_playing(self, track: dict, player_type: Optional[PlayerType] = N player = self.get_player(player_type=player_type) player.display_now_playing(track) - def get_player(self, player_type: Optional[PlayerType] = None) -> BasePlayer: + def get_player(self, player_type: Optional[PlayerType] = None) -> "BasePlayer": player_type = player_type or self.config.player if player_type is not None: if player_type not in self._player_classes: diff --git a/audius/player/base.py b/audius/player/base.py index 1417b88..35cc208 100644 --- a/audius/player/base.py +++ b/audius/player/base.py @@ -1,17 +1,15 @@ from abc import abstractmethod from typing import TYPE_CHECKING -import click - from audius.client import API -from audius.types import PlayerType if TYPE_CHECKING: from audius.sdk import Audius + from audius.types import PlayerType class BasePlayer(API): - def __init__(self, player_type: PlayerType, sdk: "Audius"): + def __init__(self, player_type: "PlayerType", sdk: "Audius"): self._type = player_type super().__init__(sdk) @@ -29,7 +27,7 @@ def play(self, url: str): """ def display_now_playing(self, track: dict): - click.echo( + print( f"({self._type.value.lower().capitalize()}) " f"Now playing '{track['title']}' by {track['user']['name']}" ) diff --git a/audius/tracks.py b/audius/tracks.py index aba85d1..4159ea6 100644 --- a/audius/tracks.py +++ b/audius/tracks.py @@ -2,16 +2,18 @@ from collections.abc import Iterable, Iterator from pathlib import Path from random import randint -from typing import IO, Optional, Union +from typing import IO, TYPE_CHECKING, Optional, Union -import click -from requests import Response from requests.exceptions import HTTPError from tqdm import tqdm # type: ignore from audius.client import API from audius.exceptions import OutputPathError, TrackNotFoundError -from audius.types import FileDestination, PlayerType + +if TYPE_CHECKING: + from requests import Response + + from audius.types import FileDestination, PlayerType class DownloadProgressBar(tqdm): @@ -22,8 +24,8 @@ def update_to(self, b=1, bsize=1, tsize=None): def _write_response( - output_paths: list[FileDestination], - response: Response, + output_paths: list["FileDestination"], + response: "Response", progress_bar: Optional[DownloadProgressBar] = None, chunk_size: int = 1, ): @@ -53,9 +55,9 @@ def _write_response( def _validate_output_paths( - output_paths: Union[FileDestination, Iterable[FileDestination]], -) -> list[FileDestination]: - output_path_ls: list[FileDestination] + output_paths: Union["FileDestination", Iterable["FileDestination"]], +) -> list["FileDestination"]: + output_path_ls: list["FileDestination"] if not isinstance(output_paths, (list, tuple)): output_path_ls = [output_paths] # type: ignore else: @@ -88,7 +90,7 @@ def search(self, query: str = "") -> list[dict]: result = self.client.get("tracks/search", params={"query": query}) return result.get("data", []) - def play(self, track_id: Optional[str], player: Optional[PlayerType] = None): + def play(self, track_id: Optional[str], player: Optional["PlayerType"] = None): if track_id is None: # Get a random track. result = self.search() @@ -107,7 +109,7 @@ def play(self, track_id: Optional[str], player: Optional[PlayerType] = None): def download( self, track_id: str, - output_paths: Union[FileDestination, Iterable[FileDestination]], + output_paths: Union["FileDestination", Iterable["FileDestination"]], hide_output: bool = False, chunk_size: int = 1, ): @@ -128,10 +130,10 @@ def download( unit="B", unit_scale=True, miniters=1, desc=uri.split("/")[-1] ) track = self.get(track_id) - click.echo(f"Downloading '{track['title']}' by {track['user']['name']}") + print(f"Downloading '{track['title']}' by {track['user']['name']}") dest = ", ".join([str(x) for x in output_path_ls]) - click.echo(f"Saving at '{dest}'.") + print(f"Saving at '{dest}'.") with progress_bar as bar: _write_response(output_path_ls, response, progress_bar=bar) diff --git a/setup.cfg b/setup.cfg index 7da1f96..1293e65 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,2 +1,3 @@ [flake8] max-line-length = 100 +ignore = E704,W503,PYD002,TC003,TC006 diff --git a/setup.py b/setup.py index 2c52e16..626177c 100644 --- a/setup.py +++ b/setup.py @@ -12,6 +12,7 @@ "types-requests", "types-setuptools", "flake8>=7.1.1", + "flake8-type-checking", "isort>=5.13.2", "mdformat>=0.7.21", "mdformat-gfm>=0.3.5", @@ -58,7 +59,8 @@ include_package_data=True, install_requires=[ "requests>=2.32.3,<3", - "" "tqdm>=4.67.1,<5", + "cyclopts>=3.4.1,<4", + "tqdm>=4.67.1,<5", "afplay-py>=0.2.0,<0.3", ], python_requires=">=3.9,<4", From ad6a67760326d8da7ba39261f946df9c6176556d Mon Sep 17 00:00:00 2001 From: antazoey Date: Mon, 3 Feb 2025 19:37:15 -0600 Subject: [PATCH 4/4] test: fix hosts test --- tests/test_cli.py | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/tests/test_cli.py b/tests/test_cli.py index ba50f82..13b552d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -2,6 +2,8 @@ import pytest +from audius.cli import hosts + @pytest.fixture def run_cmd(): @@ -38,14 +40,15 @@ def test_app_name(run_cmd): assert result.output == "audius-py\n" -def test_hosts(mocker, runner, cli): +def test_hosts(mocker, capsys): hosts_patch = mocker.patch("audius.cli.get_hosts") - hosts = ["http://host1.example.com", "https://host2.example.com"] + host_strings = ["http://host1.example.com", "https://host2.example.com"] def patch(): - yield from hosts + yield from host_strings hosts_patch.side_effect = patch - result = runner.invoke(cli, "hosts") - assert result.output == "\n".join(hosts) + "\n" + hosts() + result = capsys.readouterr() + assert result.out == "\n".join(host_strings) + "\n"