Perago

https://perago.readthedocs.io/zh-cn/latest/
Admin
Perago is a typed Python runtime layer for Conductor workers that integrates task modules, Pydantic...
Tokens:95,602
Snippets:1,209
Trust Score:9.6
Update:4 months ago
Show doc for...
Tokens:
Raw
### perago start command example

Source: https://perago.readthedocs.io/zh-cn/latest/runtime/cli.html

Example of using the `perago start` command to launch a supervisor process that manages worker child processes.

```bash
PYTHONPATH=tests/fixtures perago start app.workers.features_build -j 2

```

--------------------------------

### LakeFSConfig Example

Source: https://perago.readthedocs.io/zh-cn/latest/_modules/perago/config.html

Example of how to instantiate LakeFSConfig.

```python
>>> LakeFSConfig(
...     endpoint_url="http://localhost:8000",
...     access_key_id="key",
...     secret_access_key="secret",
... )
LakeFSConfig(...)
```

--------------------------------

### ConductorConfig Example

Source: https://perago.readthedocs.io/zh-cn/latest/_modules/perago/config.html

Example of how to instantiate ConductorConfig.

```python
>>> ConductorConfig(server_url="http://localhost:8080/api")
ConductorConfig(...)
```

--------------------------------

### WorkerRuntime Example

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.WorkerRuntime.html

Example demonstrating the creation and usage of WorkerRuntime.

```python
>>> WorkerRuntime(
...     worker_id="featuresBuild0001",
...     log_file=Path("/tmp/perago/logs/worker.jsonl"),
...     swept_workspaces=[],
... ).worker_id
'featuresBuild0001'
```

--------------------------------

### WorkspaceSyncPlan Example

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.WorkspaceSyncPlan.html

Example demonstrating how to create a WorkspaceSyncPlan and access its attributes.

```python
>>> plan = WorkspaceSyncPlan(
...     upload_files=[WorkspaceUploadFile(Path("raw/input.wav"), "audio/render/raw/input.wav")],
...     delete_object_paths=["audio/render/old.tmp"],
... )
>>> plan.changed_object_count
2
```

--------------------------------

### Example Usage

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.workspace_download_files.html

Demonstrates how to use workspace_download_files to get download records for objects within a specified workspace prefix.

```python
>>> files = workspace_download_files(
...     workspace_dir,
...     WorkspaceSpec(prefix="/audio/render"),
...     ["audio/render/raw/input.wav", "other/input.wav"],
... )
>>> [file.local_path.relative_to(workspace_dir).as_posix() for file in files]
['raw/input.wav']

```

--------------------------------

### Example Usage

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.workspace_upload_files.html

Demonstrates how to use workspace_upload_files to get a list of files to upload and extract their object paths.

```python
>>> files = workspace_upload_files(workspace_dir, WorkspaceSpec(prefix="/audio/render"))
>>> [file.object_path for file in files]
['audio/render/raw/input.wav']
```

--------------------------------

### TaskControls Example

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.TaskControls.html

Example of how to instantiate TaskControls with retry, timeout, and limits.

```python
>>> TaskControls(
...     retry=RetryPolicy(count=4),
...     timeout=TimeoutPolicy(response_seconds=900),
...     limits=ExecutionLimits(concurrent_exec_limit=2),
... )
TaskControls(...)

```

--------------------------------

### TaskControls Example

Source: https://perago.readthedocs.io/zh-cn/latest/_modules/perago/models.html

An example demonstrating how to instantiate TaskControls with retry, timeout, and limits.

```python
>>> TaskControls(
...     retry=RetryPolicy(count=4),
...     timeout=TimeoutPolicy(response_seconds=900),
...     limits=ExecutionLimits(concurrent_exec_limit=2),
... )
TaskControls(...)
```

--------------------------------

### Log Retention Examples

Source: https://perago.readthedocs.io/zh-cn/latest/runtime/logging.html

Examples of valid configurations for PERAGO_LOG_RETENTION.

```text
例如 `7d` 或 `30d`

```

--------------------------------

### WorkspaceSpec Example

Source: https://perago.readthedocs.io/zh-cn/latest/_modules/perago/models.html

An example demonstrating how to instantiate WorkspaceSpec with a prefix and pre guardrails.

