Dicts and queues

import modal

app = modal.App()  # Note: prior to April 2024, "app" was called "stub"

# Create a persisted dict - the data gets retained between app runs
my_dict = modal.Dict.from_name("my-persisted-dict", create_if_missing=True)


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

import modal

app = modal.App()  # Note: prior to April 2024, "app" was called "stub"
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

from modal import Queue, App

app = App()  # Note: prior to April 2024, "app" was called "stub"
my_queue = 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_key="foo")
    my_queue.put(2, partition_key="bar")

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

    # Set custom 10s expiration time on "foo" partition.
    my_queue.put(3, partition_key="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():
    await my_queue.put.aio(100)
    assert await my_queue.get.aio() == 100

    await my_dict.put.aio("hello", 400)
    assert await my_dict.get.aio("hello") == 400

import queue
import sys
from datetime import datetime

from modal import Dict, Image, Queue, App


app = App(image=Image.debian_slim().pip_install("requests", "beautifulsoup4"))  # Note: prior to April 2024, "app" was called "stub"


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: Queue, d: 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 Dict.ephemeral() as d, 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/")