_actor.py

from __future__ import annotations

import asyncio
import sys
import warnings
from contextlib import suppress
from datetime import datetime, timedelta, timezone
from functools import cached_property
from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast, overload

from lazy_object_proxy import Proxy
from more_itertools import flatten
from pydantic import AliasChoices

from apify_client import ApifyClientAsync
from apify_shared.consts import ActorEnvVars, ActorExitCodes, ApifyEnvVars
from crawlee import service_locator
from crawlee.errors import ServiceConflictError
from crawlee.events import (
    Event,
    EventAbortingData,
    EventExitData,
    EventListener,
    EventMigratingData,
    EventPersistStateData,
    EventSystemInfoData,
)

from apify._charging import ChargeResult, ChargingManager, ChargingManagerImplementation
from apify._configuration import Configuration
from apify._consts import EVENT_LISTENERS_TIMEOUT
from apify._crypto import decrypt_input_secrets, load_private_key
from apify._models import ActorRun
from apify._proxy_configuration import ProxyConfiguration
from apify._utils import docs_group, docs_name, ensure_context, get_system_info, is_running_in_ipython
from apify.events import ApifyEventManager, EventManager, LocalEventManager
from apify.log import _configure_logging, logger
from apify.storage_clients import ApifyStorageClient, SmartApifyStorageClient
from apify.storage_clients._file_system import ApifyFileSystemStorageClient
from apify.storages import Dataset, KeyValueStore, RequestQueue

if TYPE_CHECKING:
    import logging
    from collections.abc import Callable
    from types import TracebackType

    from typing_extensions import Self

    from crawlee._types import JsonSerializable
    from crawlee.proxy_configuration import _NewUrlFunction

    from apify._models import Webhook

MainReturnType = TypeVar('MainReturnType')

_ensure_context = ensure_context('_active')


