code signal progressive file manager database with unit tests

codesignal progressive file manager with unit tests

Your task is to implement a simplified version of an in-memory database. Plan your design according to the level specifications below:

Level 1: In-memory database should support basic operations to manipulate records, fields, and values within fields.
Level 2: In-memory database should support displaying a specific record's fields based on a filter.
Level 3: In-memory database should support TTL (Time-To-Live) configurations on database records.
Level 4: In-memory database should support backup and restore functionality.

I only got to level 3, got 750/750.
0/?? for Level 4. 


from in_memory_db import InMemoryDB
from collections import defaultdict

class InMemoryDBImpl(InMemoryDB):

    def __init__(self):
        # TODO: implement
        self.d = defaultdict(defaultdict)
        self.ttl = defaultdict(defaultdict)

    # TODO: implement interface methods here
    
    def set(self, key: str, field: str, value: str) -> None:
        
        if key not in self.d:
            self.d[key] = dict()        
            
        self.d[key][field] = value
        
    def set_at(self, key: str, field: str, value: str, timestamp: int) -> None:
        # if key not in self.d:
        #     self.d[key] = dict()        
            
        # self.d[key][field] = value  
        
        self.set_at_with_ttl(key,field,value,timestamp, float('inf'))
        
        # TTL
        # if key not in self.ttl:
        #     self.ttl[key] = dict()        
        # self.ttl[key][field] = timestamp     

    def set_at_with_ttl(self, key: str, field: str, value: str, timestamp: int, ttl: int) -> None:
        if key not in self.d:
            self.d[key] = dict()        
            
        self.d[key][field] = value  
        
        # TTL
        if key not in self.ttl:
            self.ttl[key] = dict()        
        self.ttl[key][field] = timestamp+ttl      

    def get(self, key: str, field: str) -> str | None:
        
        if key in self.d:
            if field in self.d[key]:
                return self.d[key][field]
        return None
        
    def get_at(self, key: str, field: str, timestamp: int) -> str | None:
        
        if key in self.ttl and field in self.ttl[key]:     
            if self.ttl[key][field] <= timestamp:
                return None
                
        if key in self.d:
            if field in self.d[key]:
                return self.d[key][field]
                
        return None
    
    def delete(self, key: str, field: str) -> bool:
        
        if key in self.d:
            if field in self.d[key]:
                del self.d[key][field]
                return True
        
        return False
    
    def delete_at(self, key: str, field: str, timestamp: int) -> bool:
        if key in self.d:
            if field in self.d[key]:
                if key in self.ttl and field in self.ttl[key]:
                    if self.ttl[key][field] <= timestamp:
                        del self.d[key][field]
                        del self.ttl[key][field]
                        return False # expired ; ghost delete
                    else:
                        del self.d[key][field]
                        del self.ttl[key][field]
                        return True # non expired     
                
                else: #not in TTL
                
                    del self.d[key][field]
                    return True
        
        return False
        
    def scan(self, key: str) -> list[str]:
        ls = []
        #if key in self.d:
        for k, v in self.d[key].items():
            ls.append(k+'('+v+')')
            
        ls.sort(reverse=False)
        return ls
    
    def scan_at(self, key: str, timestamp: int) -> list[str]:
        ls = []
        #if key in self.d:
        for k, v in self.d[key].items():
            if key in self.ttl and k in self.ttl[key]:
                if self.ttl[key][k] > timestamp:
                    ls.append(k+'('+v+')')
            else:
                ls.append(k+'('+v+')')
            
        ls.sort(reverse=False)
        return ls   
        
        
    def scan_by_prefix(self,  key: str, prefix: str) -> list[str]:
        ls = []
        #if key in self.d:
        for k, v in self.d[key].items():
            if prefix == k[:len(prefix)]:
                ls.append(k+'('+v+')')
        
        ls.sort(reverse=False)
        return ls
    
    def scan_by_prefix_at(self, key: str, prefix: str, timestamp: int) -> list[str]:
        ls = []
        #if key in self.d:
        for k, v in self.d[key].items():
            if prefix == k[:len(prefix)]:
                if key in self.ttl and k in self.ttl[key]:
                    if self.ttl[key][k] > timestamp:
                        ls.append(k+'('+v+')')
                else: 
                    ls.append(k+'('+v+')')
        
        ls.sort(reverse=False)
        return ls