Create an in-memory database of key–field–value records. Each top-level key owns a group of field/value pairs, much like a map whose values are maps. The assignment is cumulative: each stage adds capabilities while retaining everything introduced earlier.
Every operation accepts a positive-integer timestamp. Calls are guaranteed to arrive with strictly increasing timestamps, so no two operations use the same time. Later levels must remain compatible with the earlier API.
Implement the database through the following four levels.
Write an InMemoryDB class that can assign and read field values, remove fields, and perform conditional compare-and-set updates.
class InMemoryDB:
def __init__(self):
"""Create an empty in-memory database."""
pass
def set(self, timestamp: int, key: str, field: str, value: str) -> None:
"""
Save value under the specified key and field.
An existing value for that key-field pair must be replaced.
Args:
timestamp: The current time.
key: The top-level key.
field: The field name within the key.
value: The value to store.
"""
pass
def get(self, timestamp: int, key: str, field: str) -> str:
"""
Look up a value by key and field.
Args:
timestamp: The current time.
key: The top-level key.
field: The field name within the key.
Returns:
The saved value, or "" when either the key or field is absent.
"""
pass
def delete(self, timestamp: int, key: str, field: str) -> bool:
"""
Remove one field from a key.
Args:
timestamp: The current time.
key: The top-level key.
field: The field name to remove.
Returns:
True when the field was present and removed; otherwise False.
"""
pass
def compare_and_set(self, timestamp: int, key: str, field: str,
expected_value: str, new_value: str) -> bool:
"""
Replace a field only when its current value equals expected_value.
Args:
timestamp: The current time.
key: The top-level key.
field: The field name to change.
expected_value: The value required before the update.
new_value: The value to save when the requirement is satisfied.
Returns:
True when the field existed, matched expected_value, and changed;
otherwise False.
"""
pass
db = InMemoryDB()
db.set(1, "account7", "city", "Paris")
db.set(2, "account7", "level", "5")
db.get(3, "account7", "city") # "Paris" (the field is present)
db.get(4, "account7", "phone") # "" (the field is absent)
db.get(5, "account8", "city") # "" (the key is absent)
db.delete(6, "account7", "level") # True (the field is removed)
db.delete(7, "account7", "level") # False (it was already removed)
db.compare_and_set(8, "account7", "city", "Paris", "Rome") # True (the expected value matches)
db.get(9, "account7", "city") # "Rome" (the update succeeded)
db.compare_and_set(10, "account7", "city", "Paris", "Berlin") # False (the current value is "Rome")
db.compare_and_set(11, "account7", "code", "A1", "B2") # False (the field is absent)
Add operations for listing every field belonging to a key and for listing only fields beginning with a specified prefix. In both cases, format entries as field(value) and order them lexicographically by field name.
def scan(self, timestamp: int, key: str) -> list:
"""
List all field-value entries associated with key.
Args:
timestamp: The current time.
key: The top-level key to inspect.
Returns:
Strings formatted as "field(value)", ordered lexicographically by
field name. Return an empty list when key is not present.
"""
pass
def scan_with_prefix(self, timestamp: int, key: str, prefix: str) -> list:
"""
List entries whose field names begin with prefix.
Args:
timestamp: The current time.
key: The top-level key to inspect.
prefix: The field-name prefix to match.
Returns:
Matching entries in "field(value)" format, sorted lexicographically
by field name. Return [] when key is missing or no field matches.
"""
pass
db = InMemoryDB()
db.set(1, "account7", "city", "Paris")
db.set(2, "account7", "level", "5")
db.set(3, "account7", "label", "vip")
db.scan(4, "account7")
# ["city(Paris)", "label(vip)", "level(5)"]
# The entries are ordered by their field names.
db.scan_with_prefix(5, "account7", "la")
# ["label(vip)"]
# Only the field beginning with "la" is included.
db.scan_with_prefix(6, "account7", "l")
# ["label(vip)", "level(5)"]
# Both matching fields are returned in lexical order.
db.scan_with_prefix(7, "account7", "z")
# []
# No field starts with "z".
db.scan(8, "account99")
# []
# The requested key does not exist.
Enhance records with an optional time-to-live. A value written with a TTL expires at timestamp + ttl; at that exact time, and at every later time, it must be treated as expired and omitted from query results.
The simulated clock is the timestamp supplied to each operation, and that clock never decreases.
def set_with_ttl(self, timestamp: int, key: str, field: str,
value: str, ttl: int) -> None:
"""
Save value with an expiration time of timestamp + ttl.
Args:
timestamp: The current time.
key: The top-level key.
field: The field name within the key.
value: The value to store.
ttl: The lifetime in time units. The record is expired whenever
current_time >= timestamp + ttl.
Notes:
- This operation replaces both the value and TTL for the key-field pair.
- The existing set() operation remains valid and creates a record
without an expiration time.
"""
pass
def compare_and_set_with_ttl(self, timestamp: int, key: str, field: str,
expected_value: str, new_value: str,
ttl: int) -> bool:
"""
Conditionally replace a value while assigning a fresh TTL.
Args:
timestamp: The current time.
key: The top-level key.
field: The field name to change.
expected_value: The value required before the update.
new_value: The replacement value.
ttl: The lifetime assigned to the updated record.
Returns:
True only when a non-expired field existed with expected_value and was
updated; otherwise return False.
"""
pass
The existing get, delete, scan, and scan_with_prefix operations must also apply expiration rules. An expired record behaves as though it does not exist.
db = InMemoryDB()
db.set(1, "account7", "name", "Mia") # No TTL, so it does not expire
db.set_with_ttl(2, "account7", "session", "k9", 6) # Expires at time 8
db.get(3, "account7", "session") # "k9" (3 is before 8)
db.get(7, "account7", "session") # "k9" (7 is before 8)
db.get(8, "account7", "session") # "" (8 reaches the expiration time)
db.get(9, "account7", "name") # "Mia" (the record has no TTL)
db.set_with_ttl(10, "account8", "token", "q2", 4) # Expires at time 14
db.scan(11, "account8") # ["token(q2)"] (the record is still valid)
db.scan(14, "account8") # [] (the record has expired)
db.set_with_ttl(15, "account9", "code", "246", 8) # Expires at time 23
db.compare_and_set_with_ttl(16, "account9", "code", "246", "357", 3) # True (new expiry is 19)
db.get(18, "account9", "code") # "357" (18 is before 19)
db.get(19, "account9", "code") # "" (19 reaches the new expiration time)
The assignment continues with Level 4, which builds on the preceding functionality.