@docs_name('Actor')
@docs_group('Actor')
class _ActorType:
    """The core class for building Actors on the Apify platform.

    Actors are serverless programs running in the cloud that can perform anything from simple actions
    (such as filling out a web form or sending an email) to complex operations (such as crawling an
    entire website or removing duplicates from a large dataset). They are packaged as Docker containers
    which accept well-defined JSON input, perform an action, and optionally produce well-defined output.

    ### References

    - Apify platform documentation: https://docs.apify.com/platform/actors
    - Actor whitepaper: https://whitepaper.actor/

    ### Usage

    ```python
    import asyncio

    import httpx
    from apify import Actor
    from bs4 import BeautifulSoup


    async def main() -> None:
        async with Actor:
            actor_input = await Actor.get_input()
            async with httpx.AsyncClient() as client:
                response = await client.get(actor_input['url'])
            soup = BeautifulSoup(response.content, 'html.parser')
            data = {
                'url': actor_input['url'],
                'title': soup.title.string if soup.title else None,
            }
            await Actor.push_data(data)

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

    _ACTOR_STATE_KEY = 'APIFY_GLOBAL_STATE'

    def __init__(
        self,
        configuration: Configuration | None = None,
        *,
        configure_logging: bool = True,
        exit_process: bool | None = None,
        exit_code: int = 0,
        status_message: str | None = None,
        event_listeners_timeout: timedelta | None = EVENT_LISTENERS_TIMEOUT,
        cleanup_timeout: timedelta = timedelta(seconds=30),
    ) -> None:
        """Initialize a new instance.

        Args:
            configuration: The Actor configuration to use. If not provided, a default configuration is created.
            configure_logging: Whether to set up the default logging configuration.
            exit_process: Whether the Actor should call `sys.exit` when the context manager exits.
                Defaults to True, except in IPython, Pytest, and Scrapy environments.
            exit_code: The exit code the Actor should use when exiting.
            status_message: Final status message to display upon Actor termination.
            event_listeners_timeout: Maximum time to wait for Actor event listeners to complete before exiting.
            cleanup_timeout: Maximum time to wait for cleanup tasks to finish.
        """
        self._configuration = configuration
        self._configure_logging = configure_logging
        self._exit_process = self._get_default_exit_process() if exit_process is None else exit_process
        self._exit_code = exit_code
        self._status_message = status_message
        self._event_listeners_timeout = event_listeners_timeout
        self._cleanup_timeout = cleanup_timeout

        # Actor state when this method is being executed is unpredictable.
        # Actor can be initialized by lazy object proxy or by user directly, or by both.
        # Until `init` method is run, this state of uncertainty remains. This is the reason why any setting done here in
        # `__init__` method should not be considered final.

        self._apify_client: ApifyClientAsync | None = None

        # Keep track of all used state stores to persist their values on exit
        self._use_state_stores: set[str | None] = set()

        self._active = False
        """Whether the Actor instance is currently active (initialized and within context)."""

        self._is_rebooting = False
        """Whether the Actor is currently rebooting."""

        self._is_exiting = False
        """Whether the Actor is currently exiting."""

    async def __aenter__(self) -> Self:
        """Enter the Actor context.

        Initializes the Actor when used in an `async with` block. This method:

        - Sets up local or cloud storage clients depending on whether the Actor runs locally or on the Apify platform.
        - Configures the event manager and starts periodic state persistence.
        - Initializes the charging manager for handling charging events.
        - Configures logging after all core services are registered.

        This method must be called exactly once per Actor instance. Re-initializing an Actor or having multiple
        active Actor instances is not standard usage and may lead to warnings or unexpected behavior.
        """
        if self._active:
            raise RuntimeError('The Actor was already initialized!')

        # Initialize configuration first - it's required for the next steps.
        if self._configuration:
            # User provided explicit configuration - register it in the service locator.
            service_locator.set_configuration(self.configuration)
        else:
            # No explicit configuration provided - trigger creation of default configuration.
            _ = self.configuration

        # Configure logging based on the configuration, any logs before this point are lost.
        if self._configure_logging:
            _configure_logging()
            self.log.debug('Logging configured')

        self.log.info('Initializing Actor', extra=get_system_info())
        self.log.debug('Configuration initialized')

        # Update the global Actor proxy to refer to this instance.
        cast('Proxy', Actor).__wrapped__ = self  # ty: ignore[invalid-assignment]
        self._is_exiting = False
        self._was_final_persist_state_emitted = False

        # Initialize the storage client and register it in the service locator.
        _ = self._storage_client
        self.log.debug('Storage client initialized')

        # Initialize the event manager and register it in the service locator.
        await self.event_manager.__aenter__()
        self.log.debug('Event manager initialized')

        # Initialize the charging manager.
        await self._charging_manager_implementation.__aenter__()
        self.log.debug('Charging manager initialized')

        # Mark initialization as complete and update global state.
        self._active = True

        if not Actor.is_at_home():
            # Make sure that the input related KVS is initialized to ensure that the input aware client is used
            await self.open_key_value_store()
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        exc_traceback: TracebackType | None,
    ) -> None:
        """Exit the Actor context.

        If the block exits with an exception, the Actor fails with a non-zero exit code.
        Otherwise, it exits cleanly. In both cases the Actor:

        - Cancels periodic `PERSIST_STATE` events.
        - Sends a final `PERSIST_STATE` event.
        - Waits for all event listeners to finish.
        - Stops the event manager and the charging manager.
        - Optionally terminates the process with the selected exit code.
        """
        if self._is_exiting:
            return

        if not self._active:
            raise RuntimeError('The _ActorType is not active. Use it within the async context.')

        if exc_value and not is_running_in_ipython():
            # In IPython, we don't run `sys.exit()` during Actor exits,
            # so the exception traceback will be printed on its own
            self.log.exception('Actor failed with an exception', exc_info=exc_value)
            self.exit_code = ActorExitCodes.ERROR_USER_FUNCTION_THREW.value

        self._is_exiting = True
        self.log.info('Exiting Actor', extra={'exit_code': self.exit_code})

        async def finalize() -> None:
            if self.status_message is not None:
                await self.set_status_message(self.status_message, is_terminal=True)

            # Sleep for a bit so that the listeners have a chance to trigger
            await asyncio.sleep(0.1)

            if self._event_listeners_timeout:
                await self.event_manager.wait_for_all_listeners_to_complete(timeout=self._event_listeners_timeout)

            await self.event_manager.__aexit__(None, None, None)
            await self._charging_manager_implementation.__aexit__(None, None, None)

            # Persist Actor state
            await self._save_actor_state()

        try:
            await asyncio.wait_for(finalize(), self._cleanup_timeout.total_seconds())
        except TimeoutError:
            self.log.exception('Actor cleanup timed out')
        finally:
            self._active = False

        if self._exit_process:
            sys.exit(self.exit_code)

    def __repr__(self) -> str:
        if self is cast('Proxy', Actor).__wrapped__:
            return '<apify.Actor>'

        return super().__repr__()

    def __call__(
        self,
        configuration: Configuration | None = None,
        *,
        configure_logging: bool = True,
        exit_process: bool | None = None,
        exit_code: int = 0,
        event_listeners_timeout: timedelta | None = EVENT_LISTENERS_TIMEOUT,
        status_message: str | None = None,
        cleanup_timeout: timedelta = timedelta(seconds=30),
    ) -> Self:
        """Make a new Actor instance with a non-default configuration.

        This is necessary due to the lazy object proxying of the global `Actor` instance.
        """
        return self.__class__(
            configuration=configuration,
            configure_logging=configure_logging,
            exit_process=exit_process,
            exit_code=exit_code,
            event_listeners_timeout=event_listeners_timeout,
            status_message=status_message,
            cleanup_timeout=cleanup_timeout,
        )

    @property
    def log(self) -> logging.Logger:
        """Logger configured for this Actor."""
        return logger

    @property
    def exit_code(self) -> int:
        """The exit code the Actor will use when exiting."""
        return self._exit_code

    @exit_code.setter
    def exit_code(self, value: int) -> None:
        self._exit_code = value

    @property
    def status_message(self) -> str | None:
        """The final status message that the Actor will display upon termination."""
        return self._status_message

    @status_message.setter
    def status_message(self, value: str | None) -> None:
        self._status_message = value

    @property
    def apify_client(self) -> ApifyClientAsync:
        """Asynchronous Apify client for interacting with the Apify API."""
        if not self._apify_client:
            self._apify_client = self.new_client()
        return self._apify_client

    @cached_property
    def configuration(self) -> Configuration:
        """Actor configuration, uses the default instance if not explicitly set."""
        if self._configuration:
            return self._configuration

        try:
            # Set implicit default Apify configuration, unless configuration was already set.
            implicit_configuration = Configuration()
            service_locator.set_configuration(implicit_configuration)
            self._configuration = implicit_configuration
        except ServiceConflictError:
            self.log.debug(
                'Configuration in service locator was set explicitly before Actor.init was called.'
                'Using the existing configuration as implicit configuration for the Actor.'
            )

        # Use the configuration from the service locator
        self._configuration = Configuration.get_global_configuration()
        return self._configuration

    @cached_property
    def event_manager(self) -> EventManager:
        """Manages Apify platform events.

        It uses `ApifyEventManager` on the Apify platform and `LocalEventManager` otherwise.
        """
        try:
            event_manager = (
                ApifyEventManager(
                    configuration=self.configuration,
                    persist_state_interval=self.configuration.persist_state_interval,
                )
                if self.is_at_home()
                else LocalEventManager(
                    system_info_interval=self.configuration.system_info_interval,
                    persist_state_interval=self.configuration.persist_state_interval,
                )
            )
            service_locator.set_event_manager(event_manager)
        except ServiceConflictError:
            self.log.debug(
                'Event manager already exists in service locator (set by previous Actor context or explicitly by '
                'user). Using the existing event manager.'
            )
            # Use the event manager from the service locator
            event_manager = service_locator.get_event_manager()

        return event_manager

    @cached_property
    def _charging_manager_implementation(self) -> ChargingManagerImplementation:
        return ChargingManagerImplementation(self.configuration, self.apify_client)

    @cached_property
    def _storage_client(self) -> SmartApifyStorageClient:
        """Storage client used by the Actor.

        Depending on the initialization of the service locator the client can be created in different ways.
        """
        try:
            # Nothing was set by the user.
            implicit_storage_client = SmartApifyStorageClient(
                local_storage_client=ApifyFileSystemStorageClient(), cloud_storage_client=ApifyStorageClient()
            )
            service_locator.set_storage_client(implicit_storage_client)
        except ServiceConflictError:
            self.log.debug(
                'Storage client in service locator was set explicitly before Actor.init was called. '
                'Using the existing storage client as implicit storage client for the Actor.'
            )
        else:
            return implicit_storage_client

        # User set something in the service locator.
        explicit_storage_client = service_locator.get_storage_client()
        if isinstance(explicit_storage_client, SmartApifyStorageClient):
            # The client was manually set to the right type in the service locator. This is the explicit way.
            return explicit_storage_client

        raise RuntimeError(
            'The storage client in the service locator has to be instance of SmartApifyStorageClient. If you want to '
            'set the storage client manually you have to call '
            '`service_locator.set_storage_client(SmartApifyStorageClient(...))` before entering Actor context or '
            'awaiting `Actor.init`.'
        )

    async def init(self) -> None:
        """Initialize the Actor without using context-manager syntax.

        Equivalent to `await Actor.__aenter__()`.
        """
        await self.__aenter__()

    async def exit(
        self,
        *,
        exit_code: int = 0,
        status_message: str | None = None,
        event_listeners_timeout: timedelta | None = EVENT_LISTENERS_TIMEOUT,
        cleanup_timeout: timedelta = timedelta(seconds=30),
    ) -> None:
        """Exit the Actor without using context-manager syntax.

        Equivalent to `await Actor.__aexit__()`.

        Args:
            exit_code: The exit code the Actor should use when exiting.
            status_message: Final status message to display upon Actor termination.
            event_listeners_timeout: Maximum time to wait for Actor event listeners to complete before exiting.
            cleanup_timeout: Maximum time to wait for cleanup tasks to finish.
        """
        self.exit_code = exit_code
        self.status_message = status_message
        self._event_listeners_timeout = event_listeners_timeout
        self._cleanup_timeout = cleanup_timeout
        await self.__aexit__(None, None, None)

    async def fail(
        self,
        *,
        exit_code: int = 1,
        exception: BaseException | None = None,
        status_message: str | None = None,
    ) -> None:
        """Fail the Actor instance without using context-manager syntax.

        Equivalent to setting the `self.exit_code` and `self.status_message` properties and using
        `await Actor.__aexit__()`.

        Args:
            exit_code: The exit code with which the Actor should fail (defaults to `1`).
            exception: The exception with which the Actor failed.
            status_message: The final status message that the Actor should display.
        """
        self.exit_code = exit_code
        self.status_message = status_message

        await self.__aexit__(
            exc_type=type(exception) if exception else None,
            exc_value=exception,
            exc_traceback=exception.__traceback__ if exception else None,
        )

    def new_client(
        self,
        *,
        token: str | None = None,
        api_url: str | None = None,
        max_retries: int | None = None,
        min_delay_between_retries: timedelta | None = None,
        timeout: timedelta | None = None,
    ) -> ApifyClientAsync:
        """Return a new instance of the Apify API client.

        The `ApifyClientAsync` class is provided by the [apify-client](https://github.com/apify/apify-client-python)
        package, and it is automatically configured using the `APIFY_API_BASE_URL` and `APIFY_TOKEN` environment
        variables.

        You can override the token via the available options. That's useful if you want to use the client
        as a different Apify user than the SDK internals are using.

        Args:
            token: The Apify API token.
            api_url: The URL of the Apify API server to which to connect to. Defaults to https://api.apify.com.
            max_retries: How many times to retry a failed request at most.
            min_delay_between_retries: How long will the client wait between retrying requests
                (increases exponentially from this value).
            timeout: The socket timeout of the HTTP requests sent to the Apify API.
        """
        token = token or self.configuration.token
        api_url = api_url or self.configuration.api_base_url
        return ApifyClientAsync(
            token=token,
            api_url=api_url,
            max_retries=max_retries,
            min_delay_between_retries_millis=int(min_delay_between_retries.total_seconds() * 1000)
            if min_delay_between_retries is not None
            else None,
            timeout_secs=int(timeout.total_seconds()) if timeout else None,
        )

    @_ensure_context
    async def open_dataset(
        self,
        *,
        id: str | None = None,
        alias: str | None = None,
        name: str | None = None,
        force_cloud: bool = False,
    ) -> Dataset:
        """Open a dataset.

        Datasets are used to store structured data where each object stored has the same attributes, such as online
        store products or real estate offers. The actual data is stored either on the local filesystem or in
        the Apify cloud.

        Args:
            id: The ID of the dataset to open. If provided, searches for existing dataset by ID.
                Mutually exclusive with name and alias.
            name: The name of the dataset to open (global scope, persists across runs).
                Mutually exclusive with id and alias.
            alias: The alias of the dataset to open (run scope, creates unnamed storage).
                Mutually exclusive with id and name.
            force_cloud: If set to `True` then the Apify cloud storage is always used. This way it is possible
                to combine local and cloud storage.

        Returns:
            An instance of the `Dataset` class for the given ID or name.
        """
        return await Dataset.open(
            id=id,
            name=name,
            alias=alias,
            storage_client=self._storage_client.get_suitable_storage_client(force_cloud=force_cloud),
        )

    @_ensure_context
    async def open_key_value_store(
        self,
        *,
        id: str | None = None,
        alias: str | None = None,
        name: str | None = None,
        force_cloud: bool = False,
    ) -> KeyValueStore:
        """Open a key-value store.

        Key-value stores are used to store records or files, along with their MIME content type. The records are stored
        and retrieved using a unique key. The actual data is stored either on a local filesystem or in the Apify cloud.

        Args:
            id: The ID of the KVS to open. If provided, searches for existing KVS by ID.
                Mutually exclusive with name and alias.
            name: The name of the KVS to open (global scope, persists across runs).
                Mutually exclusive with id and alias.
            alias: The alias of the KVS to open (run scope, creates unnamed storage).
                Mutually exclusive with id and name.
            force_cloud: If set to `True` then the Apify cloud storage is always used. This way it is possible
                to combine local and cloud storage.

        Returns:
            An instance of the `KeyValueStore` class for the given ID or name.
        """
        return await KeyValueStore.open(
            id=id,
            name=name,
            alias=alias,
            storage_client=self._storage_client.get_suitable_storage_client(force_cloud=force_cloud),
        )

    @_ensure_context
    async def open_request_queue(
        self,
        *,
        id: str | None = None,
        alias: str | None = None,
        name: str | None = None,
        force_cloud: bool = False,
    ) -> RequestQueue:
        """Open a request queue.

        Request queue represents a queue of URLs to crawl, which is stored either on local filesystem or in
        the Apify cloud. The queue is used for deep crawling of websites, where you start with several URLs and then
        recursively follow links to other pages. The data structure supports both breadth-first and depth-first
        crawling orders.

        Args:
            id: The ID of the RQ to open. If provided, searches for existing RQ by ID.
                Mutually exclusive with name and alias.
            name: The name of the RQ to open (global scope, persists across runs).
                Mutually exclusive with id and alias.
            alias: The alias of the RQ to open (run scope, creates unnamed storage).
                Mutually exclusive with id and name.
            force_cloud: If set to `True` then the Apify cloud storage is always used. This way it is possible
                to combine local and cloud storage.

        Returns:
            An instance of the `RequestQueue` class for the given ID or name.
        """
        return await RequestQueue.open(
            id=id,
            name=name,
            alias=alias,
            storage_client=self._storage_client.get_suitable_storage_client(force_cloud=force_cloud),
        )

    @overload
    async def push_data(self, data: dict | list[dict]) -> None: ...
    @overload
    async def push_data(self, data: dict | list[dict], charged_event_name: str) -> ChargeResult: ...
    @_ensure_context
    async def push_data(self, data: dict | list[dict], charged_event_name: str | None = None) -> ChargeResult | None:
        """Store an object or a list of objects to the default dataset of the current Actor run.

        Args:
            data: The data to push to the default dataset.
            charged_event_name: If provided and if the Actor uses the pay-per-event pricing model,
                the method will attempt to charge for the event for each pushed item.
        """
        if not data:
            return None

        data = data if isinstance(data, list) else [data]

        if charged_event_name and charged_event_name.startswith('apify-'):
            raise ValueError(f'Cannot charge for synthetic event "{charged_event_name}" manually')

        charging_manager = self.get_charging_manager()

        # Acquire the charge lock to prevent race conditions between concurrent
        # push_data calls. We need to hold the lock for the entire push_data + charge sequence.
        async with charging_manager.charge_lock:
            # No explicit charging requested; synthetic events are handled within dataset.push_data.
            if charged_event_name is None:
                dataset = await self.open_dataset()
                await dataset.push_data(data)
                return None

            pushed_items_count = self.get_charging_manager().compute_push_data_limit(
                items_count=len(data),
                event_name=charged_event_name,
                is_default_dataset=True,
            )

            dataset = await self.open_dataset()

            if pushed_items_count < len(data):
                await dataset.push_data(data[:pushed_items_count])
            elif pushed_items_count > 0:
                await dataset.push_data(data)

            # Only charge explicit events; synthetic events will be processed within the client.
            return await self.get_charging_manager().charge(
                event_name=charged_event_name,
                count=pushed_items_count,
            )

    @_ensure_context
    async def get_input(self) -> Any:
        """Get the Actor input value from the default key-value store associated with the current Actor run."""
        input_value = await self.get_value(self.configuration.input_key)
        input_secrets_private_key = self.configuration.input_secrets_private_key_file
        input_secrets_key_passphrase = self.configuration.input_secrets_private_key_passphrase
        if input_secrets_private_key and input_secrets_key_passphrase:
            private_key = load_private_key(
                input_secrets_private_key,
                input_secrets_key_passphrase,
            )
            input_value = decrypt_input_secrets(private_key, input_value)

        return input_value

    @_ensure_context
    async def get_value(self, key: str, default_value: Any = None) -> Any:
        """Get a value from the default key-value store associated with the current Actor run.

        Args:
            key: The key of the record which to retrieve.
            default_value: Default value returned in case the record does not exist.
        """
        key_value_store = await self.open_key_value_store()
        return await key_value_store.get_value(key, default_value)

    @_ensure_context
    async def set_value(
        self,
        key: str,
        value: Any,
        *,
        content_type: str | None = None,
    ) -> None:
        """Set or delete a value in the default key-value store associated with the current Actor run.

        Args:
            key: The key of the record which to set.
            value: The value of the record which to set, or None, if the record should be deleted.
            content_type: The content type which should be set to the value.
        """
        key_value_store = await self.open_key_value_store()
        return await key_value_store.set_value(key, value, content_type=content_type)

    @_ensure_context
    def get_charging_manager(self) -> ChargingManager:
        """Retrieve the charging manager to access granular pricing information."""
        return self._charging_manager_implementation

    @_ensure_context
    async def charge(self, event_name: str, count: int = 1) -> ChargeResult:
        """Charge for a specified number of events - sub-operations of the Actor.

        This is relevant only for the pay-per-event pricing model.

        Args:
            event_name: Name of the event to be charged for.
            count: Number of events to charge for.
        """
        # Acquire lock to prevent race conditions with concurrent charge/push_data calls.
        charging_manager = self.get_charging_manager()
        async with charging_manager.charge_lock:
            return await charging_manager.charge(event_name, count)

    @overload
    def on(
        self, event_name: Literal[Event.PERSIST_STATE], listener: EventListener[EventPersistStateData]
    ) -> EventListener[EventPersistStateData]: ...
    @overload
    def on(
        self, event_name: Literal[Event.SYSTEM_INFO], listener: EventListener[EventSystemInfoData]
    ) -> EventListener[EventSystemInfoData]: ...
    @overload
    def on(
        self, event_name: Literal[Event.MIGRATING], listener: EventListener[EventMigratingData]
    ) -> EventListener[EventMigratingData]: ...
    @overload
    def on(
        self, event_name: Literal[Event.ABORTING], listener: EventListener[EventAbortingData]
    ) -> EventListener[EventAbortingData]: ...
    @overload
    def on(
        self, event_name: Literal[Event.EXIT], listener: EventListener[EventExitData]
    ) -> EventListener[EventExitData]: ...
    @overload
    def on(self, event_name: Event, listener: EventListener[None]) -> EventListener[Any]: ...

    @_ensure_context
    def on(self, event_name: Event, listener: EventListener[Any]) -> EventListener[Any]:
        """Add an event listener to the Actor's event manager.

        The following events can be emitted:

        - `Event.SYSTEM_INFO`: Emitted every minute; the event data contains information about the Actor's resource
          usage.

        - `Event.MIGRATING`: Emitted when the Actor on the Apify platform is about to be migrated to another worker
          server. Use this event to persist the Actor's state and gracefully stop in-progress tasks, preventing
          disruption.

        - `Event.PERSIST_STATE`: Emitted regularly (default: 60 seconds) to notify the Actor to persist its state,
          preventing work repetition after a restart. This event is emitted together with the `MIGRATING` event, where
          the `isMigrating` flag in the event data is `True`; otherwise, the flag is `False`. This event is for
          convenience; the same effect can be achieved by setting an interval and listening for the `MIGRATING` event.

        - `Event.ABORTING`: Emitted when a user aborts an Actor run on the Apify platform, allowing the Actor time
          to clean up its state if the abort is graceful.

        Args:
            event_name: The Actor event to listen for.
            listener: The function to be called when the event is emitted (can be async).
        """
        self.event_manager.on(event=event_name, listener=listener)
        return listener

    @overload
    def off(self, event_name: Literal[Event.PERSIST_STATE], listener: EventListener[EventPersistStateData]) -> None: ...
    @overload
    def off(self, event_name: Literal[Event.SYSTEM_INFO], listener: EventListener[EventSystemInfoData]) -> None: ...
    @overload
    def off(self, event_name: Literal[Event.MIGRATING], listener: EventListener[EventMigratingData]) -> None: ...
    @overload
    def off(self, event_name: Literal[Event.ABORTING], listener: EventListener[EventAbortingData]) -> None: ...
    @overload
    def off(self, event_name: Literal[Event.EXIT], listener: EventListener[EventExitData]) -> None: ...
    @overload
    def off(self, event_name: Event, listener: EventListener[None]) -> None: ...

    @_ensure_context
    def off(self, event_name: Event, listener: Callable | None = None) -> None:
        """Remove a listener, or all listeners, from an Actor event.

        Args:
            event_name: The Actor event for which to remove listeners.
            listener: The listener which is supposed to be removed. If not passed, all listeners of this event
                are removed.
        """
        self.event_manager.off(event=event_name, listener=listener)

    def is_at_home(self) -> bool:
        """Return `True` when the Actor is running on the Apify platform, and `False` otherwise (e.g. local run)."""
        return self.configuration.is_at_home

    @_ensure_context
    def get_env(self) -> dict:
        """Return a dictionary with information parsed from all the `APIFY_XXX` environment variables.

        For a list of all the environment variables, see the
        [Actor documentation](https://docs.apify.com/actors/development/environment-variables). If some variables
        are not defined or are invalid, the corresponding value in the resulting dictionary will be None.
        """
        config = dict[str, Any]()
        for field_name, field in Configuration.model_fields.items():
            if field.deprecated:
                continue

            if field.alias:
                aliases = [field.alias]
            elif isinstance(field.validation_alias, str):
                aliases = [field.validation_alias]
            elif isinstance(field.validation_alias, AliasChoices):
                aliases = cast('list[str]', field.validation_alias.choices)
            else:
                aliases = [field_name]

            for alias in aliases:
                config[alias] = getattr(self.configuration, field_name)

        env_vars = {env_var.value.lower(): env_var.name.lower() for env_var in [*ActorEnvVars, *ApifyEnvVars]}
        return {option_name: config[env_var] for env_var, option_name in env_vars.items() if env_var in config}

    @_ensure_context
    async def start(
        self,
        actor_id: str,
        run_input: Any = None,
        *,
        token: str | None = None,
        content_type: str | None = None,
        build: str | None = None,
        memory_mbytes: int | None = None,
        timeout: timedelta | None | Literal['inherit', 'RemainingTime'] = None,
        wait_for_finish: int | None = None,
        webhooks: list[Webhook] | None = None,
    ) -> ActorRun:
        """Run an Actor on the Apify platform.

        Unlike `Actor.call`, this method just starts the run without waiting for finish.

        Args:
            actor_id: The ID of the Actor to be run.
            run_input: The input to pass to the Actor run.
            token: The Apify API token to use for this request (defaults to the `APIFY_TOKEN` environment variable).
            content_type: The content type of the input.
            build: Specifies the Actor build to run. It can be either a build tag or build number. By default,
                the run uses the build specified in the default run configuration for the Actor (typically latest).
            memory_mbytes: Memory limit for the run, in megabytes. By default, the run uses a memory limit specified
                in the default run configuration for the Actor.
            timeout: Optional timeout for the run, in seconds. By default, the run uses timeout specified in
                the default run configuration for the Actor. Using `inherit` or `RemainingTime` will set timeout of the
                other Actor to the time remaining from this Actor timeout.
            wait_for_finish: The maximum number of seconds the server waits for the run to finish. By default,
                it is 0, the maximum value is 300.
            webhooks: Optional ad-hoc webhooks (https://docs.apify.com/webhooks/ad-hoc-webhooks) associated with
                the Actor run which can be used to receive a notification, e.g. when the Actor finished or failed.
                If you already have a webhook set up for the Actor or task, you do not have to add it again here.

        Returns:
            Info about the started Actor run
        """
        client = self.new_client(token=token) if token else self.apify_client

        if webhooks:
            serialized_webhooks = [
                hook.model_dump(by_alias=True, exclude_unset=True, exclude_defaults=True) for hook in webhooks
            ]
        else:
            serialized_webhooks = None

        if timeout in {'inherit', 'RemainingTime'}:
            if timeout == 'RemainingTime':
                warnings.warn(
                    '`RemainingTime` is deprecated and will be removed in version 4.0.0. Use `inherit` instead.',
                    DeprecationWarning,
                    stacklevel=2,
                )
            actor_start_timeout = self._get_remaining_time()
        elif timeout is None:
            actor_start_timeout = None
        elif isinstance(timeout, timedelta):
            actor_start_timeout = timeout
        else:
            raise ValueError(
                f'Invalid timeout {timeout!r}: expected `None`, `"inherit"`, `"RemainingTime"`, or a `timedelta`.'
            )

        api_result = await client.actor(actor_id).start(
            run_input=run_input,
            content_type=content_type,
            build=build,
            memory_mbytes=memory_mbytes,
            timeout_secs=int(actor_start_timeout.total_seconds()) if actor_start_timeout is not None else None,
            wait_for_finish=wait_for_finish,
            webhooks=serialized_webhooks,
        )

        return ActorRun.model_validate(api_result)

    @_ensure_context
    async def abort(
        self,
        run_id: str,
        *,
        token: str | None = None,
        status_message: str | None = None,
        gracefully: bool | None = None,
    ) -> ActorRun:
        """Abort given Actor run on the Apify platform using the current user account.

        The user account is determined by the `APIFY_TOKEN` environment variable.

        Args:
            run_id: The ID of the Actor run to be aborted.
            token: The Apify API token to use for this request (defaults to the `APIFY_TOKEN` environment variable).
            status_message: Status message of the Actor to be set on the platform.
            gracefully: If True, the Actor run will abort gracefully. It will send `aborting` and `persistState`
                events into the run and force-stop the run after 30 seconds. It is helpful in cases where you plan
                to resurrect the run later.

        Returns:
            Info about the aborted Actor run.
        """
        client = self.new_client(token=token) if token else self.apify_client

        if status_message:
            await client.run(run_id).update(status_message=status_message)

        api_result = await client.run(run_id).abort(gracefully=gracefully)

        return ActorRun.model_validate(api_result)

    @_ensure_context
    async def call(
        self,
        actor_id: str,
        run_input: Any = None,
        *,
        token: str | None = None,
        content_type: str | None = None,
        build: str | None = None,
        memory_mbytes: int | None = None,
        timeout: timedelta | None | Literal['inherit', 'RemainingTime'] = None,
        webhooks: list[Webhook] | None = None,
        wait: timedelta | None = None,
        logger: logging.Logger | None | Literal['default'] = 'default',
    ) -> ActorRun | None:
        """Start an Actor on the Apify Platform and wait for it to finish before returning.

        It waits indefinitely, unless the wait argument is provided.

        Args:
            actor_id: The ID of the Actor to be run.
            run_input: The input to pass to the Actor run.
            token: The Apify API token to use for this request (defaults to the `APIFY_TOKEN` environment variable).
            content_type: The content type of the input.
            build: Specifies the Actor build to run. It can be either a build tag or build number. By default,
                the run uses the build specified in the default run configuration for the Actor (typically latest).
            memory_mbytes: Memory limit for the run, in megabytes. By default, the run uses a memory limit specified
                in the default run configuration for the Actor.
            timeout: Optional timeout for the run, in seconds. By default, the run uses timeout specified in
                the default run configuration for the Actor. Using `inherit` or `RemainingTime` will set timeout of the
                other Actor to the time remaining from this Actor timeout.
            webhooks: Optional webhooks (https://docs.apify.com/webhooks) associated with the Actor run, which can
                be used to receive a notification, e.g. when the Actor finished or failed. If you already have
                a webhook set up for the Actor, you do not have to add it again here.
            wait: The maximum number of seconds the server waits for the run to finish. If not provided,
                waits indefinitely.
            logger: Logger used to redirect logs from the Actor run. Using "default" literal means that a predefined
                default logger will be used. Setting `None` will disable any log propagation. Passing custom logger
                will redirect logs to the provided logger.

        Returns:
            Info about the started Actor run.
        """
        client = self.new_client(token=token) if token else self.apify_client

        if webhooks:
            serialized_webhooks = [
                hook.model_dump(by_alias=True, exclude_unset=True, exclude_defaults=True) for hook in webhooks
            ]
        else:
            serialized_webhooks = None

        if timeout in {'inherit', 'RemainingTime'}:
            if timeout == 'RemainingTime':
                warnings.warn(
                    '`RemainingTime` is deprecated and will be removed in version 4.0.0. Use `inherit` instead.',
                    DeprecationWarning,
                    stacklevel=2,
                )

            actor_call_timeout = self._get_remaining_time()
        elif timeout is None:
            actor_call_timeout = None
        elif isinstance(timeout, timedelta):
            actor_call_timeout = timeout
        else:
            raise ValueError(
                f'Invalid timeout {timeout!r}: expected `None`, `"inherit"`, `"RemainingTime"`, or a `timedelta`.'
            )

        api_result = await client.actor(actor_id).call(
            run_input=run_input,
            content_type=content_type,
            build=build,
            memory_mbytes=memory_mbytes,
            timeout_secs=int(actor_call_timeout.total_seconds()) if actor_call_timeout is not None else None,
            webhooks=serialized_webhooks,
            wait_secs=int(wait.total_seconds()) if wait is not None else None,
            logger=logger,
        )

        return ActorRun.model_validate(api_result)

    @_ensure_context
    async def call_task(
        self,
        task_id: str,
        task_input: dict | None = None,
        *,
        build: str | None = None,
        memory_mbytes: int | None = None,
        timeout: timedelta | None | Literal['inherit'] = None,
        webhooks: list[Webhook] | None = None,
        wait: timedelta | None = None,
        token: str | None = None,
    ) -> ActorRun | None:
        """Start an Actor task on the Apify Platform and wait for it to finish before returning.

        It waits indefinitely, unless the wait argument is provided.

        Note that an Actor task is a saved input configuration and options for an Actor. If you want to run an Actor
        directly rather than an Actor task, please use the `Actor.call`

        Args:
            task_id: The ID of the Actor to be run.
            task_input: Overrides the input to pass to the Actor run.
            token: The Apify API token to use for this request (defaults to the `APIFY_TOKEN` environment variable).
            content_type: The content type of the input.
            build: Specifies the Actor build to run. It can be either a build tag or build number. By default,
                the run uses the build specified in the default run configuration for the Actor (typically latest).
            memory_mbytes: Memory limit for the run, in megabytes. By default, the run uses a memory limit specified
                in the default run configuration for the Actor.
            timeout: Optional timeout for the run, in seconds. By default, the run uses timeout specified in
                the default run configuration for the Actor. Using `inherit` will set timeout of the other Actor to the
                time remaining from this Actor timeout.
            webhooks: Optional webhooks (https://docs.apify.com/webhooks) associated with the Actor run, which can
                be used to receive a notification, e.g. when the Actor finished or failed. If you already have
                a webhook set up for the Actor, you do not have to add it again here.
            wait: The maximum number of seconds the server waits for the run to finish. If not provided, waits
                indefinitely.

        Returns:
            Info about the started Actor run.
        """
        client = self.new_client(token=token) if token else self.apify_client

        if webhooks:
            serialized_webhooks = [
                hook.model_dump(by_alias=True, exclude_unset=True, exclude_defaults=True) for hook in webhooks
            ]
        else:
            serialized_webhooks = None

        if timeout == 'inherit':
            task_call_timeout = self._get_remaining_time()
        elif timeout is None:
            task_call_timeout = None
        elif isinstance(timeout, timedelta):
            task_call_timeout = timeout
        else:
            raise ValueError(f'Invalid timeout {timeout!r}: expected `None`, `"inherit"`, or a `timedelta`.')

        api_result = await client.task(task_id).call(
            task_input=task_input,
            build=build,
            memory_mbytes=memory_mbytes,
            timeout_secs=int(task_call_timeout.total_seconds()) if task_call_timeout is not None else None,
            webhooks=serialized_webhooks,
            wait_secs=int(wait.total_seconds()) if wait is not None else None,
        )

        return ActorRun.model_validate(api_result)

    @_ensure_context
    async def metamorph(
        self,
        target_actor_id: str,
        run_input: Any = None,
        *,
        target_actor_build: str | None = None,
        content_type: str | None = None,
        custom_after_sleep: timedelta | None = None,
    ) -> None:
        """Transform this Actor run to an Actor run of a different Actor.

        The platform stops the current Actor container and starts a new container with the new Actor instead. All
        the default storages are preserved, and the new input is stored under the `INPUT-METAMORPH-1` key in the same
        default key-value store.

        Args:
            target_actor_id: ID of the target Actor that the run should be transformed into
            run_input: The input to pass to the new run.
            target_actor_build: The build of the target Actor. It can be either a build tag or build number.
                By default, the run uses the build specified in the default run configuration for the target Actor
                (typically the latest build).
            content_type: The content type of the input.
            custom_after_sleep: How long to sleep for after the metamorph, to wait for the container to be stopped.
        """
        if not self.is_at_home():
            self.log.error('Actor.metamorph() is only supported when running on the Apify platform.')
            return

        if not custom_after_sleep:
            custom_after_sleep = self.configuration.metamorph_after_sleep

        # If is_at_home() is True, configuration.actor_run_id is always set
        if not self.configuration.actor_run_id:
            raise RuntimeError('actor_run_id cannot be None when running on the Apify platform.')

        await self.apify_client.run(self.configuration.actor_run_id).metamorph(
            target_actor_id=target_actor_id,
            run_input=run_input,
            target_actor_build=target_actor_build,
            content_type=content_type,
        )

        if custom_after_sleep:
            await asyncio.sleep(custom_after_sleep.total_seconds())

    @_ensure_context
    async def reboot(
        self,
        *,
        event_listeners_timeout: timedelta | None = EVENT_LISTENERS_TIMEOUT,  # noqa: ARG002
        custom_after_sleep: timedelta | None = None,
    ) -> None:
        """Internally reboot this Actor.

        The system stops the current container and starts a new one, with the same run ID and default storages.

        Args:
            event_listeners_timeout: How long should the Actor wait for Actor event listeners to finish before exiting.
            custom_after_sleep: How long to sleep for after the reboot, to wait for the container to be stopped.
        """
        if not self.is_at_home():
            self.log.error('Actor.reboot() is only supported when running on the Apify platform.')
            return

        if self._is_rebooting:
            self.log.debug('Actor is already rebooting, skipping the additional reboot call.')
            return

        self._is_rebooting = True

        if not custom_after_sleep:
            custom_after_sleep = self.configuration.metamorph_after_sleep

        # Call all the listeners for the PERSIST_STATE and MIGRATING events, and wait for them to finish.
        # PERSIST_STATE listeners are called to allow the Actor to persist its state before the reboot.
        # MIGRATING listeners are called to allow the Actor to gracefully stop in-progress tasks before the reboot.
        # Typically, crawlers are listening for the MIIGRATING event to stop processing new requests.
        # We can't just emit the events and wait for all listeners to finish,
        # because this method might be called from an event listener itself, and we would deadlock.
        persist_state_listeners = flatten(
            (self.event_manager._listeners_to_wrappers[Event.PERSIST_STATE] or {}).values()  # noqa: SLF001
        )
        migrating_listeners = flatten(
            (self.event_manager._listeners_to_wrappers[Event.MIGRATING] or {}).values()  # noqa: SLF001
        )

        await asyncio.gather(
            *[listener(EventPersistStateData(is_migrating=True)) for listener in persist_state_listeners],
            *[listener(EventMigratingData()) for listener in migrating_listeners],
        )

        if not self.configuration.actor_run_id:
            raise RuntimeError('actor_run_id cannot be None when running on the Apify platform.')

        await self.apify_client.run(self.configuration.actor_run_id).reboot()

        if custom_after_sleep:
            await asyncio.sleep(custom_after_sleep.total_seconds())

    @_ensure_context
    async def add_webhook(
        self,
        webhook: Webhook,
        *,
        ignore_ssl_errors: bool | None = None,
        do_not_retry: bool | None = None,
        idempotency_key: str | None = None,
    ) -> None:
        """Create an ad-hoc webhook for the current Actor run.

        This webhook lets you receive a notification when the Actor run finished or failed.

        Note that webhooks are only supported for Actors running on the Apify platform. When running the Actor locally,
        the function will print a warning and have no effect.

        For more information about Apify Actor webhooks, please see the [documentation](https://docs.apify.com/webhooks).

        Args:
            webhook: The webhook to be added
            ignore_ssl_errors: Whether the webhook should ignore SSL errors returned by request_url
            do_not_retry: Whether the webhook should retry sending the payload to request_url upon failure.
            idempotency_key: A unique identifier of a webhook. You can use it to ensure that you won't create
                the same webhook multiple times.

        Returns:
            The created webhook.
        """
        if not self.is_at_home():
            self.log.error('Actor.add_webhook() is only supported when running on the Apify platform.')
            return

        # If is_at_home() is True, config.actor_run_id is always set
        if not self.configuration.actor_run_id:
            raise RuntimeError('actor_run_id cannot be None when running on the Apify platform.')

        await self.apify_client.webhooks().create(
            actor_run_id=self.configuration.actor_run_id,
            event_types=webhook.event_types,
            request_url=webhook.request_url,
            payload_template=webhook.payload_template,
            ignore_ssl_errors=ignore_ssl_errors,
            do_not_retry=do_not_retry,
            idempotency_key=idempotency_key,
        )

    @_ensure_context
    async def set_status_message(
        self,
        status_message: str,
        *,
        is_terminal: bool | None = None,
    ) -> ActorRun | None:
        """Set the status message for the current Actor run.

        Args:
            status_message: The status message to set to the run.
            is_terminal: Set this flag to True if this is the final status message of the Actor run.

        Returns:
            The updated Actor run object.
        """
        if not self.is_at_home():
            title = 'Terminal status message' if is_terminal else 'Status message'
            self.log.info(f'[{title}]: {status_message}')
            return None

        # If is_at_home() is True, config.actor_run_id is always set
        if not self.configuration.actor_run_id:
            raise RuntimeError('actor_run_id cannot be None when running on the Apify platform.')

        api_result = await self.apify_client.run(self.configuration.actor_run_id).update(
            status_message=status_message, is_status_message_terminal=is_terminal
        )

        return ActorRun.model_validate(api_result)

    @_ensure_context
    async def create_proxy_configuration(
        self,
        *,
        actor_proxy_input: dict
        | None = None,  # this is the raw proxy input from the Actor run input, it is not spread or snake_cased in here
        password: str | None = None,
        groups: list[str] | None = None,
        country_code: str | None = None,
        proxy_urls: list[str | None] | None = None,
        new_url_function: _NewUrlFunction | None = None,
    ) -> ProxyConfiguration | None:
        """Create a ProxyConfiguration object with the passed proxy configuration.

        Configures connection to a proxy server with the provided options. Proxy servers are used to prevent target
        websites from blocking your crawlers based on IP address rate limits or blacklists.

        For more details and code examples, see the `ProxyConfiguration` class.

        Args:
            actor_proxy_input: Proxy configuration field from the Actor input, if input has such input field. If you
                pass this argument, all the other arguments will be inferred from it.
            password: Password for the Apify Proxy. If not provided, will use os.environ['APIFY_PROXY_PASSWORD'],
                if available.
            groups: Proxy groups which the Apify Proxy should use, if provided.
            country_code: Country which the Apify Proxy should use, if provided.
            proxy_urls: Custom proxy server URLs which should be rotated through.
            new_url_function: Function which returns a custom proxy URL to be used.

        Returns:
            ProxyConfiguration object with the passed configuration, or None, if no proxy should be used based
            on the configuration.
        """
        if actor_proxy_input is not None:
            if actor_proxy_input.get('useApifyProxy', False):
                country_code = country_code or actor_proxy_input.get('apifyProxyCountry')
                groups = groups or actor_proxy_input.get('apifyProxyGroups')
            else:
                proxy_urls = actor_proxy_input.get('proxyUrls', [])
                if not proxy_urls:
                    return None

        proxy_configuration = ProxyConfiguration(
            password=password,
            groups=groups,
            country_code=country_code,
            proxy_urls=proxy_urls,
            new_url_function=new_url_function,
            _actor_config=self.configuration,
            _apify_client=self._apify_client,
        )

        await proxy_configuration.initialize()

        return proxy_configuration

    @_ensure_context
    async def use_state(
        self,
        default_value: dict[str, JsonSerializable] | None = None,
        key: str | None = None,
        kvs_name: str | None = None,
    ) -> dict[str, JsonSerializable]:
        """Easily create and manage state values. All state values are automatically persisted.

        Values can be modified by simply using the assignment operator.

        Args:
            default_value: The default value to initialize the state if it is not already set.
            key: The key in the key-value store where the state is stored. If not provided, a default key is used.
            kvs_name: The name of the key-value store where the state is stored. If not provided, the default
                key-value store associated with the Actor run is used.

        Returns:
            The state dictionary with automatic persistence.
        """
        self._use_state_stores.add(kvs_name)
        kvs = await self.open_key_value_store(name=kvs_name)
        return await kvs.get_auto_saved_value(key or self._ACTOR_STATE_KEY, default_value)

    async def _save_actor_state(self) -> None:
        for kvs_name in self._use_state_stores:
            store = await self.open_key_value_store(name=kvs_name)
            await store.persist_autosaved_values()

    def _get_default_exit_process(self) -> bool:
        """Return False for IPython and Scrapy environments, True otherwise."""
        if is_running_in_ipython():
            self.log.debug('Running in IPython, setting default `exit_process` to False.')
            return False

        # Check if running in Scrapy by attempting to import it.
        with suppress(ImportError):
            import scrapy  # noqa: F401 PLC0415

            self.log.debug('Running in Scrapy, setting default `exit_process` to False.')
            return False

        return True

    def _get_remaining_time(self) -> timedelta | None:
        """Get time remaining from the Actor timeout. Returns `None` if not on an Apify platform."""
        if self.is_at_home() and self.configuration.timeout_at:
            return max(self.configuration.timeout_at - datetime.now(tz=timezone.utc), timedelta(0))

        self.log.warning(
            'Using `inherit` or `RemainingTime` argument is only possible when the Actor'
            ' is running on the Apify platform and when the timeout for the Actor run is set. '
            f'{self.is_at_home()=}, {self.configuration.timeout_at=}'
        )
        return None


Actor = cast('_ActorType', Proxy(_ActorType))
"""The entry point of the SDK, through which all the Actor operations should be done."""