Need some advice on code structure

General
1

I just finished coding #la7 and refactored my code to use this class.

#from abc import ABC, abstractmethod
#from pydantic import BaseModel
from enum import Enum


class Command(Enum):
    ECHO = 1
    PING = 2
    SET = 3
    GET = 4

class CommandExecutor:
    @staticmethod
    def echo( *args):
        obj, message = args[0], args[1][1]
        return ("+"+message+"\r\n")

    @staticmethod
    def ping( *args):
        return "+PONG\r\n"

    @staticmethod
    def set( *args):

        print(args)
        
        obj, key, value = args[0], args[1][1], args[1][2]
        
        obj.kvstore[key] = value
        
        return "+OK\r\n"

    @staticmethod
    def get( *args):

        obj, key = args[0], args[1][1]
        
        val = obj.kvstore[key]
        resp = f'${len(val)}\r\n{val}\r\n'

        return resp
    

# operation = Operation.ADD
# method = getattr(Calculator, operation.name.lower())
# result = method(2, 3
    

This is my main class:

import socket  # noqa: F401
import threading
import concurrent.futures
import asyncio
from asyncio import events
from .commandhandler import CommandExecutor


TIMEOUT=5000

class RedisDB:

    
    kvstore: dict = {}


    def parse_input(self, data: bytearray) -> str:
        print("data received is ",data)
        data_list = []
        
        data_list = data.decode().split("\r\n") 

        bulk_string_count = data_list[0][1]

        bulk_string_data = [data_list[i] for i in range(1,len(data_list)) if i%2==0] #parses the data and stores only the bulk strings

        print(bulk_string_data)

        command = bulk_string_data[0] # assumes only the first bulk string value will be the command

        method = getattr(CommandExecutor, command.lower())

        resp=method(self, bulk_string_data)

        return resp # send decoded string response
    
  #  def socket_accept(server_socket):
  #      return server_socket.accept()

    async def client_req_resp(self, reader, writer) -> None:
        while True:

            data_input = await asyncio.wait_for(reader.read(2048), timeout=TIMEOUT) # read the data from the client

            if not data_input:
                break
            
            resp=''
            resp = self.parse_input(data_input)
            
            # creates a RESP simple string from the response will probably need a method to send any form of output depending on command in the future
            resp = resp.encode()

            writer.write(resp) # reply to the client with pong (hardcoded for now)
        writer.close



    async def main(self):
        # You can use print statements as follows for debugging, they'll be visible when running tests.
        print("Logs from your program will appear here!")

        server = await asyncio.start_server(self.client_req_resp,"localhost", 6379)
        
        async with server:
            await server.serve_forever()


if __name__ == "__main__":
    dbobj = RedisDB()
    asyncio.run(dbobj.main())

I was wondering if how I implemented it is a normal way to structure code or if there are any bad practices that im displaying other than the excessive commenting

2

It’s looking pretty good. I would say that you will probably want to set up a response class that handles and build the string responses instead of hard coding all of them in as a string.

In the same vein, create a class that validates the redis protocol input.

Class ProtocolParser, Class Validation, Class Request, Class Response

This produces code that follows the control flow of the application and makes it very readable.

2 Likes
3

Thanks for this super helpful advice. I find myself making only minor changes to my code as the stages go on which I think is a good sign but yeah I definitely need to segregate the code some. more and add some unit testing of my own.

1 Like