code cleanup and more unit tests

src/meeple/command/collections.py

@@ -1,7 +1,7 @@
 import click
 
 from meeple.type.collection import Collection
-from meeple.util.collection_util import get_collections, is_pending_updates
+from meeple.util.collection_util import get_collection_names, is_pending_updates
 from meeple.util.data_util import get_collection_data, last_updated
 from meeple.util.fmt_util import fmt_collection_name, fmt_headers
 from meeple.util.message_util import no_collections_exist_error, warn_msg
@@ -21,25 +21,27 @@
 @click.help_option("-h", "--help")
 def collections(sort: str, verbose: bool) -> None:
     """List all collections."""
-    # attempt to retrieve collections
-    collections = get_collections()
+    # attempt to retrieve collection names
+    collection_names = get_collection_names()
 
     # check that local collections exist
-    if not collections:
+    if not collection_names:
         no_collections_exist_error()
 
-    collection_list = []
+    collections = []
     pending_changes = False
-    for collection in collections:
-        board_games, expansions = get_collection_data(collection)
-        collection_list.append(
-            Collection(collection, board_games, expansions, last_updated(collection))
+    for collection_name in collection_names:
+        board_games, expansions = get_collection_data(collection_name)
+        collections.append(
+            Collection(
+                collection_name, board_games, expansions, last_updated(collection_name)
+            )
         )
-        if is_pending_updates(collection):
+        if is_pending_updates(collection_name):
             pending_changes = True
 
     # sort output
-    collection_list, sort_direction = sort_collections(collection_list, sort)
+    collections, sort_direction = sort_collections(collections, sort)
 
     # prepare table data
     headers = [CollectionHeader.NAME]
@@ -56,7 +58,7 @@ def collections(sort: str, verbose: bool) -> None:
     headers = fmt_headers(headers, sort, sort_direction)
 
     rows = []
-    for collection in collection_list:
+    for collection in collections:
         cols = [fmt_collection_name(collection.name)]
         # include additional data if the user chose verbose output
         if verbose:

src/meeple/command/find.py

@@ -1,7 +1,7 @@
 import click
 
 from meeple.type.collection import Collection
-from meeple.util.collection_util import are_collections, get_collections
+from meeple.util.collection_util import are_collections, get_collection_names
 from meeple.util.completion_util import complete_collections
 from meeple.util.data_util import get_collection_data
 from meeple.util.filter_util import filterby_players, filterby_playtime, filterby_weight
@@ -81,7 +81,7 @@ def find(
 
     # if no collections provided, default to all local collections
     if not collections:
-        collections = get_collections()
+        collections = get_collection_names()
 
     # check that local collections exist
     if not collections:

src/meeple/command/open.py

@@ -2,7 +2,7 @@
 
 import click
 
-from meeple.util.api_util import BGG_DOMAIN, get_bgg_items
+from meeple.util.api_util import BGG_DOMAIN, get_bgg_item
 from meeple.util.input_util import bool_input
 from meeple.util.message_util import info_msg, invalid_id_error, print_msg
 
@@ -18,12 +18,11 @@ def open_on_bgg(id: int, yes: bool) -> None:
     """
     # check that the given id is a valid BoardGameGeek ID
     bgg_id = id
-    api_result = get_bgg_items([bgg_id])
-    if not api_result:
+    item = get_bgg_item(bgg_id)
+    if not item:
         invalid_id_error(bgg_id)
 
     # confirm the user wants to open the board game/expansion on BoardGameGeek website
-    item = api_result[0]
     url = f"https://{BGG_DOMAIN}/{item.type}/{bgg_id}"
     name = item.name
     if yes or bool_input(f"Open [i blue]{name}[/i blue] on {BGG_DOMAIN}?"):

src/meeple/command/update.py

@@ -4,7 +4,7 @@
 
 from meeple.util.api_util import BOARDGAME_TYPE, EXPANSION_TYPE, get_bgg_items
 from meeple.util.collection_util import (
-    get_collections,
+    get_collection_names,
     is_collection,
     is_pending_updates,
     read_collection,
@@ -39,7 +39,7 @@ def update(collection: str, force: bool) -> None:
             invalid_collection_error(collection)
         collections = [collection]
     else:
-        collections = get_collections()
+        collections = get_collection_names()
 
     # check that local collections exist
     if not collections:

src/meeple/util/api_util.py

@@ -14,36 +14,76 @@
 EXPANSION_TYPE = "boardgameexpansion"
 
 
-def _bgg_api_get_items(url: str) -> List[Item]:
-    response = requests.get(url)
+def _bgg_api_get_items(endpoint: str, params: dict) -> List[Item]:
+    """Get items from the provided BGG endpoint.
+
+    Args:
+        endpoint (str): BGG endpoint.
+
+    Returns:
+        List[Item]: List of items.
+    """
+    response = requests.get(f"{API2_BASE_URL}/{endpoint}", params=params)
     if response.status_code == 200:
         resp_dict = xmltodict.parse(response.content)
         resp_list = resp_dict["items"].get("item", [])
         if not isinstance(resp_list, list):
             resp_list = [resp_list]
         return [Item.from_bgg_dict(bgg_dict) for bgg_dict in resp_list]
-    # TODO: provide better error handling and logging for bad requests/responses
+    # TODO: log this error and print out a friendly message to the user
     sys.exit(f"Error: HTTP {response.status_code}: {response.content}")
 
 
 def get_bgg_items(bgg_ids: List[int]) -> List[Item]:
-    ids_str = ",".join(map(str, bgg_ids))
-    url = f"{API2_BASE_URL}/thing?id={ids_str}&type={BOARDGAME_TYPE},{EXPANSION_TYPE}&stats=1"
-    return _bgg_api_get_items(url)
+    """Get a list of items with their details from the BGG API.
+
+    Args:
+        bgg_ids (List[int]): Item IDs to fetch.
+
+    Returns:
+        List[Item]: List of items.
+    """
+    params = {
+        "id": ",".join(map(str, bgg_ids)),
+        "type": f"{BOARDGAME_TYPE},{EXPANSION_TYPE}",
+        "stats": 1,
+    }
+    return _bgg_api_get_items(endpoint="thing", params=params)
 
 
 def get_bgg_item(bgg_id: int) -> Item:
+    """Get a single item with its details from the BGG API.
+
+    Args:
+        bgg_id (List[int]): Item ID to fetch.
+
+    Returns:
+        List[Item]: Item.
+    """
     bgg_items = get_bgg_items([bgg_id])
     if bgg_items:
         return bgg_items[0]
     return None
 
 
 def get_bgg_hot() -> List[Item]:
-    url = f"{API2_BASE_URL}/hot?type={BOARDGAME_TYPE}"
-    return _bgg_api_get_items(url)
+    """Get the current BGG Hotness items.
+
+    Returns:
+        List[Item]: List of items.
+    """
+    params = {"type": BOARDGAME_TYPE}
+    return _bgg_api_get_items(endpoint="hot", params=params)
 
 
 def search_bgg(query: str) -> List[Item]:
-    url = f"{API2_BASE_URL}/search?type={BOARDGAME_TYPE}&query={query}"
-    return _bgg_api_get_items(url)
+    """Search BGG for items by name.
+
+    Args:
+        query (str): Search query.
+
+    Returns:
+        List[Item]: List of items.
+    """
+    params = {"type": BOARDGAME_TYPE, "query": query}
+    return _bgg_api_get_items(endpoint="search", params=params)

src/meeple/util/cmd_util.py

@@ -4,7 +4,7 @@
 
 
 class SectionedHelpGroup(Group):
-    """Organize commands as sections"""
+    """Organize commands in sections."""
 
     def __init__(self, *args, **kwargs):
         self.section_commands = collections.defaultdict(list)

src/meeple/util/collection_util.py

@@ -3,9 +3,7 @@
 from pathlib import Path
 from typing import List
 
-import yaml
-
-from meeple.util.fs_util import get_collection_dir
+from meeple.util.fs_util import get_collection_dir, read_yaml_file, write_yaml_file
 
 COLLECTION_DIR = get_collection_dir()
 
@@ -31,82 +29,79 @@ def _get_ids(data: dict, list_key: str) -> List[int]:
     return ids
 
 
-def get_collections() -> List[str]:
+def get_collection_names() -> List[str]:
     # create in_path dir and exit if it does not exist
     if not Path(COLLECTION_DIR).exists():
         Path(COLLECTION_DIR).mkdir(parents=True)
 
     # retrieve collection source files from in_path
     collection_files = next(walk(COLLECTION_DIR))[2]
-    collections = []
+    collection_names = []
     for collection_file in collection_files:
-        collection, ext = splitext(collection_file)
+        collection_name, ext = splitext(collection_file)
         if ext == ".yml":
-            collections.append(collection)
-    return collections
+            collection_names.append(collection_name)
+    return collection_names
 
 
-def is_collection(name: str) -> bool:
-    return name in get_collections()
+def is_collection(collection_name: str) -> bool:
+    return collection_name in get_collection_names()
 
 
-def is_pending_updates(name: str) -> bool:
-    _, to_add_ids, to_drop_ids = read_collection(name)
+def is_pending_updates(collection_name: str) -> bool:
+    _, to_add_ids, to_drop_ids = read_collection(collection_name)
     return len(to_add_ids) > 0 or len(to_drop_ids) > 0
 
 
-def are_collections(names: [str]) -> bool:
-    return set(names) <= set(get_collections())
+def are_collections(collection_names: [str]) -> bool:
+    return set(collection_names) <= set(get_collection_names())
 
 
-def read_collection(name: str) -> (List[int], List[int], List[int]):
-    with open(_collection_file(name), "r") as f:
-        data = yaml.load(f, Loader=yaml.FullLoader)
+def read_collection(collection_name: str) -> (List[int], List[int], List[int]):
+    data = read_yaml_file(_collection_file(collection_name))
 
-        # check if data is in old format
-        # TODO: deprecated - eventually remove
-        if data and _OLD_ITEM_LIST_KEY in data:
-            bgg_ids = data[_OLD_ITEM_LIST_KEY]
-            if not bgg_ids:
-                return [], [], []
+    # check if data is in old format
+    # TODO: deprecated - eventually remove
+    if data and _OLD_ITEM_LIST_KEY in data:
+        bgg_ids = data[_OLD_ITEM_LIST_KEY]
+        if not bgg_ids:
+            return [], [], []
 
-            # remove non int values from list
-            for bgg_id in bgg_ids:
-                if not isinstance(bgg_id, int):
-                    bgg_ids.remove(bgg_id)
-            return bgg_ids, [], []
+        # remove non int values from list
+        for bgg_id in bgg_ids:
+            if not isinstance(bgg_id, int):
+                bgg_ids.remove(bgg_id)
+        return bgg_ids, [], []
 
-        if data:
-            return (
-                _get_ids(data, _ITEM_LIST_KEY),
-                _get_ids(data, _TO_ADD_LIST_KEY),
-                _get_ids(data, _TO_DROP_LIST_KEY),
-            )
-        return [], [], []
+    if data:
+        return (
+            _get_ids(data, _ITEM_LIST_KEY),
+            _get_ids(data, _TO_ADD_LIST_KEY),
+            _get_ids(data, _TO_DROP_LIST_KEY),
+        )
+    return [], [], []
 
 
-def create_collection(name: str) -> None:
+def create_collection(collection_name: str) -> None:
     # TODO: create a class for this Collection object
     data = {_ITEM_LIST_KEY: [], _TO_ADD_LIST_KEY: [], _TO_DROP_LIST_KEY: []}
-    with open(_collection_file(name), "w") as f:
-        yaml.dump(data, f)
+    write_yaml_file(_collection_file(collection_name), data)
 
 
 def update_collection(
-    name: str, item_ids: list, to_add_ids: list, to_drop_ids: list
+    collection_name: str, item_ids: list, to_add_ids: list, to_drop_ids: list
 ) -> None:
     data = {
         _ITEM_LIST_KEY: item_ids,
         _TO_ADD_LIST_KEY: to_add_ids,
         _TO_DROP_LIST_KEY: to_drop_ids,
     }
-    with open(_collection_file(name), "w") as f:
-        yaml.dump(data, f)
+    write_yaml_file(_collection_file(collection_name), data)
 
 
 def rename_collection(current_name: str, new_name: str) -> None:
     Path(_collection_file(current_name)).rename(join(COLLECTION_DIR, f"{new_name}.yml"))
 
 
-def delete_collection(name: str) -> None:
-    Path(_collection_file(name)).unlink()
+def delete_collection(collection_name: str) -> None:
+    Path(_collection_file(collection_name)).unlink()

src/meeple/util/completion_util.py

@@ -1,9 +1,9 @@
-from meeple.util.collection_util import get_collections
+from meeple.util.collection_util import get_collection_names
 
 
 def complete_collections(ctx, param, incomplete):
     return [
         collection
-        for collection in get_collections()
+        for collection in get_collection_names()
         if collection.startswith(incomplete)
     ]

src/meeple/util/data_util.py

@@ -7,7 +7,7 @@
 from typing import List
 
 from meeple.type.item import Item
-from meeple.util.fs_util import get_data_dir
+from meeple.util.fs_util import get_data_dir, read_json_file, write_json_file
 
 DATA_DIR = get_data_dir()
 
@@ -52,15 +52,14 @@ def get_collection_data(collection_name: str) -> (List[Item], List[Item]):
         return board_games, expansions
 
     # get latest collection data
-    with open(data_path, "r") as f:
-        data = json.load(f)
+    data_dict = read_json_file(data_path)
 
-    for dict_item in data[_BOARD_GAME_LIST_KEY]:
+    for item_dict in data_dict[_BOARD_GAME_LIST_KEY]:
         board_games.append(
-            json.loads(json.dumps(dict_item), object_hook=Item.from_json)
+            json.loads(json.dumps(item_dict), object_hook=Item.from_json)
         )
-    for dict_item in data[_EXPANSION_LIST_KEY]:
-        expansions.append(json.loads(json.dumps(dict_item), object_hook=Item.from_json))
+    for item_dict in data_dict[_EXPANSION_LIST_KEY]:
+        expansions.append(json.loads(json.dumps(item_dict), object_hook=Item.from_json))
     return board_games, expansions
 
 
@@ -82,8 +81,7 @@ def write_collection_data(
     }
 
     # persist data
-    with open(join(data_path, filename), "w") as f:
-        json.dump(data_dict, f, indent=4, ensure_ascii=False)
+    write_json_file(join(data_path, filename), data_dict)
 
 
 def delete_collection_data(collection_name: str) -> None: