erikbern

class MyDict(dict):
    def __init__(self):
        self._dict = {}

    def __getitem__(self, k):
        print(f'Looking up {k}')
        return self._dict[k]

    def __setitem__(self, k, v):
        print(f'Assigning {k} to {v}')

erikbern

class MyDict(dict):
    def __init__(self):
        self._dict = {}

    def __getitem__(self, k):
        print(f'Looking up {k}')
        return self._dict[k]

    def __setitem__(self, k, v):
        print(f'Assigning {k} to {v}')

erikbern

import asyncio


class AsyncConstructorMeta(type):
    """Metaclass to support asynchronous constructors in Python.
    
    Basically we're exploiting the fact that __new__ can return anything in Python.
    So we're taking the old __init__ code, removing it from the class, and instead,
    we create a custom __new__ method that returns a coroutine wrapping the original
    constructor.

erikbern

# First, let's create an awaitable object.
# In this case it's a very dumb container of integers.
# Any time a coroutine runs `await Thing(n)` it just resolves to n
# However, we could resolve it to something completely different if we wanted to

class Thing:
    def __init__(self, n):
        self._n = n

    def __await__(self):

erikbern

# Uses American Community Survey data to estimate property taxes
# https://www.census.gov/programs-surveys/acs/
# The data is a f-ing PITA to parse, but here's an attempt

import bs4
import csv
import io
import os
import requests
import sys

erikbern

def get_n_results_dumb(q):
    r = requests.get('http://www.google.com/search',
                     params={'q': q,
                             "tbs": "li:1"})
    r.raise_for_status()
    soup = bs4.BeautifulSoup(r.text)
    s = soup.find('div', {'id': 'resultStats'}).text
    if not s:
        return 0
    m = re.search(r'([0-9,]+)', s)

erikbern

async def intercept_coro(coro, interceptor):
    # This roughly corresponds to https://gist.github.com/erikbern/ad7615d22b700e8dbbafd8e4d2f335e1
    # The underlying idea is that we can execute a coroutine ourselves and use it to intercept
    # any awaitable object. This lets the coroutine await arbitrary awaitable objects, not just
    # asyncio futures. See how this is used in object.load.
    value_to_send = None
    while True:
        try:
            awaitable = coro.send(value_to_send)
            assert inspect.isawaitable(awaitable)

erikbern

# Let's say you have a method `map(f, inputs)` that performs some operation on inputs
# Let's also sayu that this function returns a generator.
# A user might pass a function f that has some side effect, e.g. `def f(x): print(x)`
# This might lead to confusing results if the output is never consumed:
# `map(f, inputs)` -> this does nothing, can be confusing to the user
# `list(map(f, inputs))` -> this executes `f` on all inputs
# To remedy the situation, we can provide a helpful warning to the user if the generator
# is never consumed.

### Helper code:

erikbern

import io
import modal

stub = modal.Stub(image=modal.DebianSlim().pip_install(["prophet"]))

@stub.function
def run():
    import pandas as pd
    from prophet import Prophet
    from matplotlib import pyplot

erikbern

import json
import subprocess
import sys
import tempfile

import modal

stub = modal.Stub()
stub.sv = modal.SharedVolume().persist("valhalla")
image = modal.DockerhubImage("valhalla/valhalla:run-latest", setup_commands=["apt-get update", "apt-get install -y python3-pip"])