Optionally persist management tokens to disk

This commit is contained in:
grym 2022-12-07 18:18:57 -05:00
parent 884408fac2
commit bf06856e75
2 changed files with 73 additions and 15 deletions

View File

@ -38,16 +38,25 @@ oxo files --help
Usage: oxo files [OPTIONS] FILES...
Upload one or more files.
Upload one or more files. If one is provided by the 0x0 site used on a
successful upload, a management token will be printed to stderr along with
instructions on how to use it to delete or adjust the expiration time on the
uploaded file(s). If `--token-cache-dir` is passed or `OXO_TOKEN_CACHE_DIR`
is set, the management token will also be cached as a json file in that
directory. The cache directory will be created on first use.
╭─ Arguments ──────────────────────────────────────────────────────────────────╮
│ * files FILES... [default: None] [required] │
╰──────────────────────────────────────────────────────────────────────────────╯
╭─ Options ────────────────────────────────────────────────────────────────────╮
│ --expires INTEGER Expiration time, in hours or epoch milliseconds │
│ [default: None] │
│ --base-url TEXT [env var: OXO_BASE_URL] [default: https://0x0.st] │
│ --help Show this message and exit. │
│ --expires INTEGER Expiration time, in hours or epoch │
│ milliseconds │
│ [default: None] │
│ --base-url TEXT [env var: OXO_BASE_URL] │
│ [default: https://0x0.st] │
│ --token-cache-dir TEXT [env var: OXO_TOKEN_CACHE_DIR] │
│ [default: None] │
│ --help Show this message and exit. │
╰──────────────────────────────────────────────────────────────────────────────╯
#+end_example

69
oxo.py
View File

@ -1,4 +1,5 @@
import io
import json
import typing as t
from pathlib import Path
@ -16,12 +17,17 @@ __version__ = version("oxo")
app = typer.Typer()
BASE_URL = typer.Option("https://0x0.st", envvar="OXO_BASE_URL")
TOKEN_CACHE_DIR: t.Optional[str] = typer.Option(None, envvar="OXO_TOKEN_CACHE_DIR")
err_console = Console(stderr=True, color_system=None)
def post_to(
base_url: str, *, data: t.Iterable[t.Dict], expires: t.Optional[int]
base_url: str,
*,
data: t.Iterable[t.Dict],
expires: t.Optional[int],
token_dir: t.Optional[Path],
) -> str:
retval = []
with httpx.Client() as client:
@ -33,12 +39,24 @@ def post_to(
res = client.post(base_url, files=d)
token = res.headers.get("x-token")
if token:
err_console.print(
f"To remove post, curl -Ftoken={token} -Fdelete= {res.text.strip()}"
)
err_console.print(
f"To update expiration date, curl -Ftoken={token} -Fexpires=NUMHOURS {res.text.strip()}"
curl_rm = f"curl -Ftoken={token} -Fdelete= {res.text.strip()}"
curl_expiry = (
f"curl -Ftoken={token} -Fexpires=NUMHOURS {res.text.strip()}"
)
err_console.print(f"To remove post, {curl_rm}")
err_console.print(f"To update expiration date, {curl_rm}")
if token_dir:
print(f"{d.get('file').name=}")
fname = Path(d.get("file").name).name
token_dir.joinpath(f"{fname}.token").write_text(
json.dumps(
dict(
token=token,
curl_rm=curl_rm,
curl_expiry=curl_expiry,
)
)
)
else:
res = client.post(base_url, data=d)
res.raise_for_status()
@ -46,9 +64,18 @@ def post_to(
return " ".join(retval)
def post_files(base_url, files: t.List[Path], expires: t.Optional[int]):
def post_files(
base_url,
files: t.List[Path],
*,
expires: t.Optional[int],
token_dir: t.Optional[Path],
):
return post_to(
base_url, data=({"file": f.open("rb")} for f in files), expires=expires
base_url,
data=({"file": f.open("rb")} for f in files),
expires=expires,
token_dir=token_dir,
)
@ -92,9 +119,31 @@ def files(
None, help="Expiration time, in hours or epoch milliseconds"
),
base_url: str = BASE_URL,
token_cache_dir: str = TOKEN_CACHE_DIR,
):
"""Upload one or more files."""
typer.echo(post_files(base_url, files, expires))
"""Upload one or more files. If one is provided by the 0x0 site used on a
successful upload, a management token will be printed to stderr along with
instructions on how to use it to delete or adjust the expiration time on
the uploaded file(s). If `--token-cache-dir` is passed or
`OXO_TOKEN_CACHE_DIR` is set, the management token will also be cached as a
json file in that directory. The cache directory will be created on first
use.
"""
print(f"{token_cache_dir=}")
if token_cache_dir:
token_dir = Path(token_cache_dir).expanduser().resolve()
if not token_dir.is_dir():
typer.secho(
f"{token_dir=} is not a directory; cannot store tokens here.",
fg=typer.colors.RED,
err=True,
)
raise typer.Exit(-1)
token_dir.mkdir(exist_ok=True, parents=True)
else:
token_dir = None
typer.echo(post_files(base_url, files, expires=expires, token_dir=token_dir))
@app.command()