Events and state persistence

Crawlee components talk to each other through events. An event manager emits events such as "save your state now" or "here's the current CPU and memory usage", and any component can register a listener for them. The crawler doesn't need to know which components care about an event. The Snapshotter listens to system info events to drive autoscaling, and statistics, storages, and your own code listen to persist state events to save their progress.

Event manager

The EventManager class keeps a registry of listeners and invokes them whenever an event is emitted. It also emits the PERSIST_STATE event on its own at a regular interval.

Locally, Crawlee uses the LocalEventManager subclass, which additionally measures CPU and memory usage and emits it as the SYSTEM_INFO event. The intervals come from the Configuration:

• persist_state_interval defaults to 1 minute. You can set it with the CRAWLEE_PERSIST_STATE_INTERVAL_MILLIS environment variable.
• system_info_interval defaults to 1 second. You can set it with the CRAWLEE_SYSTEM_INFO_INTERVAL_MILLIS environment variable.

The event manager is one of the global services of the service locator. Call service_locator.get_event_manager() to get it. If none was set, the service locator creates a LocalEventManager from the current configuration on first access. To use a different one, register it with service_locator.set_event_manager() before you create the crawler. A crawler also accepts an event_manager argument, but components such as statistics and storages subscribe through the global service locator, so the global registration is the one to change.

The event manager is an async context manager. The crawler enters it when run() starts and exits it when the run finishes, so events are only emitted while a crawler runs. Outside a crawler, use async with event_manager: yourself.

Events

The Event enum lists the event names. Each event comes with a payload model:

• PERSIST_STATE carries EventPersistStateData. It's emitted when the event manager starts, then at every persist_state_interval, and one last time when it closes. Its is_migrating flag tells whether the process is about to be migrated.
• SYSTEM_INFO carries EventSystemInfoData with the CPU usage ratio and the memory used by the current process and its children. The LocalEventManager emits it at every system_info_interval.
• CRAWLER_STATUS carries EventCrawlerStatusData with the same status message that the crawler periodically logs.
• MIGRATING, ABORTING, and EXIT carry EventMigratingData, EventAbortingData, and EventExitData. Crawlee itself never emits them locally. They come from a platform, for example from the event manager of the Apify SDK when an Actor run is migrated to another server or aborted.

The enum also defines session pool and browser pool event names, such as SESSION_RETIRED or BROWSER_LAUNCHED, which Crawlee for Python doesn't emit.

Listening to events

Register a listener with on() and remove it with off(). Calling off() without a listener removes all listeners of the event. A listener can be sync or async and accepts either the event payload or no argument at all. Each invocation runs in its own task, and sync listeners run in a separate thread, so a slow listener doesn't block the crawler. An exception raised by a listener is logged and doesn't affect other listeners.

The following example registers an event manager with a shorter persist state interval and prints the payload of the persist state and system info events during a short crawl:

Run on

import asyncio
from datetime import timedelta

from crawlee import service_locator
from crawlee.crawlers import ParselCrawler, ParselCrawlingContext
from crawlee.events import (
    Event,
    EventPersistStateData,
    EventSystemInfoData,
    LocalEventManager,
)


async def main() -> None:
    # Register an event manager with a shorter persist state interval, so that
    # the event fires several times during this short crawl.
    event_manager = LocalEventManager(persist_state_interval=timedelta(seconds=2))
    service_locator.set_event_manager(event_manager)

    async def on_persist_state(event_data: EventPersistStateData) -> None:
        print(f'Persist state event (is_migrating={event_data.is_migrating})')

    async def on_system_info(event_data: EventSystemInfoData) -> None:
        cpu = event_data.cpu_info.used_ratio
        memory = event_data.memory_info.current_size
        print(f'System info event: CPU {cpu:.0%}, memory {memory}')

    event_manager.on(event=Event.PERSIST_STATE, listener=on_persist_state)
    event_manager.on(event=Event.SYSTEM_INFO, listener=on_system_info)

    crawler = ParselCrawler(max_requests_per_crawl=20)

    @crawler.router.default_handler
    async def request_handler(context: ParselCrawlingContext) -> None:
        context.log.info(f'Processing {context.request.url} ...')
        await context.enqueue_links()

    await crawler.run(['https://crawlee.dev'])

    # Stop listening to the system info events.
    event_manager.off(event=Event.SYSTEM_INFO, listener=on_system_info)


if __name__ == '__main__':
    asyncio.run(main())

State persistence

State persistence builds on the PERSIST_STATE event. Components that need to survive a restart or a migration keep their state in memory, save it to a KeyValueStore whenever the event fires, and load it back on the next start. The default key-value store of a crawler run then holds records like these:

• __CRAWLER_STATISTICS_<id> holds the crawler's statistics. The crawler's default Statistics instance has persistence enabled.
• __RQ_STATE_<id> holds the ordering and in-progress information of a request queue backed by the file system storage client. The requests themselves are stored as individual files.
• CRAWLEE_STATE_<id> holds the state you manage with use_state, as described in Crawler state.

Other components persist their state when you ask them to. The SessionPool does so with persistence_enabled=True, and the RequestList with a persist_state_key. For details, see Request loaders.

Internally, these components use a helper class that loads the state on start, subscribes to PERSIST_STATE, and saves the state one last time on teardown. The helper isn't part of the public API. In your own code, use use_state, or listen to PERSIST_STATE and save your data to a key-value store.

Persisted state only helps if it's still there on the next run. Crawlee purges the default storages on start, so to resume an interrupted crawl, disable purge_on_start. For details, see Stopping and resuming crawlers.

Crawler state

The use_state helper of the crawling context returns a dictionary that's saved automatically. Every call within a crawler returns the same dictionary, so you can read and update it from any request handler. Crawlee saves it to the default key-value store on every PERSIST_STATE event and when the run finishes, and loads it back from there when the crawler starts again with existing storage. The optional argument sets the initial value when no saved state exists yet.

The dictionary is stored under the CRAWLEE_STATE_<id> key, where the id is the crawler's id argument or, without one, a per-process counter of crawler instances. Pass the same id to the constructors of several crawlers to make them share their state. Outside request handlers, use crawler.use_state(). Values have to be JSON serializable.

The following example stores the page title of every visited URL in the crawler state and reads it after the run:

Run on

import asyncio

from crawlee.crawlers import ParselCrawler, ParselCrawlingContext


async def main() -> None:
    crawler = ParselCrawler(max_requests_per_crawl=10)

    @crawler.router.default_handler
    async def request_handler(context: ParselCrawlingContext) -> None:
        # Every call returns the same dictionary. Crawlee saves it to the default
        # key-value store on each persist state event and at the end of the run.
        state = await context.use_state()
        state[context.request.url] = context.selector.css('title::text').get()

        await context.enqueue_links()

    await crawler.run(['https://crawlee.dev'])

    # The crawler exposes the same state outside the request handlers.
    state = await crawler.use_state()
    print(f'Title of the start page: {state["https://crawlee.dev"]}')


if __name__ == '__main__':
    asyncio.run(main())

use_state is an experimental feature, so its behavior and interface may change in future versions.