Fixes to update incorrect debug logs. Adding basic user client module.

This commit is contained in:
Michael Woods
2025-02-16 14:54:50 -05:00
parent e2f4932dbc
commit 54709796be
3 changed files with 30 additions and 5 deletions

View File

@@ -55,7 +55,7 @@ def get_bulletin_by_id(client: Client, bbs_callsign: str, bid: int) -> BulletinW
req.method = Request.Method.GET req.method = Request.Method.GET
response = client.send_receive_callsign(req, bbs_callsign) response = client.send_receive_callsign(req, bbs_callsign)
if response.status_code != 200: if response.status_code != 200:
raise RuntimeError(f"Sending job failed: {response.status_code}: {response.payload}") raise RuntimeError(f"GET bulletin {bid} failed: {response.status_code}: {response.payload}")
return BulletinWrapper(response.payload) return BulletinWrapper(response.payload)
def get_bulletins_recent(client: Client, bbs_callsign: str, limit: int = None) -> list[BulletinWrapper]: def get_bulletins_recent(client: Client, bbs_callsign: str, limit: int = None) -> list[BulletinWrapper]:
@@ -66,7 +66,7 @@ def get_bulletins_recent(client: Client, bbs_callsign: str, limit: int = None) -
req.set_var('limit', limit) req.set_var('limit', limit)
response = client.send_receive_callsign(req, bbs_callsign) response = client.send_receive_callsign(req, bbs_callsign)
if response.status_code != 200: if response.status_code != 200:
raise RuntimeError(f"Sending job failed: {response.status_code}: {response.payload}") raise RuntimeError(f"Listing bulletins failed: {response.status_code}: {response.payload}")
out_list = [] out_list = []
for b in response.payload: for b in response.payload:
out_list.append(BulletinWrapper(b)) out_list.append(BulletinWrapper(b))

View File

@@ -104,7 +104,7 @@ def get_object_by_uuid(client: Client, bbs_callsign: str, uuid: Union[str, bytes
req.method = Request.Method.GET req.method = Request.Method.GET
response = client.send_receive_callsign(req, bbs_callsign) response = client.send_receive_callsign(req, bbs_callsign)
if response.status_code != 200: if response.status_code != 200:
raise RuntimeError(f"Sending job failed: {response.status_code}: {response.payload}") raise RuntimeError(f"GET object {uid} failed: {response.status_code}: {response.payload}")
return ObjectWrapper(response.payload) return ObjectWrapper(response.payload)
def get_user_objects(client: Client, bbs_callsign: str, limit: int = 10, include_data: bool = True, search: str = None, def get_user_objects(client: Client, bbs_callsign: str, limit: int = 10, include_data: bool = True, search: str = None,
@@ -128,7 +128,7 @@ def get_user_objects(client: Client, bbs_callsign: str, limit: int = 10, include
req.method = Request.Method.GET req.method = Request.Method.GET
response = client.send_receive_callsign(req, bbs_callsign) response = client.send_receive_callsign(req, bbs_callsign)
if response.status_code != 200: if response.status_code != 200:
raise RuntimeError(f"Sending job failed: {response.status_code}: {response.payload}") raise RuntimeError(f"Listing objects failed: {response.status_code}: {response.payload}")
out_list = [] out_list = []
for o in response.payload: for o in response.payload:
out_list.append(ObjectWrapper(o)) out_list.append(ObjectWrapper(o))
@@ -153,4 +153,8 @@ def delete_object_by_uuid(client: Client, bbs_callsign: str, uuid: Union[str, by
response = client.send_receive_callsign(req, bbs_callsign) response = client.send_receive_callsign(req, bbs_callsign)
if response.status_code != 200: if response.status_code != 200:
raise RuntimeError(f"Deleting object {uid} failed: {response.status_code}: {response.payload}") raise RuntimeError(f"Deleting object {uid} failed: {response.status_code}: {response.payload}")
return True return True
def update_object_by_uuid():
# TODO update object by uuid client
pass

View File

@@ -0,0 +1,21 @@
import datetime
from packetserver.client import Client
from packetserver.common import Request, Response, PacketServerConnection
from typing import Union, Optional
from uuid import UUID, uuid4
import os.path
class UserWrapper:
def __init__(self, data: dict):
self.data = data
def get_user_by_username(client: Client, bbs_callsign: str, username: str) -> UserWrapper:
req = Request.blank()
req.path = "user"
req.set_var('username', username.strip().upper())
req.method = Request.Method.GET
response = client.send_receive_callsign(req, bbs_callsign)
if response.status_code != 200:
raise RuntimeError(f"GET user {username} failed: {response.status_code}: {response.payload}")
return UserWrapper(response.payload)