Dicts and Queues

import modal

app = modal.App()

kv = modal.Dict.from_name("kv", create_if_missing=True)
fifo = modal.Queue.from_name("fifo", create_if_missing=True)

@app.local_entrypoint()
def main(flush=True):
    n = kv.get("count", 0) + 1
    kv["count"] = n
    fifo.put(n)
    if flush:
        consume_queue.remote()

@app.function()
def consume_queue():
    while item := fifo.get(block=False):
        print(item)

# create a persistent Dict
dictionary = modal.Dict.from_name("my-dict", create_if_missing=True)


@app.local_entrypoint()
def main():
    dictionary["key"] = "value"  # setting a value
    value = dictionary["key"]    # getting a value

import modal

app = modal.App()
dictionary = modal.Dict.from_name("my-dict", create_if_missing=True)


@app.local_entrypoint()
async def main():
    await dictionary.put.aio("key", "value")  # setting a value asynchronously
    assert await dictionary.get.aio("key")   # getting a value asyncrhonrously

dictionary = modal.Dict.from_name("my-dict", create_if_missing=True)

@app.function(image=modal.Image.debian_slim().pip_install("torch"))
def fill():
    import torch

    dictionary["torch"] = torch
    dictionary["modal"] = modal
    dictionary[dictionary] = dictionary  # don't try this at home!

@app.local_entrypoint()
def main():
    fill.remote()
    print(dictionary["modal"])
    print(dictionary[dictionary]["modal"].Dict)
    # print(dictionary["torch"])  # DeserializationError, no torch locally

import modal

app = modal.App()
my_queue = modal.Queue.from_name("my-persisted-queue", create_if_missing=True)


@app.local_entrypoint()
def main():
    my_queue.put("some object")  # adding a value
    value = my_queue.get()  # retrieving a value

import modal

app = modal.App()
my_queue = modal.Queue.from_name("my-persisted-queue", create_if_missing=True)


@app.local_entrypoint()
def main():
    my_queue.put("some value")
    my_queue.put(123)

    assert my_queue.get() == "some value"
    assert my_queue.get() == 123

    my_queue.put(0)
    my_queue.put(1, partition="foo")
    my_queue.put(2, partition="bar")

    # Default and "foo" partition are ignored by the get operation.
    assert my_queue.get(partition="bar") == 2

    # Set custom 10s expiration time on "foo" partition.
    my_queue.put(3, partition="foo", partition_ttl=10)

    # (beta feature) Iterate through items in place (read immutably)
    my_queue.put(1)
    assert [v for v in my_queue.iterate()] == [0, 1]

@app.local_entrypoint()
async def main(value=None):
    await my_queue.put.aio(value or 200)
    assert await my_queue.get.aio() == value

import queue
import sys
from datetime import datetime

import modal


app = modal.App(image=modal.Image.debian_slim().pip_install("requests", "beautifulsoup4"))


def extract_links(url: str) -> list[str]:
    """Extract links from a given URL."""
    import requests
    import urllib.parse
    from bs4 import BeautifulSoup

    resp = requests.get(url, timeout=10)
    resp.raise_for_status()
    soup = BeautifulSoup(resp.text, "html.parser")
    links = []
    for link in soup.find_all("a"):
        links.append(urllib.parse.urljoin(url, link.get("href")))
    return links


@app.function()
def crawl_pages(q: modal.Queue, d: modal.Dict, urls: set[str]) -> None:
    for url in urls:
        if "stop" in d:
            return
        try:
            s = datetime.now()
            links = extract_links(url)
            print(f"Crawled: {url} in {datetime.now() - s}, with {len(links)} links")
            q.put_many(links)
        except Exception as exc:
            print(f"Failed to crawl: {url} with error {exc}, skipping...", file=sys.stderr)


@app.function()
def scrape(url: str):
    start_time = datetime.now()

    # Create ephemeral dicts and queues
    with modal.Dict.ephemeral() as d, modal.Queue.ephemeral() as q:
        # The dict is used to signal the scraping to stop
        # The queue contains the URLs that have been crawled

        # Initialize queue with a starting URL
        q.put(url)

        # Crawl until the queue is empty, or reaching some number of URLs
        visited = set()
        max_urls = 50000
        while True:
            try:
                next_urls = q.get_many(2000, timeout=5)
            except queue.Empty:
                break
            new_urls = set(next_urls) - visited
            visited |= new_urls
            if len(visited) < max_urls:
                crawl_pages.spawn(q, d, new_urls)
            else:
                d["stop"] = True

        elapsed = (datetime.now() - start_time).total_seconds()
        print(f"Crawled {len(visited)} URLs in {elapsed:.2f} seconds")


@app.local_entrypoint()
def main():
    scrape.remote("https://www.wikipedia.org/")