Use Python logging system for emitting log messages when running via the Python API

.github/workflows/run-tests.yml

@@ -57,4 +57,9 @@ jobs:
         openneuro-py download --help
         openneuro-py login --help
     - name: Test with pytest
-      run: pytest
+      run: pytest -s
+    - name: Test download from CLI
+      run: |
+        openneuro-py download --dataset=ds000248 --include=participants.tsv --target-dir=/tmp/ds000248
+        ls -lah /tmp/ds000248
+

src/openneuro/__init__.py

@@ -14,3 +14,8 @@
 
 from openneuro._download import download as download
 from openneuro._download import login as login
+
+# Assume we're not running from the CLI by default.
+# _cli.download()` or `_cli.login()` will change this.
+# Only used for logging.
+_RUNNING_FROM_CLI = False

src/openneuro/_cli.py

@@ -61,6 +61,8 @@ def download_cli(
     ] = 5,
 ) -> None:
     """Download datasets from OpenNeuro."""
+    openneuro._RUNNING_FROM_CLI = True
+
     download(
         dataset=dataset,
         tag=tag,
@@ -77,6 +79,7 @@ def download_cli(
 @app.command(name="login")
 def login_cli() -> None:
     """Login to OpenNeuro and store an access token."""
+    openneuro._RUNNING_FROM_CLI = True
     login()
 
 

src/openneuro/_config.py

@@ -6,7 +6,8 @@
 from typing import TypedDict
 
 import platformdirs
-from tqdm.auto import tqdm
+
+from openneuro._logging import log
 
 CONFIG_DIR = Path(
     platformdirs.user_config_dir(appname="openneuro-py", appauthor=False, roaming=True)
@@ -25,9 +26,11 @@ class Config(TypedDict):
 
 def init_config() -> None:
     """Initialize a new OpenNeuro configuration file."""
-    tqdm.write(
-        "🙏 Please login to your OpenNeuro account and go to: "
-        "My Account → Obtain an API Key"
+    log(
+        "Please login to your OpenNeuro account and go to: "
+        "My Account → Obtain an API Key",
+        emoji="🙏",
+        end="",
     )
     api_key = getpass.getpass("OpenNeuro API key (input hidden): ")
 

src/openneuro/_download.py

@@ -26,7 +26,6 @@
 import json
 import shlex
 import string
-import sys
 from collections.abc import Generator, Iterable
 from difflib import get_close_matches
 from pathlib import Path, PurePosixPath
@@ -40,14 +39,7 @@
 
 from openneuro import __version__
 from openneuro._config import BASE_URL, get_token, init_config
-
-if hasattr(sys.stdout, "encoding") and sys.stdout.encoding.lower() == "utf-8":
-    stdout_unicode = True
-elif hasattr(sys.stdout, "reconfigure"):
-    sys.stdout.reconfigure(encoding="utf-8")
-    stdout_unicode = True
-else:
-    stdout_unicode = False
+from openneuro._logging import _unicode, log
 
 
 def login() -> None:
@@ -134,7 +126,7 @@ def _safe_query(query, *, timeout=None) -> tuple[dict[str, Any] | None, bool]:
             session.cookies.set_cookie(
                 requests.cookies.create_cookie("accessToken", token)
             )
-            tqdm.write("🍪 Using API token to log in")
+            log("Using API token to log in", emoji="🍪")
         except ValueError:
             pass  # No login
         gql_endpoint = RequestsEndpoint(url=gql_url, session=session, timeout=timeout)
@@ -154,7 +146,7 @@ def _check_snapshot_exists(
     response_json, request_timed_out = _safe_query(query)
 
     if request_timed_out and max_retries > 0:
-        tqdm.write("Request timed out while fetching list of snapshots, retrying …")
+        log("Request timed out while fetching list of snapshots, retrying", emoji="🔄")
         asyncio.sleep(retry_backoff)  # pyright: ignore[reportUnusedCoroutine]
         max_retries -= 1
         retry_backoff *= 2
@@ -216,7 +208,7 @@ def _get_download_metadata(
         request_timed_out = True
 
     if request_timed_out and max_retries > 0:
-        tqdm.write(_unicode("Request timed out while fetching metadata, retrying"))
+        log("Request timed out while fetching metadata, retrying", emoji="🔄")
         asyncio.sleep(retry_backoff)  # pyright: ignore[reportUnusedCoroutine]
         max_retries -= 1
         retry_backoff *= 2
@@ -461,12 +453,10 @@ async def _retry_download(
     semaphore: asyncio.Semaphore,
     query_str: str,
 ) -> None:
-    tqdm.write(
-        _unicode(
-            f"Request timed out while downloading {outfile}, retrying in "
-            f"{retry_backoff} sec",
-            emoji="🔄",
-        )
+    log(
+        f"Request timed out while downloading {outfile}, retrying in "
+        f"{retry_backoff} sec",
+        emoji="🔄",
     )
     await asyncio.sleep(retry_backoff)
     max_retries -= 1
@@ -650,14 +640,6 @@ def _get_local_tag(*, dataset_id: str, dataset_dir: Path) -> str | None:
     return local_version
 
 
-def _unicode(msg: str, *, emoji: str = " ", end: str = "…") -> str:
-    if stdout_unicode:
-        msg = f"{emoji} {msg} {end}"
-    elif end == "…":
-        msg = f"{msg} ..."
-    return msg
-
-
 def _iterate_filenames(
     files: Iterable[dict],
     *,
@@ -798,22 +780,21 @@ def download(
         The maximum number of downloads to run in parallel.
 
     """
-    msg_problems = "problems 🤯" if stdout_unicode else "problems"
-    msg_bugs = "bugs 🪲" if stdout_unicode else "bugs"
-    msg_hello = "👋 Hello!" if stdout_unicode else "Hello!"
-    msg_great_to_see_you = "Great to see you!"
-    if stdout_unicode:
-        msg_great_to_see_you += " 🤗"
-    msg_please = "👉 Please" if stdout_unicode else "   Please"
-
-    msg = (
-        f"\n{msg_hello} This is openneuro-py {__version__}. "
-        f"{msg_great_to_see_you}\n\n"
-        f"   {msg_please} report {msg_problems} and {msg_bugs} at\n"
-        f"      https://github.com/hoechenberger/openneuro-py/issues\n"
+    log(
+        f"Hello! This is openneuro-py {__version__}. ",
+        emoji="👋",
+        end="",
+        cli_only=True,
+    )
+    log("Great to see you!", emoji="🤗", end="")
+    log(
+        "Please report problems and bugs at "
+        "https://github.com/hoechenberger/openneuro-py/issues",
+        emoji="👉",
+        end="\n",
+        cli_only=True,
     )
-    tqdm.write(msg)
-    tqdm.write(_unicode(f"Preparing to download {dataset}", emoji="🌍"))
+    log(f"Preparing to download {dataset}", emoji="🌍")
 
     if target_dir is None:
         target_dir = Path(dataset)
@@ -843,7 +824,7 @@ def download(
             local_tag = _get_local_tag(dataset_id=dataset, dataset_dir=target_dir)
 
             if local_tag is None:
-                tqdm.write(
+                log(
                     "Cannot determine local revision of the dataset, "
                     "and the target directory is not empty. If the "
                     "download fails, you may want to try again with a "
@@ -923,7 +904,7 @@ def download(
         f"Retrieving up to {len(files)} files "
         f"({max_concurrent_downloads} concurrent downloads)."
     )
-    tqdm.write(_unicode(msg, emoji="📥", end=""))
+    log(msg, emoji="📥", end="")
 
     query_str = snapshot_query_template.safe_substitute(
         tag=tag or "null",
@@ -948,5 +929,5 @@ def download(
     except RuntimeError:
         asyncio.run(coroutine)
 
-    tqdm.write(_unicode(f"Finished downloading {dataset}.\n", emoji="✅", end=""))
-    tqdm.write(_unicode("Please enjoy your brains.\n", emoji="🧠", end=""))
+    log(f"Finished downloading {dataset}.", emoji="✅", end="\n")
+    log("Please enjoy your brains.", emoji="🧠", end="\n", cli_only=True)

src/openneuro/_logging.py

@@ -0,0 +1,64 @@
+import logging
+import sys
+
+from tqdm.auto import tqdm
+
+logging.basicConfig(format="%(message)s", level=logging.INFO)
+logger = logging.getLogger("openneuro-py")
+
+
+if hasattr(sys.stdout, "encoding") and sys.stdout.encoding.lower() == "utf-8":
+    stdout_unicode = True
+elif hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8")
+    stdout_unicode = True
+else:
+    stdout_unicode = False
+
+
+def log(
+    message: str,
+    emoji: str | None = None,
+    end: str | None = None,
+    cli_only: bool = False,
+) -> None:
+    """Emit a log message.
+
+    Parameters
+    ----------
+    message
+        The message to emit.
+    emoji
+        Unicode eomji to prepend.
+    end
+        String to append. By default, `"…"`.
+    cli_only
+        Whether to emit the message only when running from the CLI. If `False`, the
+        message will shop up when running from the CLI and the Python API.
+
+    """
+    from openneuro import _RUNNING_FROM_CLI  # avoid circular import
+
+    if cli_only and not _RUNNING_FROM_CLI:
+        return
+
+    if emoji is None:
+        emoji = " "
+    if end is None:
+        end = "…"
+
+    message_unicode = _unicode(message, emoji=emoji, end=end)
+    del message
+
+    if _RUNNING_FROM_CLI:
+        logger.log(level=logging.INFO, msg=message_unicode)
+    else:
+        tqdm.write(message_unicode)
+
+
+def _unicode(msg: str, *, emoji: str = " ", end: str = "…") -> str:
+    if stdout_unicode:
+        msg = f"{emoji} {msg} {end}"
+    elif end == "…":
+        msg = f"{msg} ..."
+    return msg