```python
>>> WorkspaceSpec(prefix="/audio/render", pre=[require_file("raw/manifest.json")])
WorkspaceSpec(...)
```

--------------------------------

### WorkspaceInput Example

Source: https://perago.readthedocs.io/zh-cn/latest/_modules/perago/models.html

Example of creating a WorkspaceInput object and using its published_output method.

```python
input_ref = WorkspaceInput(
    repository="song-000123",
    branch="main",
    ref_type="commit",
    ref="589f8770",
)
input_ref.published_output("9c6f8770").ref
```

--------------------------------

### Minimal Local Example

Source: https://perago.readthedocs.io/zh-cn/latest/runtime/configuration.html

Example .env file for local Perago configuration.

```dotenv
CONDUCTOR_SERVER_URL=http://localhost:8080/api

# Required only when starting workspace tasks.
# LAKECTL_SERVER_ENDPOINT_URL=http://localhost:8000
# LAKECTL_CREDENTIALS_ACCESS_KEY_ID=replace-with-real-access-key
# LAKECTL_CREDENTIALS_SECRET_ACCESS_KEY=replace-with-real-secret-key

PERAGO_WORKSPACE_ROOT=/var/tmp/perago/workspaces
PERAGO_LOG_ROOT=/var/tmp/perago/logs
PERAGO_LOG_FILE_MAX_SIZE=100MB
PERAGO_LOG_RETENTION=30d
PERAGO_WORKER_ID_PREFIX=peragoLocalWorker
PERAGO_FAILURE_REASON_MAX_LENGTH=500
PERAGO_WORKSPACE_GC_TTL=24h
PERAGO_WORKSPACE_GC_INTERVAL=1h
# Optional; unset by default.
PERAGO_SHUTDOWN_FORCE_KILL_AFTER=30s

```

--------------------------------

### Target parameter example

Source: https://perago.readthedocs.io/zh-cn/latest/getting-started/commands.html

Example showing the correct way to specify the target parameter as an import path.

```bash
perago check app.workers.features_build

```

--------------------------------

### Worker ID assignment example

Source: https://perago.readthedocs.io/zh-cn/latest/runtime/cli.html

Example of how `PERAGO_WORKER_ID` is assigned to each child process when using `perago start`.

```text
PERAGO_WORKER_ID=appworkersfeaturesbuild0001
PERAGO_WORKER_ID=appworkersfeaturesbuild0002

```

--------------------------------

### Explicit PERAGO_WORKER_ID_PREFIX configuration

Source: https://perago.readthedocs.io/zh-cn/latest/runtime/configuration.html

Example of setting PERAGO_WORKER_ID_PREFIX and the resulting PERAGO_WORKER_ID values when starting workers.

```bash
PERAGO_WORKER_ID_PREFIX=prodAFeaturesBuild

perago start -j 2 app.workers.features_build

# Results in:
# PERAGO_WORKER_ID=prodAFeaturesBuild0001
# PERAGO_WORKER_ID=prodAFeaturesBuild0002
```

--------------------------------

### build_workspace_sync_plan Example

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.build_workspace_sync_plan.html

This example demonstrates how to use the build_workspace_sync_plan function to create a synchronization plan for a workspace. It shows how to specify the workspace directory, workspace specification, and existing object paths, and then accesses the changed object count from the returned plan.

```python
>>> plan = build_workspace_sync_plan(
...     workspace_dir,
...     WorkspaceSpec(prefix="/audio/render"),
...     ["audio/render/old.tmp"],
... )
>>> plan.changed_object_count
1
```

--------------------------------

### Example Usage

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.build_workspace_task_output.html

Demonstrates how to use the build_workspace_task_output function with sample task definition, input workspace, published reference, and raw result.

```python
>>> task_def = load_module_task("app.workers.features_build")
>>> output = build_workspace_task_output(
...     task_def,
...     {"repository": "song-000123", "branch": "main", "ref_type": "commit", "ref": "..."},
...     "published-ref",
...     {"row_count": 100, "feature_count": 24},
... )
>>> sorted(output)
['result', 'workspace']

```

--------------------------------

### prepare_worker_runtime example

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.prepare_worker_runtime.html

Example demonstrating how to use the prepare_worker_runtime function to set up a worker's runtime environment, including configuration, module target, and environment variables, and then accessing the prepared worker ID.

```python
>>> from datetime import timedelta
>>> config = RuntimeConfig(
...     workspace_root=Path("/tmp/perago/workspaces"),
...     log_root=Path("/tmp/perago/logs"),
...     log_file_max_size=1048576,
...     log_retention=timedelta(days=1),
...     worker_id_prefix="featuresBuild",
... )
>>> runtime = prepare_worker_runtime(
...     config=config,
...     module_target="app.workers.features_build",
...     env={"PERAGO_WORKER_ID": "featuresBuild0001"},
... )
>>> runtime.worker_id
'featuresBuild0001'

```

--------------------------------

### Error message when Conductor TaskDef is missing

Source: https://perago.readthedocs.io/zh-cn/latest/runtime/cli.html

Example of an error message from `perago start` when the required Conductor TaskDef is not registered.

```text
error: Conductor TaskDef 'features.build' is not registered; run perago extract and register it before start

```

--------------------------------

### worker_child_specs Function

Source: https://perago.readthedocs.io/zh-cn/latest/_modules/perago/supervisor.html

Builds stable worker child specifications for a supervisor run, used for 'perago start -j'. Each child gets a one-based slot and a deterministic worker ID.

```python
def worker_child_specs(
    *, 
    base_env: dict[str, str],
    module_target: str,
    process_count: int,
) -> list[WorkerChildSpec]:
    """
    Build stable worker child specifications for a supervisor run.

    The supervisor uses these specs for ``perago start -j``. Each child receives
    a one-based slot and a deterministic ``PERAGO_WORKER_ID`` derived from the
    configured prefix or, when no prefix is configured, from ``module_target``.

    Parameters
    ----------
    base_env : dict of str to str
        Base environment copied into every child. ``PERAGO_WORKER_ID_PREFIX`` is
        honored when present.
    module_target : str
        Python import path of the single task module served by all child
        workers.
    process_count : int
        Number of worker child specs to create. Must be at least one.

    Returns
    """
    pass
```

--------------------------------

### Example Usage

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.build_workspace_free_task_output.html

Demonstrates how to use build_workspace_free_task_output with a loaded task definition and a raw result.

```python
>>> task_def = load_module_task("app.workers.metadata_validate")
>>> build_workspace_free_task_output(task_def, {"valid": False, "reason": "missing"})
{'result': {'valid': False, 'reason': 'missing'}}
```

--------------------------------

### Example Usage

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.run_workspace_free_task_attempt.html

Demonstrates how to load a task definition and run a workspace-free task attempt, then check its status.

```python
>>> task_def = load_module_task("app.workers.metadata_validate")
>>> result = run_workspace_free_task_attempt(
...     task_def,
...     {"params": {"song_id": "song-000123", "min_duration_seconds": 30}},
... )
>>> result.status
'COMPLETED'
```

--------------------------------

### Example .env file content

Source: https://perago.readthedocs.io/zh-cn/latest/_sources/reference/environment-variables.md.txt

This example demonstrates the format and content of a typical .env file for Perago.

```text
export PERAGO_LOG_FILE_MAX_SIZE=512KB
PERAGO_WORKSPACE_ROOT='/tmp/perago/workspaces'
PERAGO_LOG_ROOT="/tmp/perago/logs"
PERAGO_WORKER_ID_PREFIX=localWorker
PERAGO_EXECUTION_MODE=process
PERAGO_FAILURE_REASON_MAX_LENGTH=500
PERAGO_WORKSPACE_GC_TTL=24h
PERAGO_WORKSPACE_GC_INTERVAL=1h
PERAGO_SHUTDOWN_FORCE_KILL_AFTER=30s
```

--------------------------------

### RuntimeConfigError Example

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.RuntimeConfigError.html

Example of raising a RuntimeConfigError.

```python
>>> RuntimeConfigError("CONDUCTOR_SERVER_URL is required for perago start")
RuntimeConfigError('CONDUCTOR_SERVER_URL is required for perago start')
```

--------------------------------

### PostGuardrailViolation Example

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.PostGuardrailViolation.html

Example of instantiating PostGuardrailViolation.

```python
>>> PostGuardrailViolation("features/output.parquet is missing")
PostGuardrailViolation('features/output.parquet is missing')
```

