Support for downloading bulk data files (#44) (#73)

data/bulk/.gitignore

@@ -0,0 +1,5 @@
+# Ignore everything in this directory
+*
+# Except this file and the README
+!.gitignore
+!README.md

data/bulk/README.md

@@ -0,0 +1 @@
+This directory is here for downloaded Scryfall bulk files.

pyproject.toml

@@ -27,6 +27,7 @@ fastapi = "^0.100.0"
 uvicorn = { extras = ["standard"], version = "^0.23.1" }
 motor = "^3.2.0"
 strenum = "^0.4.15"
+requests = "^2.31.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.4.0"

src/scooze/bulkdata.py

@@ -0,0 +1,54 @@
+import os
+
+import requests
+from scooze.enums import ScryfallBulkFile
+
+SCRYFALL_BULK_INFO_ENDPOINT = "https://api.scryfall.com/bulk-data"
+
+
+def download_bulk_data_file(
+    uri: str,
+    bulk_file_type: ScryfallBulkFile | None = None,
+    file_path: str = "data/bulk/",
+) -> None:
+    """
+    Download a single bulk data file from Scryfall.
+    Parameters:
+        uri (str): Location of bulk data file (generally found from bulk info endpoint).
+        bulk_file_type (ScryfallBulkFile): Type of bulk file, used to set filename.
+        file_path (str): Directory to save bulk files. Defaults to `data/bulk/` if not specified.
+    """
+    # TODO(#74): flag for check vs existing file; don't overwrite with same file or older version
+    with requests.get(uri, stream=True) as r:
+        r.raise_for_status()
+        os.makedirs(os.path.dirname(file_path), exist_ok=True)
+        file_name = f"{file_path}{bulk_file_type}.json"
+        with open(file_name, "wb") as f:
+            for chunk in r.iter_content(chunk_size=None):
+                f.write(chunk)
+
+
+def download_all_bulk_data_files(
+    file_path: str = "data/bulk/",
+) -> None:
+    """
+    Download all supported Scryfall bulk data files to local filesystem.
+    Parameters:
+        file_path (str): Directory to save bulk files. Defaults to `data/bulk/` if not specified.
+    """
+    with requests.get(SCRYFALL_BULK_INFO_ENDPOINT) as bulk_metadata_request:
+        bulk_metadata_request.raise_for_status()
+        bulk_metadata = bulk_metadata_request.json()["data"]
+    bulk_files = {t["type"]: t["download_uri"] for t in bulk_metadata}
+
+    for bulk_type in ScryfallBulkFile.list():
+        bulk_filename = bulk_files[bulk_type]
+        download_bulk_data_file(
+            uri=bulk_filename,
+            bulk_file_type=bulk_type,
+            file_path=file_path,
+        )
+
+
+if __name__ == "__main__":
+    download_all_bulk_data_files("../../data/bulk/")

src/scooze/enums.py

@@ -140,4 +140,16 @@ class Rarity(ExtendedEnum, StrEnum):
     BONUS = auto()
 
 
+class ScryfallBulkFile(ExtendedEnum, StrEnum):
+    """
+    URIs for Scryfall bulk files.
+    """
+
+    ORACLE = "oracle_cards"
+    ARTWORK = "unique_artwork"
+    DEFAULT = "default_cards"
+    ALL = "all_cards"
+    # TODO(#26): support for Rulings file
+
+
 # endregion

src/scooze/models/cardparts.py

@@ -91,7 +91,7 @@ class CardFaceModel(BaseModel, validate_assignment=True):
     )
     layout: str | None = Field(
         description="Layout of this face, if any.",
-    ) # TODO(#36): convert to enum?
+    )  # TODO(#36): convert to enum?
     loyalty: int | None = Field(
         description="Starting planeswalker loyalty of this face, if any.",
     )
@@ -200,13 +200,13 @@ class RelatedCardModel(BaseModel, validate_assignment=True):
 
     id: str = Field(
         description="ID of linked component.",
-    ) # NOTE: Scryfall ID
+    )  # NOTE: Scryfall ID
     object: str = Field(
         description="Always `related_card` for this object.",
     )
     component: str = Field(
         description="One of `token`, `meld_part`, `meld_result`, or `combo_piece`.",
-    ) # TODO(#36): convert to enum?
+    )  # TODO(#36): convert to enum?
     name: str = Field(
         description="Name of linked component.",
     )

src/scooze/utils.py

@@ -3,13 +3,16 @@
 from sys import stdout
 
 
+DEFAULT_BULK_FILE_DIR = "./data/bulk/"
+
+
 def get_logger(
     filename: str,
     logger_name: str,
     file_logging_level: int = logging.DEBUG,
     console_logging_level: int = logging.WARNING,
     formatter: logging.Formatter = logging.Formatter("%(asctime)s - %(name)s:%(levelname)s - %(message)s"),
-):
+) -> logging.Logger:
     """
     Helper function to get a new logger.