python-example.py

import requests


class RequestBin:
    def __init__(self, url: str, token: str | None = None):
        self.url = url.rstrip("/")
        self.session = requests.session()
        self.uid = None
        self.token = token

        if not self.token:
            response = self.api_post("/auth/init")
            self.uid = response["uid"]
            self.token = self.session.cookies["token"]
        else:
            self.session.cookies["token"] = self.token
            response = self.api_get("/auth/me")
            self.uid = response["uid"]


    def api_get(self, path: str, **kwargs):
        return self.api_request("GET", path, **kwargs)

    def api_post(self, path: str, **kwargs):
        return self.api_request("POST", path, **kwargs)

    def api_delete(self, path: str, **kwargs):
        return self.api_request("DELETE", path, **kwargs)

    def api_request(self, method, path: str, **kwargs):
        path = path.lstrip("/")
        res = self.session.request(method, f"{self.url}/api/{path}", **kwargs)
        res.raise_for_status()
        return res.json()

    def create_bin(
        self,
        method: str,
        route: str,
        content: str,
        headers: dict = None,
        status_code: int = None,
    ) -> tuple[str, bool]:
        data = {"method": method, "route": route, "content": content}
        if headers is not None:
            data["headers"] = headers

        if status_code is not None:
            data["statusCode"] = status_code

        response = self.api_post(
            "/bins",
            json=data,
        )
        return response["id"], response["inserted"]

    def delete_bin(self, id: str):
        self.api_delete(
            f"/bins/{id}",
            json={
                "id": id,
            },
        )

    def get_bin(self, id: str):
        return self.api_get(
            f"/bins/{id}",
        )

    def list_bins(self):
        return self.api_get("/bins")

    def logs(self, limit=100):
        return self.api_get("/logs/", params={"limit": limit})
    
    def logs_export(self):
        return self.api_get("/logs/export")

    def log(self, id: int):
        return self.api_get(f"/logs/{id}")

    def log_body(self, log_id: str) -> bytes:
        res = self.session.get(f"{self.url}/api/logs/{log_id}/body")
        res.raise_for_status()
        return res.content

    def reset_logs(self):
        self.api_delete("/logs")


URL = "http://localhost:3000/"

rbin = RequestBin(URL)

print(f"==== Test new content ====")

content_id, inserted = rbin.create_bin("GET", "/test", "<h1>hello world!</h1>")
print(f"{inserted = }")

api_content = rbin.get_bin(content_id)

print(f"{api_content = }")

api_list = rbin.list_bins()

print(f"{api_list = }")

r = requests.get(f"{URL}/{rbin.uid}/test")

print(f"content = {r.text}")
print(f"status_code = {r.status_code}")
print(f"headers = {r.headers}")


print(f"==== Test add content ====")

content_id, _ = rbin.create_bin(
    "POST", "/post", "posted!", {"X-additional": "test-header"}, status_code=201
)

api_list = rbin.list_bins()

print(f"{api_list = }")

r = requests.post(f"{URL}/{rbin.uid}/post")

print(f"content = {r.text}")
print(f"status_code = {r.status_code}")
print(f"headers = {r.headers}")

print(f"==== Test edit content ====")

content_id, inserted = rbin.create_bin(
    "GET", "/test", "edited", {"Content-Type": "text/plain"}
)
print(f"inserted = {inserted}")

r = requests.get(f"{URL}/{rbin.uid}/test")

print(f"content = {r.text}")
print(f"headers = {r.headers}")

print(f"==== Test logs ====")

r = requests.post(
    f"{URL}/{rbin.uid}/post", headers={"custom": "value"}, json={"post": "data"}
)

r = requests.post(
    f"{URL}/{rbin.uid}/post",
    headers={"custom": "value", "Content-Type": "application/octet-stream"},
    data="こんにちは".encode("Shift-JIS"),
)

logs = rbin.logs()

print(f"{logs = }")

last_log = rbin.log(logs[-1]["id"])

print(f"{last_log = }")

body1 = rbin.log_body(logs[0]["id"])

print(f"{body1.decode("Shift-JIS") = }")

body1 = rbin.log_body(logs[1]["id"])

print(f"{body1.decode() = }")

exported = rbin.logs_export()

print(f"{exported = }")


print(f"==== Test reset logs ====")
rbin.reset_logs()
logs = rbin.logs()

print(f"{logs = }")

print(f"==== Test continue ====")

rbin2 = RequestBin(URL, token=rbin.token)
print(f"{rbin.uid = }")
print(f"{rbin2.uid = }")