Source code for bacommontools.pcommands

# Released under the MIT License. See LICENSE for details.
#
"""Pcommands for bacommontools."""

from __future__ import annotations


[docs] def bacurl() -> None: """Run curl with the Ballistica API key injected. Usage: bacurl [curl-args...] <url> Reads ``ballistica_api_key`` from ``config/localconfig.json`` and passes it as a Bearer token in the Authorization header. All arguments are forwarded to curl. The ``-s`` (silent) flag is added automatically. Examples:: bacurl https://dev.ballistica.net/api/v1/admin/stats/catalog bacurl -X POST -H 'Content-Type: application/json' \\ -d '{"dry_run":true}' \\ https://dev.ballistica.net/api/v1/admin/stats/flush """ import json import subprocess from efro.error import CleanError from efrotools import pcommand args = pcommand.get_args() if not args: raise CleanError('Usage: bacurl [curl-args...] <url>') try: with open('config/localconfig.json', encoding='utf-8') as f: cfg = json.load(f) except FileNotFoundError as exc: raise CleanError('config/localconfig.json not found.') from exc api_key = cfg.get('ballistica_api_key') if not api_key: raise CleanError('No ballistica_api_key in config/localconfig.json.') cmd = [ 'curl', '-s', '-H', f'Authorization: Bearer {api_key}', *args, ] result = subprocess.run(cmd, check=False) raise SystemExit(result.returncode)
[docs] def require_ballistica_api_key() -> None: """Verify a Ballistica API key is available; error if not.""" import os from pathlib import Path from efro.error import CleanError from efrotools.project import getlocalconfig if os.environ.get('BALLISTICA_API_KEY'): return val = getlocalconfig(Path('.')).get('ballistica_api_key') if val: return raise CleanError( 'No Ballistica API key found.\n' 'Set the BALLISTICA_API_KEY env var or add' ' ballistica_api_key to config/localconfig.json.' )
# Docs-generation hack; import some stuff that we likely only forward-declared # in our actual source code so that docs tools can find it. from typing import (Coroutine, Any, Literal, Callable, Generator, Awaitable, Sequence, Self) import asyncio from concurrent.futures import Future from pathlib import Path from enum import Enum