|
|
import re |
|
|
from enum import IntEnum |
|
|
from uuid import UUID |
|
|
|
|
|
import httpx |
|
|
|
|
|
BASE_URL = "http://127.0.0.1:8000" |
|
|
TABLE_PARTICIPANTS = "participants" |
|
|
TABLE_TOPICS = "topics" |
|
|
|
|
|
NEWLINE = "\n" |
|
|
SEPARATOR_60 = "=" * 60 |
|
|
SEPARATOR_90 = "-" * 90 |
|
|
DASH_LINE_60 = "-" * 60 |
|
|
MAIN_MENU_TITLE = " FORUM CLIENT - MAIN MENU" |
|
|
PROMPT_SELECT_OPTION = "Select option: " |
|
|
MSG_GOODBYE = "Goodbye!" |
|
|
LABEL_ALL_PARTICIPANTS = "ALL PARTICIPANTS:" |
|
|
LABEL_ALL_TOPICS = "ALL TOPICS:" |
|
|
LABEL_ALL_MESSAGES = "ALL MESSAGES:" |
|
|
LABEL_CREATE_PARTICIPANT = "CREATE NEW PARTICIPANT:" |
|
|
LABEL_CREATE_TOPIC = "CREATE NEW TOPIC:" |
|
|
LABEL_PUBLISH_MESSAGE = "PUBLISH MESSAGE TO TOPIC:" |
|
|
PROMPT_PARTICIPANT_ID = "Enter participant ID: " |
|
|
PROMPT_TOPIC_ID = "Enter topic ID: " |
|
|
PROMPT_TITLE = " Title: " |
|
|
PROMPT_DESCRIPTION = " Description: " |
|
|
PROMPT_FIRST_NAME = " First name: " |
|
|
PROMPT_LAST_NAME = " Last name: " |
|
|
PROMPT_NICKNAME = " Nickname: " |
|
|
PROMPT_ACTIVITY = " Activity rating: " |
|
|
PROMPT_TOPIC_ID_INLINE = " Topic ID: " |
|
|
PROMPT_PARTICIPANT_ID_INLINE = " Participant ID: " |
|
|
PROMPT_MESSAGE_CONTENT = " Message content: " |
|
|
MSG_NO_PARTICIPANTS = "No participants found." |
|
|
MSG_NO_TOPICS = "No topics found." |
|
|
MSG_INVALID_OPTION = "Invalid option. Please try again." |
|
|
MSG_INVALID_UUID = "Invalid UUID format." |
|
|
MSG_GOODBYE_INTERRUPT = "\n\nInterrupted. Goodbye!" |
|
|
MSG_TOPIC_ID_REQUIRED = "Topic ID is required." |
|
|
MSG_PARTICIPANT_ID_REQUIRED = "Participant ID is required." |
|
|
MSG_TOPIC_NOT_FOUND = "Topic not found." |
|
|
MSG_PARTICIPANT_NOT_FOUND = "Participant not found." |
|
|
MSG_NO_MESSAGES = " No messages yet." |
|
|
MSG_PARTICIPANT_CREATED = "\nParticipant created successfully!" |
|
|
MSG_TOPIC_CREATED = "\nTopic created successfully!" |
|
|
MSG_MESSAGE_PUBLISHED = "\nMessage published successfully!" |
|
|
MSG_ERROR_TEMPLATE = "Error: {status} - {text}" |
|
|
PARTICIPANT_NAME_RULE = "First name must contain only letters and '-'." |
|
|
PARTICIPANT_LAST_NAME_RULE = "Last name must contain only letters and '-'." |
|
|
PARTICIPANT_NICKNAME_RULE = "Nickname must contain only letters and '-'." |
|
|
ACTIVITY_RANGE_RULE = "Activity rating must be between 0.1 and 5.0." |
|
|
ACTIVITY_NUMBER_RULE = "Activity rating must be a number between 0.1 and 5.0." |
|
|
PARTICIPANT_TABLE_HEADER = f"{'ID':<40} {'Name':<25} {'Nickname':<15} {'Rating':<10}" |
|
|
PARTICIPANT_ROW_TEMPLATE = ( |
|
|
"{id:<40} {full_name:<25} {nickname:<15} {rating:<10}" |
|
|
) |
|
|
TOPIC_TITLE_TEMPLATE = "\nTopic: {title}" |
|
|
TOPIC_ID_TEMPLATE = " ID: {id}" |
|
|
TOPIC_DESCRIPTION_TEMPLATE = " Description: {description}" |
|
|
TOPIC_CREATED_TEMPLATE = " Created: {created_at}" |
|
|
TOPIC_MESSAGES_TEMPLATE = " Messages ({count}):" |
|
|
TOPIC_PARTICIPANTS_TEMPLATE = " Participants: {count}" |
|
|
TOPIC_MESSAGE_TEMPLATE = " {order}. [{participant_name}]: {content}" |
|
|
MESSAGE_LIST_TEMPLATE = "{topic_title} | #{order} [{participant_name}]: {content}" |
|
|
PARTICIPANT_SUMMARY_TEMPLATE = "\nParticipant: {first} {last}" |
|
|
PARTICIPANT_ID_TEMPLATE = " ID: {id}" |
|
|
PARTICIPANT_NICKNAME_TEMPLATE = " Nickname: {nickname}" |
|
|
PARTICIPANT_RATING_TEMPLATE = " Rating: {rating}" |
|
|
PARTICIPANT_REGISTERED_TEMPLATE = " Registered: {registered}" |
|
|
TOPIC_CREATED_ID_TEMPLATE = " ID: {id}" |
|
|
PARTICIPANT_CREATED_ID_TEMPLATE = " ID: {id}" |
|
|
TOPIC_MESSAGES_COUNT_TEMPLATE = "Topic now has {count} message(s)." |
|
|
RESULT_MESSAGE_TEMPLATE = "\n{result}" |
|
|
MENU_HEADER = NEWLINE + SEPARATOR_60 |
|
|
|
|
|
class ForumClient: |
|
|
def __init__(self, base_url: str = BASE_URL): |
|
|
self.base_url = base_url |
|
|
self.client = httpx.Client(timeout=30.0) |
|
|
|
|
|
@staticmethod |
|
|
def _link(table: str, value: str) -> dict: |
|
|
return {"table": table, "value": value} |
|
|
|
|
|
def get_participants(self) -> list: |
|
|
response = self.client.get(f"{self.base_url}/participants/") |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def get_participant(self, participant_id: str) -> dict: |
|
|
response = self.client.get(f"{self.base_url}/participants/{participant_id}") |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def create_participant( |
|
|
self, first_name: str, last_name: str, nickname: str, activity_rating: float |
|
|
) -> dict: |
|
|
payload = { |
|
|
"first_name": first_name, |
|
|
"last_name": last_name, |
|
|
"nickname": nickname, |
|
|
"activity_rating": activity_rating, |
|
|
} |
|
|
response = self.client.post(f"{self.base_url}/participants/", json=payload) |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def get_topics(self) -> list: |
|
|
response = self.client.get(f"{self.base_url}/topics/") |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def get_topic(self, topic_id: str) -> dict: |
|
|
response = self.client.get(f"{self.base_url}/topics/{topic_id}") |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def get_messages(self) -> list: |
|
|
response = self.client.get(f"{self.base_url}/messages/") |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def get_messages_by_topic(self, topic_id: str) -> list: |
|
|
response = self.client.get(f"{self.base_url}/messages/topic/{topic_id}") |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def create_topic( |
|
|
self, title: str, description: str, participants: list = None |
|
|
) -> dict: |
|
|
payload = { |
|
|
"title": title, |
|
|
"description": description, |
|
|
"participants": [self._link(TABLE_PARTICIPANTS, p) for p in (participants or [])], |
|
|
"messages": [], |
|
|
} |
|
|
response = self.client.post(f"{self.base_url}/topics/", json=payload) |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def publish_message( |
|
|
self, topic_id: str, participant_id: str, content: str |
|
|
) -> dict | str: |
|
|
payload = { |
|
|
"participant_id": self._link(TABLE_PARTICIPANTS, participant_id), |
|
|
"content": content, |
|
|
} |
|
|
response = self.client.post( |
|
|
f"{self.base_url}/topics/{topic_id}/messages", json=payload |
|
|
) |
|
|
if response.status_code == 400: |
|
|
error_detail = response.json().get("detail", "Unknown error") |
|
|
return f"ERROR: {error_detail}" |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def close(self): |
|
|
self.client.close() |
|
|
|
|
|
|
|
|
def print_separator(): |
|
|
print(SEPARATOR_60) |
|
|
|
|
|
|
|
|
def print_participants(participants: list): |
|
|
if not participants: |
|
|
print(MSG_NO_PARTICIPANTS) |
|
|
return |
|
|
print(PARTICIPANT_TABLE_HEADER) |
|
|
print(SEPARATOR_90) |
|
|
for p in participants: |
|
|
full_name = f"{p['first_name']} {p['last_name']}" |
|
|
print( |
|
|
PARTICIPANT_ROW_TEMPLATE.format( |
|
|
id=p["id"], |
|
|
full_name=full_name, |
|
|
nickname=p["nickname"], |
|
|
rating=p["activity_rating"], |
|
|
) |
|
|
) |
|
|
|
|
|
|
|
|
def print_topics(topics: list): |
|
|
if not topics: |
|
|
print(MSG_NO_TOPICS) |
|
|
return |
|
|
for topic in topics: |
|
|
print(TOPIC_TITLE_TEMPLATE.format(title=topic["title"])) |
|
|
print(TOPIC_ID_TEMPLATE.format(id=topic["id"])) |
|
|
print(TOPIC_DESCRIPTION_TEMPLATE.format(description=topic["description"])) |
|
|
print(TOPIC_CREATED_TEMPLATE.format(created_at=topic["created_at"])) |
|
|
print(TOPIC_MESSAGES_TEMPLATE.format(count=len(topic["messages"]))) |
|
|
for msg in topic["messages"]: |
|
|
print( |
|
|
TOPIC_MESSAGE_TEMPLATE.format( |
|
|
participant_name=msg["participant_name"], |
|
|
content=msg["content"], |
|
|
order=msg.get("order_in_topic", 0), |
|
|
) |
|
|
) |
|
|
|
|
|
|
|
|
def print_topic_detail(topic: dict): |
|
|
print(TOPIC_TITLE_TEMPLATE.format(title=topic["title"])) |
|
|
print(TOPIC_ID_TEMPLATE.format(id=topic["id"])) |
|
|
print(TOPIC_DESCRIPTION_TEMPLATE.format(description=topic["description"])) |
|
|
print(TOPIC_CREATED_TEMPLATE.format(created_at=topic["created_at"])) |
|
|
print(TOPIC_PARTICIPANTS_TEMPLATE.format(count=len(topic["participants"]))) |
|
|
print(NEWLINE + TOPIC_MESSAGES_TEMPLATE.format(count=len(topic["messages"]))) |
|
|
if not topic["messages"]: |
|
|
print(MSG_NO_MESSAGES) |
|
|
for msg in topic["messages"]: |
|
|
print( |
|
|
TOPIC_MESSAGE_TEMPLATE.format( |
|
|
participant_name=msg["participant_name"], |
|
|
content=msg["content"], |
|
|
order=msg.get("order_in_topic", 0), |
|
|
) |
|
|
) |
|
|
|
|
|
|
|
|
def print_messages(messages: list): |
|
|
if not messages: |
|
|
print(MSG_NO_MESSAGES) |
|
|
return |
|
|
for msg in messages: |
|
|
print( |
|
|
MESSAGE_LIST_TEMPLATE.format( |
|
|
topic_title=msg["topic_title"], |
|
|
order=msg["order_in_topic"], |
|
|
participant_name=msg["participant_name"], |
|
|
content=msg["content"], |
|
|
) |
|
|
) |
|
|
|
|
|
|
|
|
def main_menu(): |
|
|
print(MENU_HEADER) |
|
|
print(MAIN_MENU_TITLE) |
|
|
print(SEPARATOR_60) |
|
|
for option in MENU_OPTIONS: |
|
|
print(option) |
|
|
print(DASH_LINE_60) |
|
|
return input(PROMPT_SELECT_OPTION).strip() |
|
|
|
|
|
|
|
|
class MenuChoice(IntEnum): |
|
|
LIST_PARTICIPANTS = 1, "List all participants" |
|
|
GET_PARTICIPANT = 2, "Get participant by ID" |
|
|
CREATE_PARTICIPANT = 3, "Create new participant" |
|
|
LIST_TOPICS = 4, "List all topics" |
|
|
GET_TOPIC = 5, "Get topic by ID" |
|
|
CREATE_TOPIC = 6, "Create new topic" |
|
|
PUBLISH_MESSAGE = 7, "Publish message to topic" |
|
|
LIST_MESSAGES = 8, "List all messages" |
|
|
EXIT = 0, "Exit" |
|
|
|
|
|
def __new__(cls, value, label=""): |
|
|
obj = int.__new__(cls, value) |
|
|
obj._value_ = value |
|
|
obj.label = label |
|
|
return obj |
|
|
|
|
|
|
|
|
MENU_OPTIONS = tuple(f"{choice.value}. {choice.label}" for choice in MenuChoice) |
|
|
|
|
|
|
|
|
def main(): |
|
|
client = ForumClient() |
|
|
|
|
|
try: |
|
|
while True: |
|
|
raw_choice = main_menu() |
|
|
try: |
|
|
choice = MenuChoice(int(raw_choice)) |
|
|
except (ValueError, TypeError): |
|
|
print(MSG_INVALID_OPTION) |
|
|
continue |
|
|
|
|
|
if choice == MenuChoice.EXIT: |
|
|
print(MSG_GOODBYE) |
|
|
break |
|
|
|
|
|
elif choice == MenuChoice.LIST_PARTICIPANTS: |
|
|
print_separator() |
|
|
print(LABEL_ALL_PARTICIPANTS) |
|
|
try: |
|
|
participants = client.get_participants() |
|
|
print_participants(participants) |
|
|
except httpx.HTTPStatusError as e: |
|
|
print( |
|
|
MSG_ERROR_TEMPLATE.format( |
|
|
status=e.response.status_code, text=e.response.text |
|
|
) |
|
|
) |
|
|
|
|
|
elif choice == MenuChoice.GET_PARTICIPANT: |
|
|
print_separator() |
|
|
participant_id = input(PROMPT_PARTICIPANT_ID).strip() |
|
|
try: |
|
|
UUID(participant_id) |
|
|
participant = client.get_participant(participant_id) |
|
|
print( |
|
|
PARTICIPANT_SUMMARY_TEMPLATE.format( |
|
|
first=participant["first_name"], |
|
|
last=participant["last_name"], |
|
|
) |
|
|
) |
|
|
print(PARTICIPANT_ID_TEMPLATE.format(id=participant["id"])) |
|
|
print( |
|
|
PARTICIPANT_NICKNAME_TEMPLATE.format( |
|
|
nickname=participant["nickname"] |
|
|
) |
|
|
) |
|
|
print( |
|
|
PARTICIPANT_RATING_TEMPLATE.format( |
|
|
rating=participant["activity_rating"] |
|
|
) |
|
|
) |
|
|
print( |
|
|
PARTICIPANT_REGISTERED_TEMPLATE.format( |
|
|
registered=participant["registered_at"] |
|
|
) |
|
|
) |
|
|
except ValueError: |
|
|
print(MSG_INVALID_UUID) |
|
|
except httpx.HTTPStatusError as e: |
|
|
print( |
|
|
MSG_ERROR_TEMPLATE.format( |
|
|
status=e.response.status_code, text=e.response.text |
|
|
) |
|
|
) |
|
|
|
|
|
elif choice == MenuChoice.CREATE_PARTICIPANT: |
|
|
print_separator() |
|
|
print(LABEL_CREATE_PARTICIPANT) |
|
|
while True: |
|
|
first_name = input(PROMPT_FIRST_NAME).strip() |
|
|
if not re.fullmatch(r"[A-Za-zА-Яа-яЁё-]+", first_name): |
|
|
print(PARTICIPANT_NAME_RULE) |
|
|
continue |
|
|
break |
|
|
|
|
|
while True: |
|
|
last_name = input(PROMPT_LAST_NAME).strip() |
|
|
if not re.fullmatch(r"[A-Za-zА-Яа-яЁё-]+", last_name): |
|
|
print(PARTICIPANT_LAST_NAME_RULE) |
|
|
continue |
|
|
break |
|
|
|
|
|
while True: |
|
|
nickname = input(PROMPT_NICKNAME).strip() |
|
|
if not re.fullmatch(r"[A-Za-zА-Яа-яЁё-]+", nickname): |
|
|
print(PARTICIPANT_NICKNAME_RULE) |
|
|
continue |
|
|
break |
|
|
|
|
|
while True: |
|
|
activity_rating_raw = input(PROMPT_ACTIVITY).strip() |
|
|
try: |
|
|
activity_rating = float(activity_rating_raw) |
|
|
if activity_rating < 0.1 or activity_rating > 5.0: |
|
|
print(ACTIVITY_RANGE_RULE) |
|
|
continue |
|
|
except ValueError: |
|
|
print(ACTIVITY_NUMBER_RULE) |
|
|
continue |
|
|
break |
|
|
|
|
|
try: |
|
|
participant = client.create_participant( |
|
|
first_name, last_name, nickname, activity_rating |
|
|
) |
|
|
print(MSG_PARTICIPANT_CREATED) |
|
|
print( |
|
|
PARTICIPANT_CREATED_ID_TEMPLATE.format(id=participant["id"]) |
|
|
) |
|
|
except httpx.HTTPStatusError as e: |
|
|
print( |
|
|
MSG_ERROR_TEMPLATE.format( |
|
|
status=e.response.status_code, text=e.response.text |
|
|
) |
|
|
) |
|
|
|
|
|
elif choice == MenuChoice.LIST_TOPICS: |
|
|
print_separator() |
|
|
print(LABEL_ALL_TOPICS) |
|
|
try: |
|
|
topics = client.get_topics() |
|
|
print_topics(topics) |
|
|
except httpx.HTTPStatusError as e: |
|
|
print( |
|
|
MSG_ERROR_TEMPLATE.format( |
|
|
status=e.response.status_code, text=e.response.text |
|
|
) |
|
|
) |
|
|
|
|
|
elif choice == MenuChoice.LIST_MESSAGES: |
|
|
print_separator() |
|
|
print(LABEL_ALL_MESSAGES) |
|
|
try: |
|
|
messages = client.get_messages() |
|
|
print_messages(messages) |
|
|
except httpx.HTTPStatusError as e: |
|
|
print( |
|
|
MSG_ERROR_TEMPLATE.format( |
|
|
status=e.response.status_code, text=e.response.text |
|
|
) |
|
|
) |
|
|
|
|
|
elif choice == MenuChoice.GET_TOPIC: |
|
|
print_separator() |
|
|
topic_id = input(PROMPT_TOPIC_ID).strip() |
|
|
try: |
|
|
UUID(topic_id) |
|
|
topic = client.get_topic(topic_id) |
|
|
print_topic_detail(topic) |
|
|
except ValueError: |
|
|
print(MSG_INVALID_UUID) |
|
|
except httpx.HTTPStatusError as e: |
|
|
print( |
|
|
MSG_ERROR_TEMPLATE.format( |
|
|
status=e.response.status_code, text=e.response.text |
|
|
) |
|
|
) |
|
|
|
|
|
elif choice == MenuChoice.CREATE_TOPIC: |
|
|
print_separator() |
|
|
print(LABEL_CREATE_TOPIC) |
|
|
title = input(PROMPT_TITLE).strip() |
|
|
description = input(PROMPT_DESCRIPTION).strip() |
|
|
try: |
|
|
topic = client.create_topic(title, description) |
|
|
print(MSG_TOPIC_CREATED) |
|
|
print(TOPIC_CREATED_ID_TEMPLATE.format(id=topic["id"])) |
|
|
except httpx.HTTPStatusError as e: |
|
|
print( |
|
|
MSG_ERROR_TEMPLATE.format( |
|
|
status=e.response.status_code, text=e.response.text |
|
|
) |
|
|
) |
|
|
|
|
|
elif choice == MenuChoice.PUBLISH_MESSAGE: |
|
|
print_separator() |
|
|
print(LABEL_PUBLISH_MESSAGE) |
|
|
while True: |
|
|
topic_id = input(PROMPT_TOPIC_ID_INLINE).strip() |
|
|
if not topic_id: |
|
|
print(MSG_TOPIC_ID_REQUIRED) |
|
|
continue |
|
|
try: |
|
|
client.get_topic(topic_id) |
|
|
except httpx.HTTPStatusError as e: |
|
|
if e.response.status_code in (404, 422): |
|
|
print(MSG_TOPIC_NOT_FOUND) |
|
|
else: |
|
|
print( |
|
|
MSG_ERROR_TEMPLATE.format( |
|
|
status=e.response.status_code, text=e.response.text |
|
|
) |
|
|
) |
|
|
continue |
|
|
break |
|
|
|
|
|
while True: |
|
|
participant_id = input(PROMPT_PARTICIPANT_ID_INLINE).strip() |
|
|
if not participant_id: |
|
|
print(MSG_PARTICIPANT_ID_REQUIRED) |
|
|
continue |
|
|
try: |
|
|
client.get_participant(participant_id) |
|
|
except httpx.HTTPStatusError as e: |
|
|
if e.response.status_code in (404, 422): |
|
|
print(MSG_PARTICIPANT_NOT_FOUND) |
|
|
else: |
|
|
print( |
|
|
MSG_ERROR_TEMPLATE.format( |
|
|
status=e.response.status_code, text=e.response.text |
|
|
) |
|
|
) |
|
|
continue |
|
|
break |
|
|
|
|
|
content = input(PROMPT_MESSAGE_CONTENT).strip() |
|
|
|
|
|
result = client.publish_message(topic_id, participant_id, content) |
|
|
if isinstance(result, str): |
|
|
print(RESULT_MESSAGE_TEMPLATE.format(result=result)) |
|
|
else: |
|
|
print(MSG_MESSAGE_PUBLISHED) |
|
|
print( |
|
|
TOPIC_MESSAGES_COUNT_TEMPLATE.format( |
|
|
count=len(result["messages"]) |
|
|
) |
|
|
) |
|
|
|
|
|
else: |
|
|
print(MSG_INVALID_OPTION) |
|
|
|
|
|
except KeyboardInterrupt: |
|
|
print(MSG_GOODBYE_INTERRUPT) |
|
|
finally: |
|
|
client.close() |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
main() |
|
|
|
