Merge pull request #4 from Fr0stFree/feature/huge-update

Feature/huge update

client/main.py

@@ -0,0 +1,14 @@
+import asyncio
+
+from client.src.client import Client
+from client.src.settings import ClientSettings
+
+
+async def main():
+    settings = ClientSettings()
+    async with Client(settings) as client:
+        await client.handle_input()
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

client/requirements.txt

@@ -0,0 +1,4 @@
+aioconsole==0.7.0
+pydantic==2.7.3
+pydantic_settings==2.3.1
+python-dotenv==1.0.1

client/src/client.py

@@ -0,0 +1,66 @@
+import asyncio
+from types import TracebackType
+from typing import Self, Type
+
+from aioconsole import ainput
+
+from client.src.printer import Printer
+from client.src.settings import ClientSettings
+from shared.schemas import actions
+from shared.schemas.actions import ActionTypes, ActionFrame
+from shared.schemas.notifications import NotificationFrame
+from shared.transport import DataTransport
+
+
+class Client:
+    def __init__(self, settings: ClientSettings) -> None:
+        self._settings = settings
+        self._printer = Printer()
+        self._transport: DataTransport | None = None
+        self._receiver: asyncio.Task | None = None
+
+    async def __aenter__(self) -> Self:
+        self._printer.info(f"Connecting to {self._settings.host}:{self._settings.port}...")
+        try:
+            reader, writer = await asyncio.open_connection(self._settings.host, self._settings.port)
+        except ConnectionRefusedError as error:
+            self._printer.error("Connection refused")
+            raise error
+        self._transport = DataTransport(writer, reader)
+        self._receiver = asyncio.ensure_future(self._receive_data())
+        self._printer.success("Successfully connected")
+        return self
+
+    async def __aexit__(self, exc_type: Type[BaseException], exc: BaseException, tb: TracebackType) -> None:
+        self._printer.info("Disconnecting from the server...")
+        self._receiver.cancel()
+        await self._transport.close()
+        self._printer.success("Successfully disconnected")
+
+    async def _receive_data(self) -> None:
+        while True:
+            data = await self._transport.receive()
+            frame = NotificationFrame.model_validate_json(data)
+            self._printer.message(frame)
+
+    async def send(self, frame: ActionFrame) -> None:
+        await self._transport.transfer(frame.model_dump_json())
+
+    async def handle_input(self) -> None:
+        while True:
+            statement = await ainput()
+            match statement.split():
+                case ["send", *text]:
+                    payload = actions.BroadcastMessagePayload(text=" ".join(text))
+                    frame = actions.BroadcastMessageActionFrame(payload=payload)
+                    await self.send(frame)
+                case ["help"]:
+                    frame = ActionFrame(type=ActionTypes.HELP)
+                    await self.send(frame)
+                case ["exit" | "quit" | "logout"]:
+                    frame = ActionFrame(type=ActionTypes.LOGOUT)
+                    await self.send(frame)
+                    break
+                case _:
+                    text = f"Unknown command: '{statement}'. Use 'help' to see available commands."
+                    self._printer.error(text)

client/src/printer.py

@@ -0,0 +1,38 @@
+from enum import StrEnum
+
+from shared.schemas.notifications import NotificationFrame, NotificationTypes, PrivateMessageNotificationPayload, \
+    BroadcastMessageNotificationPayload
+
+
+class Colors(StrEnum):
+    BLUE = '\033[94m'
+    GREEN = '\033[92m'
+    RED = '\033[91m'
+    YELLOW = '\033[93m'
+    RESET = '\033[0m'
+
+
+class Printer:
+    def message(self, frame: NotificationFrame) -> None:
+        match NotificationTypes(frame.type):
+            case NotificationTypes.PRIVATE_MESSAGE:
+                payload = PrivateMessageNotificationPayload.model_validate(frame.payload)
+                text = f'{payload.sender} >>> {payload.text}'
+                print(self._with_color(text, Colors.BLUE))
+
+            case NotificationTypes.BROADCAST_MESSAGE:
+                payload = BroadcastMessageNotificationPayload.model_validate(frame.payload)
+                text = f'{payload.sender} >>> {payload.text}'
+                print(self._with_color(text, Colors.GREEN))
+
+    def info(self, text: str) -> None:
+        print(self._with_color(text, Colors.YELLOW))
+
+    def error(self, text: str) -> None:
+        print(self._with_color(text, Colors.RED))
+
+    def success(self, text: str) -> None:
+        print(self._with_color(text, Colors.GREEN))
+
+    def _with_color(self, text: str, color: Colors = Colors.GREEN) -> str:
+        return f'{color}{text}{Colors.RESET}'

client/src/settings.py

@@ -0,0 +1,9 @@
+from typing import Final
+
+from pydantic import Field
+from pydantic_settings import BaseSettings
+
+
+class ClientSettings(BaseSettings):
+    host: Final[str] = Field(..., alias='SERVER_HOST')
+    port: Final[int] = Field(..., alias='SERVER_PORT')

server/.env.example

@@ -0,0 +1,2 @@
+SERVER_HOST=127.0.0.1
+SERVER_PORT=8000