Token storage stubbed in.
This commit is contained in:
parent
c666e6f803
commit
d64fe67df0
76
oxo.py
76
oxo.py
|
@ -1,6 +1,7 @@
|
|||
import io
|
||||
import json
|
||||
import typing as t
|
||||
from dataclasses import asdict, dataclass
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
|
@ -22,6 +23,20 @@ TOKEN_CACHE_DIR: t.Optional[str] = typer.Option(None, envvar="OXO_TOKEN_CACHE_DI
|
|||
err_console = Console(stderr=True, color_system=None)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TokenData:
|
||||
token: str
|
||||
oxo_url: str
|
||||
|
||||
@property
|
||||
def curl_rm(self):
|
||||
return f"curl -s -Ftoken={self.token} -Fdelete= {self.oxo_url}"
|
||||
|
||||
@property
|
||||
def curl_expiry(self, expiry_time="NUMHOURS"):
|
||||
return f"curl -s -Ftoken={self.token} -Fexpires={expiry_time} {self.oxo_url}"
|
||||
|
||||
|
||||
def post_to(
|
||||
base_url: str,
|
||||
*,
|
||||
|
@ -34,33 +49,31 @@ def post_to(
|
|||
for d in data:
|
||||
if isinstance(d.get("file"), io.BufferedReader):
|
||||
if expires:
|
||||
res = client.post(base_url, files=d, data={"expires": expires})
|
||||
res = client.post(
|
||||
base_url, files=d, data={"expires": expires, "secret": ""}
|
||||
)
|
||||
else:
|
||||
res = client.post(base_url, files=d)
|
||||
token = res.headers.get("x-token")
|
||||
remote_url = res.text.strip()
|
||||
if token:
|
||||
curl_rm = f"curl -Ftoken={token} -Fdelete= {res.text.strip()}"
|
||||
curl_expiry = (
|
||||
f"curl -Ftoken={token} -Fexpires=NUMHOURS {res.text.strip()}"
|
||||
token_data = TokenData(
|
||||
token=token,
|
||||
oxo_url=remote_url,
|
||||
)
|
||||
err_console.print(f"To remove post, {token_data.curl_rm}")
|
||||
err_console.print(
|
||||
f"To update expiration date, {token_data.curl_expiry}"
|
||||
)
|
||||
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,
|
||||
)
|
||||
)
|
||||
json.dumps(asdict(token_data))
|
||||
)
|
||||
else:
|
||||
res = client.post(base_url, data=d)
|
||||
res.raise_for_status()
|
||||
retval.append(res.text.strip())
|
||||
retval.append(remote_url)
|
||||
return " ".join(retval)
|
||||
|
||||
|
||||
|
@ -134,7 +147,7 @@ def files(
|
|||
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.",
|
||||
f"token_dir={token_dir} is not a directory; cannot store tokens here.",
|
||||
fg=typer.colors.RED,
|
||||
err=True,
|
||||
)
|
||||
|
@ -162,3 +175,34 @@ def shorten(urls: t.List[str] = typer.Argument(..., min=1), base_url=BASE_URL):
|
|||
err=True,
|
||||
)
|
||||
typer.echo(post_shorten(base_url, urls))
|
||||
|
||||
|
||||
@app.command()
|
||||
def manage(
|
||||
token_cache_dir: Path = typer.Argument("", envvar="OXO_TOKEN_CACHE_DIR"),
|
||||
# interactive: bool = typer.Option(
|
||||
# False, help="If True, prompt for each token found."
|
||||
# ),
|
||||
delete: bool = typer.Option(True, help="If True, send delete requests."),
|
||||
base_url=BASE_URL,
|
||||
):
|
||||
token_files = token_cache_dir.glob("*.token")
|
||||
tokens = []
|
||||
for t in token_files:
|
||||
try:
|
||||
tokens.append(TokenData(**json.loads(t.read_text())))
|
||||
except Exception as e:
|
||||
typer.secho(f"Ignoring {t}: {e}", err=True, fg=typer.colors.YELLOW)
|
||||
with httpx.Client() as client:
|
||||
for t in tokens:
|
||||
if delete:
|
||||
try:
|
||||
res = client.post(
|
||||
base_url, files=dict(token=(None, t.token), delete=(None, ""))
|
||||
)
|
||||
res.raise_for_status()
|
||||
typer.echo(res.text.strip())
|
||||
t.unlink()
|
||||
typer.echo(f"Removed stale token {t}")
|
||||
except Exception as e:
|
||||
typer.secho(e, err=True, fg=typer.colors.RED)
|
||||
|
|
Loading…
Reference in New Issue