Backend Live Coding

# ======================================================================
# Q: Find the top 5 customers by revenue over the last 30 days (paid orders
# only).
#
# Note: filters in the same filter() call apply to the same order. One query
# with a JOIN and GROUP BY. I use unit_price from OrderItem (the price at the
# time of purchase), not the product's current price.
# ======================================================================

since = timezone.now() - timedelta(days=30)
revenue = ExpressionWrapper(F('orders__items__quantity') * F('orders__items__unit_price'),
                            output_field=DecimalField(max_digits=14, decimal_places=2))
top = (Customer.objects
       .filter(orders__status='paid', orders__created_at__gte=since)
       .annotate(revenue=Sum(revenue))
       .order_by('-revenue')[:5])

# ======================================================================
# Q: Which models are used for the Django ORM live-coding exercises?
#
# This is the most common part of a Django technical interview. The interviewer
# gives you models and asks for queries. Use these models for all the exercises:
# ======================================================================

class Customer(models.Model):
    name = models.CharField(max_length=200)
    company = models.ForeignKey(Company, on_delete=models.PROTECT)

class Product(models.Model):
    sku = models.CharField(max_length=50, unique=True)
    name = models.CharField(max_length=200)
    price = models.DecimalField(max_digits=10, decimal_places=2)
    stock = models.IntegerField(default=0)

class Order(models.Model):
    customer = models.ForeignKey(Customer, on_delete=models.PROTECT, related_name='orders')
    status = models.CharField(max_length=20)   # pending, paid, shipped, cancelled
    created_at = models.DateTimeField(auto_now_add=True)

class OrderItem(models.Model):
    order = models.ForeignKey(Order, on_delete=models.CASCADE, related_name='items')
    product = models.ForeignKey(Product, on_delete=models.PROTECT)
    quantity = models.PositiveIntegerField()
    unit_price = models.DecimalField(max_digits=10, decimal_places=2)

# ======================================================================
# Q: Threads vs processes vs asyncio: when to use each?
#
# Threads: I/O-bound work with blocking libraries (e.g. 20 HTTP calls with
# requests via ThreadPoolExecutor). Processes: CPU-bound work (calculations,
# images), to bypass the GIL. asyncio: thousands of concurrent I/O connections
# (WebSockets, many external APIs) with async libraries (httpx, asyncpg).
# ======================================================================

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=10) as pool:
    prices = list(pool.map(fetch_supplier_price, supplier_urls))

# ======================================================================
# Q: What are *args, **kwargs and keyword-only arguments?
#
# Keyword-only arguments are useful in dangerous functions (money), to avoid
# mistakes with argument order.
# ======================================================================

def create_order(customer, *items, notify=True, **meta):
    ...
create_order(c, item1, item2, notify=False, source='b2b')
# items = (item1, item2), meta = {'source': 'b2b'}

def transfer(src, dst, *, amount, reference):   # after the * only by name
    ...
transfer(a, b, amount=Decimal('10.00'), reference='TX-1')

# ======================================================================
# Q: What is the difference between an iterator, an iterable and a generator?
#
# Iterable: has __iter__ and returns an iterator (list, dict, QuerySet).
# Iterator: has __next__ and is exhausted after one pass. Generator: the easiest
# way to build an iterator, using yield. Generator expression: (x * 2 for x in
# data), which does not create a list in memory.
# ======================================================================

def read_in_chunks(qs, size=2000):
    batch = []
    for obj in qs.iterator(chunk_size=size):
        batch.append(obj)
        if len(batch) == size:
            yield batch
            batch = []
    if batch:
        yield batch