LLD- In Memory Database Python

Dipesh Kumar Singh — Wed, 26 Jun 2024 07:08:55 +0000

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)

Output:

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.

DEV Community: Dipesh Kumar Singh

LLD- In Memory Database Python