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()