--------------------------------

### GuardrailViolation Example

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.GuardrailViolation.html

Example of instantiating GuardrailViolation.

```python
>>> GuardrailViolation("required file is missing")
GuardrailViolation('required file is missing')
```

--------------------------------

### Fixture examples in this repository

Source: https://perago.readthedocs.io/zh-cn/latest/getting-started/commands.html

Commands for validating fixtures locally.

```bash
PYTHONPATH=tests/fixtures uv run perago check app.workers.features_build
PYTHONPATH=tests/fixtures uv run perago extract app.workers.features_build --output /tmp/features.build.json
PYTHONPATH=tests/fixtures uv run perago check app.workers.metadata_validate
PYTHONPATH=tests/fixtures uv run perago extract app.workers.metadata_validate --output /tmp/metadata.validate.json

```

--------------------------------

### TaskDefinitionError Example

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.TaskDefinitionError.html

Example of how to raise and catch a TaskDefinitionError.

```python
>>> TaskDefinitionError("task module must define exactly one task")
TaskDefinitionError('task module must define exactly one task')
```

--------------------------------

### TaskInputError Example

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.TaskInputError.html

Example of how to raise and catch a TaskInputError.

```python
>>> TaskInputError("workspace task input requires workspace and params")
TaskInputError('workspace task input requires workspace and params')
```

--------------------------------

### StaleAttemptError Example

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.StaleAttemptError.html

Example of how to instantiate and represent a StaleAttemptError.

```python
>>> StaleAttemptError("task-9b4c")
StaleAttemptError('task-9b4c')
```

--------------------------------

### Example Usage

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.run_workspace_task_attempt.html

Demonstrates how to run a workspace task attempt and check its status.

```python
>>> result.status
'COMPLETED'
```

--------------------------------

### Example Usage

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.require_file.html

Demonstrates how to use `require_file` within `WorkspaceSpec`.

```python
>>> WorkspaceSpec(prefix="/", pre=[require_file("input/data.csv")])
WorkspaceSpec(...)
```

--------------------------------

### ConductorConfig Example

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.ConductorConfig.html

Example of creating a ConductorConfig object.

```python
>>> ConductorConfig(server_url="http://localhost:8080/api")
ConductorConfig(...)

```

--------------------------------

### Example Usage

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.invoke_workspace_free_task.html

Demonstrates how to load a module task and invoke it using perago.invoke_workspace_free_task.

```python
>>> task_def = load_module_task("app.workers.metadata_validate")
>>> invoke_workspace_free_task(
...     task_def,
...     {"params": {"song_id": "song-000123", "min_duration_seconds": 30}},
... )
{'result': {'valid': True, 'reason': None}}
```

--------------------------------

### Example Usage

Source: https://perago.readthedocs.io/zh-cn/latest/api/generated/perago.TaskDefinition.html

Demonstrates how to load a task definition and access its name.

```python
>>> definition = load_module_task("app.workers.features_build")
>>> definition.name
'features.build'

```

--------------------------------

### Valid Path Examples

Source: https://perago.readthedocs.io/zh-cn/latest/getting-started/guardrails.html

Examples of valid path specifications for Guardrails.

```python
from pathlib import Path, PureWindowsPath

require_file("manifest.json")
require_file(Path("raw") / "manifest.json")
require_glob(Path("raw") / "**" / "*.parquet", min_count=1)
require_file(PureWindowsPath("raw") / "windows-authored.json")

```

--------------------------------

### Example .env file

Source: https://perago.readthedocs.io/zh-cn/latest/reference/environment-variables.html

Example of how to set environment variables in an .env file.

```dotenv
export PERAGO_LOG_FILE_MAX_SIZE=512KB
PERAGO_WORKSPACE_ROOT='/tmp/perago/workspaces'
PERAGO_LOG_ROOT="/tmp/perago/logs"
PERAGO_WORKER_ID_PREFIX=localWorker
PERAGO_EXECUTION_MODE=process
PERAGO_FAILURE_REASON_MAX_LENGTH=500
PERAGO_WORKSPACE_GC_TTL=24h
PERAGO_WORKSPACE_GC_INTERVAL=1h
PERAGO_SHUTDOWN_FORCE_KILL_AFTER=30s

```
Tokens:
Raw