DEV Community

Dipesh Kumar Singh
Dipesh Kumar Singh

Posted on

LLD- In Memory Database Python

Problem Statement:
The objective is to design and implement an in-memory SQL-like database, which should support the following set of operations/functionality:

  • It should be possible to create, or delete tables in a database.
  • A table definition comprises columns which have types. They can also have constraints. The supported column types are string and int.
  • Users can give the constraint of string type that can have a maximum length of 20 characters.
  • Users can give the constraint of int type that can have a minimum value of 1024.
  • Support for mandatory fields
  • It should be possible to insert records in a table.
  • It should be possible to print all records in a table.
  • It should be possible to filter and display records whose column values match a given value.

Requirements Analysis

  • Database Operations:

Create, update, or delete tables.

  • Table Definition:

Tables consist of columns with types and optional constraints.
Supported column types: string (max length 20) and int (min value 1024).
Support for mandatory fields.

  • Data Operations:

Insert records into a table.
Print all records in a table.
Filter and display records based on column values matching a given value.

Class Diagram:

Database: Manages multiple tables.
Table: Contains columns and records.
Column: Abstract base class for StringColumn and IntColumn.
StringColumn: Represents a column of type string with optional constraints.
IntColumn: Represents a column of type int with optional constraints.
Record: Represents a record (row) in a table.

Implementation:



from abc import ABC,abstractmethod

class ColumnType:
    STRING="string"
    INT="int"

class Column(ABC):
    def __init__(self,name,column_type,required=False):
        self.name=name
        self.column_type=column_type
        self.required=required

    @abstractmethod
    def validate(self,value):
        pass

class StringColumn(Column):
    MAX_LENGTH=20
    def __init__(self,name,required=False):
        super().__init__(name,ColumnType.STRING,required)

    def validate(self,value):
        if not isinstance(value,str):
            return False
        if len(value)>self.MAX_LENGTH:
            print("Max length exceeeded !!")
            return False
        return True
class IntColumn(Column):
    MAX_LENGTH=1024
    MIN_LENGTH=-1024
    def __init__(self,name,required=False):
        super().__init__(name,ColumnType.INT,required)

    def validate(self,value):
        if not isinstance(value,int):
            return False
        if len(value)<self.MIN_LENGTH or len(value)>self.MAX_LENGTH:
            print("Min or Max length exceeeded !!")
            return False
        return True

class Record:
    def __init__(self,values):
        self.values=values

class Table:
    def __init__(self,name):
        self.name=name 
        self.columns={}
        self.records=[]

    def add_column(self,column):
        self.columns[column.name]=column

    def insert_record(self,values):
        record=Record(values)
        for column_name,column_obj in self.columns.items():
            if column_obj.required and column_name not in values:
                 raise Exception(f"Missing required column '{column_name}' in record.")
        self.records.append(record)

    def print_records(self):
        for record in self.records:
            print(record.values)

    def filter_records(self,column_name,value):
        filtered_records=[]
        for record in self.records:
            if column_name in record.values and record.values[column_name]==value:
                filtered_records.append(record.values)
        return filtered_records

class Database:
    def __init__(self):
        self.tables={}

    def create_table(self,table_name):
        if table_name in self.tables:
            raise Exception(f"table {table_name} already exists !!")
        self.tables[table_name]=Table(table_name)

    def delete_table(self,table_name):
        if table_name not in self.tables:
            raise Exception(f"table {table_name} does not exist !!")
        del self.tables[table_name]

    def get_table(self,table_name):
        if table_name not in self.tables:
            raise Exception(f"table {table_name} does not  exists !!")
        return self.tables[table_name]




if __name__=="__main__":
    db=Database()
    db.create_table("Employees")
    print(db.tables)

    db.get_table("Employees").add_column(StringColumn("Username",required=True))
    db.get_table("Employees").add_column(IntColumn("age",required=True))

    db.get_table("Employees").insert_record({"Username":"Alice","age":27})
    db.get_table("Employees").insert_record({"Username":"bob","age":28})
    db.get_table("Employees").insert_record({"Username":"Carl","age":20})

    print("All records in 'users' table:")
    db.get_table("Employees").print_records()

    filtered_records = db.get_table("Employees").filter_records("Username", "Carl")
    print("\nFiltered records where name is 'Carl':")
    for record in filtered_records:
        print(record)


Enter fullscreen mode Exit fullscreen mode

Output:

output for the inmemory db

SOLID Principles:

Single Responsibility Principle: Each class has a clear and single responsibility (e.g., Database manages tables, Table manages columns and records).

Open/Closed Principle: Code is open for extension (e.g., adding new column types) but closed for modification.

Liskov Substitution Principle: Subclasses (StringColumn, IntColumn) can be substituted for their base class (Column).

Interface Segregation Principle: No explicit interfaces in Python, but classes are designed to expose necessary methods (validate for columns, CRUD operations for tables).

Dependency Inversion Principle: High-level modules (e.g., Database, Table) depend on abstractions (Column), not concretions.

This implementation provides a basic in-memory SQL-like database system in Python, fulfilling the given requirements while adhering to OOP and SOLID principles.

Top comments (0)