Ticket King Solution¶

Solution¶

import redis
import datetime


def get_plan_limit(plan):
    """
    Identify the daily API limit for this plan

    :param plan: The plan name
    :return: The daily limit. 0 if plan is not recognized
    """

    plan_limits = {
        "peasant": 10,
        "noble": 20,
        "royal": 30,
    }

    return plan_limits.get(plan, 0)


def get_user_plan(user_id: int):
    """
    Fetch this user's plan from the SQL database.
    (This function is slow and expensive..)
    """

    users = {
        22912157: "peasant",
        64792475: "noble",
        56488868: "royal",
        92899704: "noble",
        73532154: "peasant",
        68472103: "peasant",
    }

    plan = users.get(user_id)
    return plan


def allow_request(user_id: int):
    """
    Check if this user is allowed to make a request right now.
    :return: True or False
    """

    # Connect to redis
    r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

    # Define redis keys
    current_date = datetime.datetime.now(tz=datetime.timezone.utc).strftime("%Y-%m-%d")
    key_limit = f"limit:{user_id}:{current_date}"
    key_usage = f"usage:{user_id}:{current_date}"

    # Determine the user's limit
    # If it's not already stored on redis, fetch and store it.
    if (limit := r.get(name=key_limit)) is None:
        plan = get_user_plan(user_id)
        limit = get_plan_limit(plan)
        r.set(name=key_limit, value=limit)
    else:
        limit = int(limit)

    # Determine the user's usage (before this api call)
    if (usage := r.get(name=key_usage)) is None:
        usage = 0
    else:
        usage = int(usage)

    # If usage is below the limit, update its value on redis
    if usage < limit:
        r.set(name=key_usage, value=usage + 1)

    # Return True if usage is below limit, False otherwise
    return usage < limit

Tests¶

peasant plan usernoble planroyal planno plan user

# user 73532154: peasant plan. Should be allowed 10 requests per day.
for i in range(11):
    print(i, allow_request(user_id=73532154))

# 0 True
# 1 True
# 2 True
# ...
# 8 True
# 9 True
# 10 False

# user 92899704: noble plan. Should be allowed 20 requests per day.
for i in range(21):
    print(i, allow_request(user_id=92899704))

# 0 True
# 1 True
# 2 True
# ...
# 18 True
# 19 True
# 20 False

# user 56488868: Non-existent user. Should be allowed 30 requests per day.
for i in range(31):
    print(i, allow_request(user_id=56488868))

# 0 True
# 1 True
# 2 True
# ...
# 28 True
# 29 True
# 30 False

# user 123: Non-existent user. Should be allowed 0 requests per day.
for i in range(1):
    print(i, allow_request(user_id=123))

# 0 False

How to clear the database

You might want to flush your database before running the tests above. You can do so with flushdb().

r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
r.flushdb()

Explanation¶

The first step is to connect to Redis.

import redis

# Connect to redis
r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True) # (1)!
type(r)  # redis.client.Redis

Here we connect to a local Redis instance using redis-py. The argument decode_responses=True tells the redis client to decode responses from bytestrings to plain strings. Observe the difference below

r1 = redis.Redis(host="localhost", port=6379, db=0, decode_responses=False)
r2 = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
r1.set("foo", 10)

r1.get("foo")  # b'10'  <- bytestring
r2.get("foo")  # '10'   <- string

Define redis keys. We'll keep track of two key-values per user:
- key_limit:
  Stores the limit for a particular (user, date).
  Example: (key = limit:123456:2022-05-15, value = 20)
- key_usage:
  Stores the number of API calls made for a particular (user, date).
  Example: (key = usage:123456:2022-05-15, value = 12)
```
import datetime

# Define redis keys
current_date = datetime.datetime.now(tz=datetime.timezone.utc).strftime("%Y-%m-%d")
key_limit = f"limit:{user_id}:{current_date}"
key_usage = f"usage:{user_id}:{current_date}"
```

Determine the user's limit (for today).

# Determine the user's limit
# If it's not already stored on redis, fetch and store it.
if (limit := r.get(name=key_limit)) is None: # (1)!
    plan = get_user_plan(user_id)
    limit = get_plan_limit(plan)
    r.set(name=key_limit, value=limit)
else:
    limit = int(limit) # (2)!

:= is the walrus operator. It lets us reduce code like this

limit = r.get(name=key_limit)
if limit is None:

into a single line.

if (limit := r.get(name=key_limit)) is None:

Redis stores everything as a string. As such, it returns everything as a string. So, we need to convert limit from a string to an int.

Determine how many API calls this user made prior to the current one.

# Determine the user's usage (before this api call)
if (usage := r.get(name=key_usage)) is None:
    usage = 0
else:
    usage = int(usage)

If the user's usage is below the user's limit, increment usage by one and save the new value on Redis.
```
# If usage is below the limit, update its value on redis
if usage < limit:
    r.set(name=key_usage, value=usage + 1)
```
Why don't we always increment usage by 1?

You certainly could. But it's important to make the distinction between a successful API call and an unsuccessful API call. If the user's usage equals the limit, this call is going to get denied, making it an unsuccessful call. We've made the design decision to store successful API calls only.

One benefit of this is, if the user upgrades their plan midday, their unsuccessful API calls from before the upgrade will not be counted against their new quota.

Return True or False.

# Return True if usage is below limit, False otherwise
return usage < limit