Server/client chat over TCP protocol on Python

client.py

import socket
import threading
import utils


class DMClient:
    def __init__(
        self, *, ip_addr: str = "127.0.0.1", port: int = 41995, buffer_size: int = 4192
    ) -> None:
        self.ip_addr = ip_addr
        self.port = port
        self.buffer_size = buffer_size

        client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        client.connect((ip_addr, port))
        self._client = client
        self._is_alive = False

    @utils.ignore(OSError)
    def read(self) -> None:
        while self._is_alive:
            data = self._client.recv(self.buffer_size)
            if data:
                msg = data.decode("utf-8")

                if msg == "QUIT":
                    self._is_alive = False
                    print("Server is closing connection")
                else:
                    print(f"Server says: {msg}")

    def write(self, msg: str) -> None:
        self._client.send(msg.encode("utf-8"))

    def start(self) -> None:
        self._is_alive = True

        thread = threading.Thread(target=self.read)
        thread.start()

        while self._is_alive:
            text = input("")
            self.write(text)

            if text == "QUIT":
                self._is_alive = False
        self._client.close()

main.py

#!/usr/bin/env python3

import argparse, client, server


def get_arguments():
    parser = argparse.ArgumentParser(
        description="Simple direct message TCP server/client application"
    )
    parser.add_argument("--type", choices=("client", "server"), required=True)
    parser.add_argument("--port", type=int, default=41995)
    return parser.parse_args()


if __name__ == "__main__":
    args = get_arguments()

    if args.type == "server":
        server.DMServer(ip_addr="127.0.0.1", port=args.port).listen()
    elif args.type == "client":
        client.DMClient().start()

server.py

import socket
import threading

import utils


class DMServer:
    def __init__(
        self, *, ip_addr: str, port: int, buffer_size: int = 4192, concurrency: int = 1
    ) -> None:
        self.ip_addr = ip_addr
        self.port = port
        self.buffer_size = buffer_size

        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.bind((ip_addr, port))
        server.listen(concurrency)
        self._server = server
        self._is_active = False
        self._conn = None

    @utils.ignore(OSError)
    def read(self) -> None:
        if not self._conn:
            return

        while self._is_active:
            data = self._conn.recv(self.buffer_size)
            if data:
                msg = data.decode("utf-8")

                if msg == "QUIT":
                    self._is_active = False
                    print("Client is closing connection")
                else:
                    print(f"Client says: {msg}")

    def write(self, msg: str) -> None:
        if not self._conn:
            raise Exception("Connection is not open")
        self._conn.send(msg.encode("utf-8"))

    def listen(self) -> None:
        conn, addr = self._server.accept()

        print(f"Established connection to {addr}")
        self._conn = conn

        self._is_active = True
        thread = threading.Thread(target=self.read)
        thread.start()

        while self._is_active:
            text = input("")
            self.write(text)

            if text == "QUIT":
                self._is_active = False
        self._conn.close()
        self._server.close()

utils.py

import functools


def ignore(*exceptions: Exception):
    def decorator(func):
        @functools.wraps(func)
        def with_ignore(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except exceptions as e:
                pass

        return with_ignore

    return decorator