tessl/pypi-apify-client
Apify API client for Python providing access to web scraping and automation platform resources
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
actors.mddocs/
Actor Management
Comprehensive Actor lifecycle management including creation, configuration, execution, builds, versions, and environment variables. Actors are the core computational units in the Apify platform.
Capabilities
Actor Operations
Individual Actor management operations including retrieval, updates, deletion, execution, and validation.
from typing import Any, Literal
from decimal import Decimal
from logging import Logger
from apify_shared.consts import ActorJobStatus, MetaOrigin
class ActorClient:
def get(self) -> dict | None:
"""Retrieve Actor details."""
def update(
self,
*,
name: str | None = None,
title: str | None = None,
description: str | None = None,
seo_title: str | None = None,
seo_description: str | None = None,
versions: list[dict] | None = None,
restart_on_error: bool | None = None,
is_public: bool | None = None,
is_deprecated: bool | None = None,
is_anonymously_runnable: bool | None = None,
categories: list[str] | None = None,
default_run_build: str | None = None,
default_run_max_items: int | None = None,
default_run_memory_mbytes: int | None = None,
default_run_timeout_secs: int | None = None,
example_run_input_body: Any = None,
example_run_input_content_type: str | None = None,
actor_standby_is_enabled: bool | None = None,
actor_standby_desired_requests_per_actor_run: int | None = None,
actor_standby_max_requests_per_actor_run: int | None = None,
actor_standby_idle_timeout_secs: int | None = None,
actor_standby_build: str | None = None,
actor_standby_memory_mbytes: int | None = None,
pricing_infos: list[dict] | None = None,
) -> dict:
"""Update Actor configuration.
Args:
name: The name of the Actor
title: The title of the Actor (human-readable)
description: The description for the Actor
seo_title: The title of the Actor optimized for search engines
seo_description: The description of the Actor optimized for search engines
versions: The list of Actor versions
restart_on_error: If true, the main Actor run process will be restarted whenever it exits with non-zero status code
is_public: Whether the Actor is public
is_deprecated: Whether the Actor is deprecated
is_anonymously_runnable: Whether the Actor is anonymously runnable
categories: The categories to which the Actor belongs to
default_run_build: Tag or number of the build to run by default
default_run_max_items: Default limit of results returned by runs (if charged per result)
default_run_memory_mbytes: Default memory allocated for runs in megabytes
default_run_timeout_secs: Default timeout for runs in seconds
example_run_input_body: Input to be prefilled as default for new users
example_run_input_content_type: Content type of the example run input
actor_standby_is_enabled: Whether Actor Standby is enabled
actor_standby_desired_requests_per_actor_run: Desired concurrent HTTP requests per Standby run
actor_standby_max_requests_per_actor_run: Maximum concurrent HTTP requests per Standby run
actor_standby_idle_timeout_secs: Shutdown timeout when Actor receives no requests
actor_standby_build: Build tag or number to run in Standby mode
actor_standby_memory_mbytes: Memory in megabytes for Standby mode
pricing_infos: List of objects describing Actor pricing
"""
def delete(self) -> None:
"""Delete the Actor."""
def start(
self,
*,
run_input: Any = None,
content_type: str | None = None,
build: str | None = None,
max_items: int | None = None,
max_total_charge_usd: Decimal | None = None,
memory_mbytes: int | None = None,
timeout_secs: int | None = None,
wait_for_finish: int | None = None,
webhooks: list[dict] | None = None,
) -> dict:
"""Start Actor run and return immediately.
Args:
run_input: Input data for the Actor run
content_type: Content type of the run_input
build: Tag or number of the build to run
max_items: Maximum number of results (if charged per result)
max_total_charge_usd: Maximum total charge in USD for the run
memory_mbytes: Memory limit in megabytes
timeout_secs: Timeout in seconds
wait_for_finish: Time to wait for run to finish (in seconds)
webhooks: Webhook configuration list
"""
def call(
self,
*,
run_input: Any = None,
content_type: str | None = None,
build: str | None = None,
max_items: int | None = None,
max_total_charge_usd: Decimal | None = None,
memory_mbytes: int | None = None,
timeout_secs: int | None = None,
webhooks: list[dict] | None = None,
wait_secs: int | None = None,
logger: Logger | None | Literal['default'] = 'default',
) -> dict | None:
"""Start Actor and wait for completion.
Args:
run_input: Input data for the Actor run
content_type: Content type of the run_input
build: Tag or number of the build to run
max_items: Maximum number of results (if charged per result)
max_total_charge_usd: Maximum total charge in USD for the run
memory_mbytes: Memory limit in megabytes
timeout_secs: Timeout in seconds
webhooks: Webhook configuration list
wait_secs: Maximum time to wait for completion
logger: Logger instance or 'default' for default logger
"""
def build(
self,
*,
version_number: str,
beta_packages: bool | None = None,
tag: str | None = None,
use_cache: bool | None = None,
wait_for_finish: int | None = None,
) -> dict:
"""Build the Actor.
Args:
version_number: Actor version number to be built (required)
beta_packages: If True, build with beta versions of Apify NPM packages
tag: Tag to be applied to the build on success
use_cache: If true, rebuild using Docker layer cache
wait_for_finish: Maximum seconds to wait for build completion (max 60)
"""
def builds(self) -> BuildCollectionClient:
"""Get builds collection client."""
def runs(self) -> RunCollectionClient:
"""Get runs collection client."""
def default_build(self, **kwargs) -> BuildClient:
"""Get default build client."""
def last_run(self, **kwargs) -> RunClient:
"""Get last run client.
Args:
status (str, optional): Filter by run status
origin (str, optional): Filter by run origin
"""
def versions(self) -> ActorVersionCollectionClient:
"""Get versions collection client."""
def version(self, version_number: str) -> ActorVersionClient:
"""Get specific version client."""
def webhooks(self) -> WebhookCollectionClient:
"""Get webhooks collection client."""
def validate_input(
self,
run_input: Any = None,
*,
build_tag: str | None = None,
content_type: str | None = None
) -> bool:
"""Validate Actor input against schema.
Args:
run_input: Input data to validate
build_tag: Tag of the build to validate against
content_type: Content type of the input
"""
class ActorClientAsync:
"""Async version of ActorClient with identical methods."""Actor Collection Operations
Operations for listing and creating Actors across the platform.
class ActorCollectionClient:
def list(self, **kwargs) -> ListPage[dict]:
"""List Actors.
Args:
my (bool, optional): Show only user's Actors
limit (int, optional): Maximum number of items
offset (int, optional): Offset for pagination
desc (bool, optional): Sort in descending order
sort_by (str, optional): Field to sort by
"""
def create(self, **kwargs) -> dict:
"""Create new Actor.
Args:
name (str): Actor name
title (str, optional): Display title
description (str, optional): Actor description
versions (list, optional): Initial version configuration
**kwargs: Additional Actor configuration
"""
class ActorCollectionClientAsync:
"""Async version of ActorCollectionClient with identical methods."""Actor Version Management
Management of Actor versions including source code, environment variables, and configuration.
class ActorVersionClient:
def get(self) -> dict | None:
"""Get Actor version information."""
def update(self, **kwargs) -> dict:
"""Update Actor version.
Args:
build_tag (str, optional): Build tag
env_vars (list, optional): Environment variables
source_type (str, optional): Source code type
source_files (list, optional): Source files
**kwargs: Additional version configuration
"""
def delete(self) -> None:
"""Delete Actor version."""
def env_vars(self) -> ActorEnvVarCollectionClient:
"""Get environment variables collection."""
def env_var(self, env_var_name: str) -> ActorEnvVarClient:
"""Get specific environment variable client."""
class ActorVersionClientAsync:
"""Async version of ActorVersionClient with identical methods."""
class ActorVersionCollectionClient:
def list(self) -> ListPage[dict]:
"""List Actor versions."""
def create(self, **kwargs) -> dict:
"""Create new Actor version.
Args:
version_number (str): Version number
source_type (str, optional): Source code type
**kwargs: Version configuration
"""
class ActorVersionCollectionClientAsync:
"""Async version of ActorVersionCollectionClient with identical methods."""Environment Variable Management
Management of Actor environment variables with support for secret values.
class ActorEnvVarClient:
def get(self) -> dict | None:
"""Get environment variable."""
def update(self, *, is_secret: bool | None, name: str, value: str) -> dict:
"""Update environment variable.
Args:
is_secret: Whether the value should be treated as secret
name: Variable name
value: Variable value
"""
def delete(self) -> None:
"""Delete environment variable."""
class ActorEnvVarClientAsync:
"""Async version of ActorEnvVarClient with identical methods."""
class ActorEnvVarCollectionClient:
def list(self) -> ListPage[dict]:
"""List Actor environment variables."""
def create(self, *, is_secret: bool | None, name: str, value: str) -> dict:
"""Create new environment variable.
Args:
is_secret: Whether the value should be treated as secret
name: Variable name
value: Variable value
"""
class ActorEnvVarCollectionClientAsync:
"""Async version of ActorEnvVarCollectionClient with identical methods."""Usage Examples
Basic Actor Execution
from apify_client import ApifyClient
client = ApifyClient('your-api-token')
# Get Actor client
actor = client.actor('john-doe/web-scraper')
# Start Actor and wait for completion
run = actor.call(run_input={
'startUrls': ['https://example.com'],
'maxPages': 10
})
print(f"Run finished with status: {run['status']}")Actor Management
# Create new Actor
new_actor = client.actors().create(
name='my-scraper',
title='My Web Scraper',
description='Custom web scraping solution'
)
# Update Actor configuration
actor = client.actor(new_actor['id'])
actor.update(
title='Updated Web Scraper',
description='Enhanced web scraping solution with new features'
)
# Build Actor
build = actor.build()
print(f"Build started: {build['id']}")Environment Variables
# Add environment variable to Actor version
version = client.actor('actor-id').version('1.0')
env_vars = version.env_vars()
env_vars.create(
name='API_KEY',
value='secret-api-key',
is_secret=True
)
# List all environment variables
vars_list = env_vars.list()
for var in vars_list.items:
print(f"{var['name']}: {'[SECRET]' if var['isSecret'] else var['value']}")Install with Tessl CLI
npx tessl i tessl/pypi-apify-client