Skip to content

Command Line Interface Commands

This page documents CLI commands available in orchestrator-core.

The syntax of a CLI command is:

python main.py <command> <sub_command>

Where:

  • <command> is one of the top-level headings in this page
  • <sub_command> is one of the secondary headings in this page

Some examples:

python main.py db migrate_tasks

python main.py generate workflows

python main.py scheduler show-schedule

Each command can also be run with --help to get information directly in the CLI.

Top level options:

--install-completion [bash|zsh|fish|powershell|pwsh]

Install completion for the specified shell. [default: None]

--show-completion [bash|zsh|fish|powershell|pwsh]

Show completion for the specified shell, to copy it or customize the installation. [default: None]

db

Interact with the application database. By default, does nothing, specify main.py db --help for more information.

downgrade

The downgrade command will downgrade the database to the previous revision or to the optionally specified revision.

By default the search index is left untouched. Pass --index to rebuild the search index as part of the migration. Indexing is opt-in because it can take a long time on large databases; on a production deployment it is usually better to run python main.py index all as a separate post-deployment step.

CLI Options
Arguments:
    [REVISION]  Rev id to downgrade to  [default: -1]

Options:
    --index / --no-index              Index the search tables after the migration (can be slow)  [default: no-index]
    --force-index / --no-force-index  Force re-index (ignore hash cache)  [default: no-force-index]
    --help                            Show this message and exit.
Source code in orchestrator-core/orchestrator/core/cli/database.py
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
@app.command()
def downgrade(
    revision: str = typer.Argument("-1", help="Rev id to downgrade to"),
    index: Index = False,
    force_index: ForceIndex = False,
) -> None:
    """The `downgrade` command will downgrade the database to the previous revision or to the optionally specified revision.

    By default the search index is left untouched. Pass `--index` to rebuild the search index as part of the
    migration. Indexing is opt-in because it can take a long time on large databases; on a production deployment it
    is usually better to run `python main.py index all` as a separate post-deployment step.

    Args:
        revision (str, optional): Optional argument to indicate where to downgrade to. [default: -1]
        index: Whether to index the search tables after the migration. Defaults to `False` (opt-in).
        force_index: Whether to re-index all fields regardless of the hash cache.

    Returns:
        None

    CLI Options:
        ```shell
        Arguments:
            [REVISION]  Rev id to downgrade to  [default: -1]

        Options:
            --index / --no-index              Index the search tables after the migration (can be slow)  [default: no-index]
            --force-index / --no-force-index  Force re-index (ignore hash cache)  [default: no-force-index]
            --help                            Show this message and exit.
        ```

    """
    command.downgrade(alembic_cfg(), revision)

    if index:
        _index_search(force_index=force_index)

heads

The heads command command shows the Alembic database heads.

CLI Options

None

Source code in orchestrator-core/orchestrator/core/cli/database.py
144
145
146
147
148
149
150
151
152
@app.command(help="Get the database heads")
def heads() -> None:
    """The `heads` command command shows the Alembic database heads.

    CLI Options:
        None

    """
    command.heads(alembic_cfg())

history

The history command lists Alembic revision history/changeset scripts in chronological order.

CLI Options
Options:
    -v, --verbose  Verbose output
    -c, --current  Indicate current revision
Source code in orchestrator-core/orchestrator/core/cli/database.py
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
@app.command()
def history(
    verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose output"),
    indicate_current: bool = typer.Option(True, "--current", "-c", help="Indicate current revision"),
) -> None:
    """The `history` command lists Alembic revision history/changeset scripts in chronological order.

    Args:
        verbose (bool, optional): Verbose output
        indicate_current (bool, optional): Indicate current revision

    Returns:
        None

    CLI Options:
        ```shell
        Options:
            -v, --verbose  Verbose output
            -c, --current  Indicate current revision
        ```
    """
    command.history(alembic_cfg(), verbose=verbose, indicate_current=indicate_current)

init

Initialize the migrations directory.

This command will initialize a migration directory for the orchestrator core application and setup a correct migration environment. It will also throw an exception when it detects conflicting files and directories.

CLI Options

None

Source code in orchestrator-core/orchestrator/core/cli/database.py
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
@app.command(
    help="Initialize an empty migrations environment. This command will throw an exception when it detects conflicting files and directories."
)
def init() -> None:
    """Initialize the `migrations` directory.

    This command will initialize a migration directory for the orchestrator core application and setup a correct
    migration environment. It will also throw an exception when it detects conflicting files and directories.

    Returns:
        None

    CLI Options:
        None

    """

    if os.access(migration_dir, os.F_OK) and os.listdir(migration_dir):
        raise OSError(f"Directory {migration_dir} already exists and is not empty")

    logger.info("Creating directory", directory=os.path.abspath(migration_dir))
    os.makedirs(migration_dir)
    versions = os.path.join(migration_dir, "versions")
    logger.info("Creating directory", directory=os.path.abspath(versions))
    os.makedirs(versions)
    versions_schema = os.path.join(migration_dir, "versions/schema")
    logger.info("Creating directory", directory=os.path.abspath(versions_schema))
    os.makedirs(versions_schema)

    source_env_py = os.path.join(orchestrator_module_location, f"{migration_dir}/templates/env.py.j2")
    env_py = os.path.join(migration_dir, "env.py")
    logger.info("Creating file", file=os.path.abspath(env_py))
    copyfile(source_env_py, env_py)

    source_script_py_mako = os.path.join(orchestrator_module_location, f"{migration_dir}/script.py.mako")
    script_py_mako = os.path.join(migration_dir, "script.py.mako")
    logger.info("Creating file", file=os.path.abspath(script_py_mako))
    copyfile(source_script_py_mako, script_py_mako)

    source_helpers_py = os.path.join(orchestrator_module_location, f"{migration_dir}/templates/helpers.py.j2")
    helpers_py = os.path.join(migration_dir, "helpers.py")
    logger.info("Creating file", file=os.path.abspath(helpers_py))
    copyfile(source_helpers_py, helpers_py)

    template = jinja_env.get_template("alembic.ini.j2")

    if not os.access(os.path.join(os.getcwd(), "alembic.ini"), os.F_OK):
        logger.info("Creating file", file=os.path.join(os.getcwd(), "alembic.ini"))
        with open(os.path.join(os.getcwd(), "alembic.ini"), "w") as alembic_ini:
            alembic_ini.write(template.render(migrations_dir=migration_dir))
    else:
        logger.info("Skipping Alembic.ini file. It already exists")

merge

Merge database revisions.

It is possible when using multiple git branches in your WFO development lifecycle to have multiple Alembic heads emerge. This command will allow you to merge those two (or more) heads to resolve the issue. You also might need to run this after updating your version of orchestrator-core if there have been schema changes.

Read More Here

CLI Options
Arguments:
    [REVISIONS]  Add the revision you would like to merge to this command.

Options:
    -m, --message TEXT  The revision message
Source code in orchestrator-core/orchestrator/core/cli/database.py
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
@app.command(help="Merge database revisions.")
def merge(
    revisions: list[str] = typer.Argument(help="Add the revisions you would like to merge to this command."),
    message: str | None = typer.Option(None, "--message", "-m", help="The revision message"),
) -> None:
    """Merge database revisions.

    It is possible when using multiple git branches in your WFO development lifecycle to have
    multiple Alembic heads emerge. This command will allow you to merge those two (or more)
    heads to resolve the issue. You also might need to run this after updating your version
    of orchestrator-core if there have been schema changes.

    [Read More Here](https://alembic.sqlalchemy.org/en/latest/branches.html#merging-branches)

    Args:
        revisions: List of revisions to merge
        message: Optional message for the revision.

    Returns:
        None

    CLI Options:
        ```shell
        Arguments:
            [REVISIONS]  Add the revision you would like to merge to this command.

        Options:
            -m, --message TEXT  The revision message
        ```
    """
    command.merge(alembic_cfg(), revisions, message=message)

migrate_workflows

The migrate-workflows command creates a migration file based on the difference between workflows in the database and registered WorkflowInstances in your codebase.

BACKUP YOUR DATABASE BEFORE USING THE MIGRATION!

You will be prompted with inputs for new models and resource type updates. Resource type updates are only handled when it’s renamed in all product blocks.

Returns None unless --test is used, in which case it returns: - tuple: - list of upgrade SQL statements in string format. - list of downgrade SQL statements in string format.

CLI Arguments
Arguments:
    MESSAGE  Migration name  [required]

Options:
    --test / --no-test  Optional boolean if you don't want to generate a migration
    file  [default: no-test]
Source code in orchestrator-core/orchestrator/core/cli/database.py
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
@app.command(help="Create migration file based on diff workflows in db")
def migrate_workflows(
    message: str = typer.Argument(..., help="Migration name"),
    test: bool = typer.Option(False, help="Optional boolean if you don't want to generate a migration file"),
) -> tuple[list[dict], list[dict]] | None:
    """The `migrate-workflows` command creates a migration file based on the difference between workflows in the database and registered WorkflowInstances in your codebase.

    !!! warning "BACKUP YOUR DATABASE BEFORE USING THE MIGRATION!"

    You will be prompted with inputs for new models and resource type updates.
    Resource type updates are only handled when it's renamed in all product blocks.

    Args:
        message: Message/description of the generated migration.
        test: Optional boolean if you don't want to generate a migration file.

    Returns None unless `--test` is used, in which case it returns:
        - tuple:
            - list of upgrade SQL statements in string format.
            - list of downgrade SQL statements in string format.

    CLI Arguments:
        ```shell
        Arguments:
            MESSAGE  Migration name  [required]

        Options:
            --test / --no-test  Optional boolean if you don't want to generate a migration
            file  [default: no-test]
        ```
    """
    if not app_settings.TESTING:
        init_database(app_settings)

    if test:
        print(  # noqa: T001, T201
            f"{str_fmt('NOTE:', flags=[COLOR.BOLD, COLOR.CYAN])} Running in test mode. No migration file will be generated.\n"
        )

    workflows_to_add, workflows_to_delete = create_workflows_migration_wizard()

    # String 'template' arguments
    import_str = "from orchestrator.core.migrations.helpers import create_workflow, delete_workflow\n"
    tpl_preamble_lines = []
    tpl_upgrade_lines = []
    tpl_downgrade_lines = []

    if workflows_to_add:
        tpl_preamble_lines.append(f"new_workflows = {json.dumps(workflows_to_add, indent=4)}\n")
        tpl_upgrade_lines.extend(
            [(" " * 4) + "for workflow in new_workflows:", (" " * 8) + "create_workflow(conn, workflow)"]
        )
        tpl_downgrade_lines.extend(
            [(" " * 4) + "for workflow in new_workflows:", (" " * 8) + 'delete_workflow(conn, workflow["name"])']
        )

    if workflows_to_delete:
        tpl_preamble_lines.append(f"old_workflows = {json.dumps(workflows_to_delete, indent=4)}\n")
        tpl_upgrade_lines.extend(
            [(" " * 4) + "for workflow in old_workflows:", (" " * 8) + 'delete_workflow(conn, workflow["name"])']
        )
        tpl_downgrade_lines.extend(
            [(" " * 4) + "for workflow in old_workflows:", (" " * 8) + "create_workflow(conn, workflow)"]
        )

    preamble = "\n".join(
        [
            import_str,
            *tpl_preamble_lines,
        ]
    )
    sql_upgrade_str = "\n".join(tpl_upgrade_lines)
    sql_downgrade_str = "\n".join(tpl_downgrade_lines)

    if test:
        return workflows_to_add, workflows_to_delete

    create_migration_file(alembic_cfg(), sql_upgrade_str, sql_downgrade_str, message, preamble=preamble)
    return None

migrate_tasks

The migrate-tasks command creates a migration file based on the difference between tasks in the database and registered TaskInstances in your codebase.

BACKUP YOUR DATABASE BEFORE USING THE MIGRATION!

You will be prompted with inputs for new models and resource type updates. Resource type updates are only handled when it’s renamed in all product blocks.

Returns None unless --test is used, in which case it returns: - tuple: - list of upgrade SQL statements in string format. - list of downgrade SQL statements in string format.

CLI Arguments
Arguments:
    MESSAGE  Migration name  [required]

Options:
    --test / --no-test  Optional boolean if you don't want to generate a migration
    file  [default: no-test]
Source code in orchestrator-core/orchestrator/core/cli/database.py
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
@app.command(help="Create migration file based on diff tasks in db")
def migrate_tasks(
    message: str = typer.Argument(..., help="Migration name"),
    test: bool = typer.Option(False, help="Optional boolean if you don't want to generate a migration file"),
) -> tuple[list[dict], list[dict]] | None:
    """The `migrate-tasks` command creates a migration file based on the difference between tasks in the database and registered TaskInstances in your codebase.

    !!! warning "BACKUP YOUR DATABASE BEFORE USING THE MIGRATION!"

    You will be prompted with inputs for new models and resource type updates.
    Resource type updates are only handled when it's renamed in all product blocks.

    Args:
        message: Message/description of the generated migration.
        test: Optional boolean if you don't want to generate a migration file.

    Returns None unless `--test` is used, in which case it returns:
        - tuple:
            - list of upgrade SQL statements in string format.
            - list of downgrade SQL statements in string format.

    CLI Arguments:
        ```shell
        Arguments:
            MESSAGE  Migration name  [required]

        Options:
            --test / --no-test  Optional boolean if you don't want to generate a migration
            file  [default: no-test]
        ```
    """
    if not app_settings.TESTING:
        init_database(app_settings)

    if test:
        print(  # noqa: T001, T201
            f"{str_fmt('NOTE:', flags=[COLOR.BOLD, COLOR.CYAN])} Running in test mode. No migration file will be generated.\n"
        )

    tasks_to_add, tasks_to_delete = create_tasks_migration_wizard()

    # String 'template' arguments
    import_str = "from orchestrator.core.migrations.helpers import create_task, delete_workflow\n"
    tpl_preamble_lines = []
    tpl_upgrade_lines = []
    tpl_downgrade_lines = []

    if tasks_to_add:
        tpl_preamble_lines.append(f"new_tasks = {json.dumps(tasks_to_add, indent=4)}\n")
        tpl_upgrade_lines.extend([(" " * 4) + "for task in new_tasks:", (" " * 8) + "create_task(conn, task)"])
        tpl_downgrade_lines.extend(
            [(" " * 4) + "for task in new_tasks:", (" " * 8) + 'delete_workflow(conn, task["name"])']
        )

    if tasks_to_delete:
        tpl_preamble_lines.append(f"old_tasks = {json.dumps(tasks_to_delete, indent=4)}\n")
        tpl_upgrade_lines.extend(
            [(" " * 4) + "for task in old_tasks:", (" " * 8) + 'delete_workflow(conn, task["name"])']
        )
        tpl_downgrade_lines.extend([(" " * 4) + "for task in old_tasks:", (" " * 8) + "create_task(conn, task)"])

    preamble = "\n".join(
        [
            import_str,
            *tpl_preamble_lines,
        ]
    )
    sql_upgrade_str = "\n".join(tpl_upgrade_lines)
    sql_downgrade_str = "\n".join(tpl_downgrade_lines)

    if test:
        return tasks_to_add, tasks_to_delete

    create_migration_file(alembic_cfg(), sql_upgrade_str, sql_downgrade_str, message, preamble=preamble)
    return None

revision

The revision command creates a new Alembic revision file.

CLI Options
Options:
    -m, --message TEXT              The revision message
    --version-path TEXT             Specify specific path from config for version file
    --autogenerate / --no-autogenerate
                                    Detect schema changes and add migrations [default: no-autogenerate]
    --head TEXT                     Determine the head you need to add your migration to.
Source code in orchestrator-core/orchestrator/core/cli/database.py
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
@app.command()
def revision(
    message: str | None = typer.Option(None, "--message", "-m", help="The revision message"),
    version_path: str | None = typer.Option(None, "--version-path", help="Specify specific path from config for version file"),
    autogenerate: bool = typer.Option(False, help="Detect schema changes and add migrations"),
    head: str = typer.Option("data@head", help="Determine the head you need to add your migration to."),
) -> None:
    """The `revision` command creates a new Alembic revision file.

    Args:
        message: The revision message
        version_path: Specify specific path from config for version file
        autogenerate: Whether to detect schema changes.
        head: To which head the migration applies

    Returns:
        None

    CLI Options:
        ```shell
        Options:
            -m, --message TEXT              The revision message
            --version-path TEXT             Specify specific path from config for version file
            --autogenerate / --no-autogenerate
                                            Detect schema changes and add migrations [default: no-autogenerate]
            --head TEXT                     Determine the head you need to add your migration to.
        ```
    """
    create_data_head_if_not_exists({"writer": create_writer(), "environment": get_template_environment()})
    command.revision(alembic_cfg(), message, version_path=version_path, autogenerate=autogenerate, head=head)

upgrade

The upgrade command will upgrade the database to the specified revision.

By default the search index is left untouched. Pass --index to rebuild the search index as part of the migration. Indexing is opt-in because it can take a long time on large databases; on a production deployment it is usually better to run python main.py index all as a separate post-deployment step.

CLI Options
Arguments:
    [REVISION]  Rev id to upgrade to

Options:
    --index / --no-index              Index the search tables after the migration (can be slow)  [default: no-index]
    --force-index / --no-force-index  Force re-index (ignore hash cache)  [default: no-force-index]
    --help                            Show this message and exit.
Source code in orchestrator-core/orchestrator/core/cli/database.py
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
@app.command()
def upgrade(
    revision: str = typer.Argument(help="Rev id to upgrade to"),
    index: Index = False,
    force_index: ForceIndex = False,
) -> None:
    """The `upgrade` command will upgrade the database to the specified revision.

    By default the search index is left untouched. Pass `--index` to rebuild the search index as part of the
    migration. Indexing is opt-in because it can take a long time on large databases; on a production deployment it
    is usually better to run `python main.py index all` as a separate post-deployment step.

    Args:
        revision: Optional argument to indicate where to upgrade to.
        index: Whether to index the search tables after the migration. Defaults to `False` (opt-in).
        force_index: Whether to re-index all fields regardless of the hash cache.

    Returns:
        None

    CLI Options:
        ```shell
        Arguments:
            [REVISION]  Rev id to upgrade to

        Options:
            --index / --no-index              Index the search tables after the migration (can be slow)  [default: no-index]
            --force-index / --no-force-index  Force re-index (ignore hash cache)  [default: no-force-index]
            --help                            Show this message and exit.
        ```

    """
    command.upgrade(alembic_cfg(), revision)

    if index:
        _index_search(force_index=force_index)

migrate-domain-models

The migrate-domain-models CLI command is used to automatically generate the data migrations that you’ll need when you add or change a Domain Model. It will inspect your DB and the existing domain models, analyse the differences and it will generate an Alembic data migration in the correct folder.

Features:

  • detect a new Domain Model attribute / resource type
  • detect a renamed Domain Model attribute / resource type
  • detect a removed Domain Model attribute / resource type
  • detect a new Domain Model
  • detect a removed Domain Model
  • ability to ask for human input when needed

Below in the documentation these features are discussed in more detail.

BACKUP DATABASE BEFORE USING THE MIGRATION!

Arguments

  • message: Message/description of the generated migration.
  • --test: Optional boolean if you don’t want to generate a migration file.
  • --inputs: stringified dict to prefill inputs. The inputs and updates argument is mostly used for testing, prefilling the given inputs, here examples:
    • new product: inputs = { "new_product_name": { "description": "add description", "product_type": "add_type", "tag": "add_tag" }}
    • new product fixed input: inputs = { "new_product_name": { "new_fixed_input_name": "value" }}
    • new product block: inputs = { "new_product_block_name": { "description": "add description", "tag": "add_tag" } }
    • new resource type: inputs = { "new_resource_type_name": { "description": "add description", "value": "add default value", "new_product_block_name": "add default value for block" }}
      • new_product_block_name prop inserts value specifically for that block.
      • value prop is inserted as default for all existing instances it is added to.
  • --updates: stringified dict to prefill inputs.
    • renaming a fixed input:
      • updates = { "fixed_inputs": { "product_name": { "old_fixed_input_name": "new_fixed_input_name" } } }
    • renaming a resource type to a new resource type:
      • inputs = { "new_resource_type_name": { "description": "add description" }}
      • updates = { "resource_types": { "old_resource_type_name": "new_resource_type_name" } }
    • renaming a resource type to existing resource type: updates = { "resource_types": { "old_resource_type_name": "new_resource_type_name" } }

Example

You need products in the SUBSCRIPTION_MODEL_REGISTRY, for this example I will use these models (taken out of example-orchestrator):

UserGroup Block
from orchestrator.core.domain.base import SubscriptionModel, ProductBlockModel
from orchestrator.core.types import SubscriptionLifecycle


class UserGroupBlockInactive(
    ProductBlockModel,
    lifecycle=[SubscriptionLifecycle.INITIAL],
    product_block_name="UserGroupBlock",
):
    group_name: str | None = None
    group_id: int | None = None


class UserGroupBlockProvisioning(
    UserGroupBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
):
    group_name: str
    group_id: int | None = None


class UserGroupBlock(
    UserGroupBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]
):
    group_name: str
    group_id: int
from orchestrator.domain.base import SubscriptionModel, ProductBlockModel
from orchestrator.types import SubscriptionLifecycle


class UserGroupBlockInactive(
    ProductBlockModel,
    lifecycle=[SubscriptionLifecycle.INITIAL],
    product_block_name="UserGroupBlock",
):
    group_name: str | None = None
    group_id: int | None = None


class UserGroupBlockProvisioning(
    UserGroupBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
):
    group_name: str
    group_id: int | None = None


class UserGroupBlock(
    UserGroupBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]
):
    group_name: str
    group_id: int
UserGroup Product
from orchestrator.core.domain.base import SubscriptionModel
from orchestrator.core.types import SubscriptionLifecycle

from products.product_blocks.user_group import (
    UserGroupBlock,
    UserGroupBlockInactive,
    UserGroupBlockProvisioning,
)


class UserGroupInactive(
    SubscriptionModel, is_base=True, lifecycle=[SubscriptionLifecycle.INITIAL]
):
    settings: UserGroupBlockInactive


class UserGroupProvisioning(
    UserGroupInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
):
    settings: UserGroupBlockProvisioning


class UserGroup(UserGroupProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
    settings: UserGroupBlock
from orchestrator.domain.base import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle

from products.product_blocks.user_group import (
    UserGroupBlock,
    UserGroupBlockInactive,
    UserGroupBlockProvisioning,
)


class UserGroupInactive(
    SubscriptionModel, is_base=True, lifecycle=[SubscriptionLifecycle.INITIAL]
):
    settings: UserGroupBlockInactive


class UserGroupProvisioning(
    UserGroupInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
):
    settings: UserGroupBlockProvisioning


class UserGroup(UserGroupProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
    settings: UserGroupBlock
User Block
from orchestrator.core.domain.base import ProductBlockModel
from orchestrator.core.types import SubscriptionLifecycle

from products.product_blocks.user_group import (
    UserGroupBlock,
    UserGroupBlockInactive,
    UserGroupBlockProvisioning,
)


class UserBlockInactive(
    ProductBlockModel,
    lifecycle=[SubscriptionLifecycle.INITIAL],
    product_block_name="UserBlock",
):
    group: UserGroupBlockInactive
    username: str | None = None
    age: int | None = None
    user_id: int | None = None


class UserBlockProvisioning(
    UserBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
):
    group: UserGroupBlockProvisioning
    username: str
    age: int | None = None
    user_id: int | None = None


class UserBlock(UserBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
    group: UserGroupBlock
    username: str
    age: int | None = None
    user_id: int
from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle

from products.product_blocks.user_group import (
    UserGroupBlock,
    UserGroupBlockInactive,
    UserGroupBlockProvisioning,
)


class UserBlockInactive(
    ProductBlockModel,
    lifecycle=[SubscriptionLifecycle.INITIAL],
    product_block_name="UserBlock",
):
    group: UserGroupBlockInactive
    username: str | None = None
    age: int | None = None
    user_id: int | None = None


class UserBlockProvisioning(
    UserBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
):
    group: UserGroupBlockProvisioning
    username: str
    age: int | None = None
    user_id: int | None = None


class UserBlock(UserBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
    group: UserGroupBlock
    username: str
    age: int | None = None
    user_id: int
User Product
from orchestrator.core.domain.base import SubscriptionModel
from orchestrator.core.types import SubscriptionLifecycle, strEnum

from products.product_blocks.user import (
    UserBlock,
    UserBlockInactive,
    UserBlockProvisioning,
)


class Affiliation(strEnum):
    internal = "internal"
    external = "external"


class UserInactive(SubscriptionModel, is_base=True):
    affiliation: Affiliation
    settings: UserBlockInactive


class UserProvisioning(UserInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
    affiliation: Affiliation
    settings: UserBlockProvisioning


class User(UserProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
    affiliation: Affiliation
    settings: UserBlock
from orchestrator.domain.base import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, strEnum

from products.product_blocks.user import (
    UserBlock,
    UserBlockInactive,
    UserBlockProvisioning,
)


class Affiliation(strEnum):
    internal = "internal"
    external = "external"


class UserInactive(SubscriptionModel, is_base=True):
    affiliation: Affiliation
    settings: UserBlockInactive


class UserProvisioning(UserInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
    affiliation: Affiliation
    settings: UserBlockProvisioning


class User(UserProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
    affiliation: Affiliation
    settings: UserBlock
SUBSCRIPTION_MODEL_REGISTRY
from orchestrator.core.domain import SUBSCRIPTION_MODEL_REGISTRY

from products.product_types.user import User
from products.product_types.user_group import UserGroup

# Register models to actual definitions for deserialization purposes
SUBSCRIPTION_MODEL_REGISTRY.update(
    {
        "User group": UserGroup,
        "User internal": User,
        "User external": User,
    }
)
from orchestrator.domain import SUBSCRIPTION_MODEL_REGISTRY

from products.product_types.user import User
from products.product_types.user_group import UserGroup

# Register models to actual definitions for deserialization purposes
SUBSCRIPTION_MODEL_REGISTRY.update(
    {
        "User group": UserGroup,
        "User internal": User,
        "User external": User,
    }
)

Running the command

  • only with a message

    python main.py db migrate-domain-models "message"
    

  • Running it as test

    python main.py db migrate-domain-models "message" --test
    

  • Running the command with inputs prefilled

    python main.py db migrate-domain-models "message" --inputs "{ "" }"
    

The command will first go through all products and map the differences with the database. debug log example:

2022-10-27 11:45:10 [debug] ProductTable blocks diff [orchestrator.core.domain.base] fixed_inputs_in_db=set() fixed_inputs_model=set() missing_fixed_inputs_in_db=set() missing_fixed_inputs_in_model=set() missing_product_blocks_in_db=set() missing_product_blocks_in_model=set() product_block_db=User group product_blocks_in_db={'UserGroupBlock'} product_blocks_in_model={'UserGroupBlock'}

You will be prompted with inputs when updates are found.

  • rename of resource type input (renaming age to user_age in User Block). Only works when the resource type is renamed in all Blocks:

    Update resource types
    Do you wish to rename resource type age to user_age? [y/N]:

  • rename of fixed input (renaming affiliation to affiliationing in User Product):

    Update fixed inputs
    Do you wish to rename fixed input affiliation to affiliationing for product User internal? [y/N]:

  • update of resource type per block (renaming age to user_age in User Block and not choosing to rename resource type). The input will loop until skipped or when there are no options anymore:

    • first you get to choose which old resource type to update, skip will create/delete all resource types. > Update block resource types
      Which resource type would you want to update in UserBlock Block?
      1) age
      q) skip
      ?
      
    • then you get to choose which new resource type to update with, skip will give you the first question again.
      Which resource type should update age?
      1) user_age
      q) skip
      ?
      
    • with 1 and 1, the log level difference would look like:
      2023-02-08 14:11:25 [info] update_block_resource_types [orchestrator.core.cli.migrate_domain_models] update_block_resource_types={'UserBlock': {'age': 'user_age'}}
      

It will log the differences on info level:

2022-10-27 11:45:10 [info] create_products                   [orchestrator.core.cli.migrate_domain_models] create_products={'User group': <class 'products.product_types.user_group.UserGroup'>, 'User internal': <class 'products.product_types.user.User'>, 'User external': <class 'products.product_types.user.User'>}
2022-10-27 11:45:10 [info] delete_products                   [orchestrator.core.cli.migrate_domain_models] delete_products=set()
2022-10-27 11:45:10 [info] create_product_fixed_inputs       [orchestrator.core.cli.migrate_domain_models] create_product_fixed_inputs={'affiliation': {'User external', 'User internal'}}
2022-10-27 11:45:10 [info] update_product_fixed_inputs       [orchestrator.core.cli.migrate_domain_models] update_product_fixed_inputs={}
2022-10-27 11:45:10 [info] delete_product_fixed_inputs       [orchestrator.core.cli.migrate_domain_models] delete_product_fixed_inputs={}
2022-10-27 11:45:10 [info] create_product_to_block_relations [orchestrator.core.cli.migrate_domain_models] create_product_to_block_relations={'UserGroupBlock': {'User group'}, 'UserBlock': {'User external', 'User internal'}}
2022-10-27 11:45:10 [info] delete_product_to_block_relations [orchestrator.core.cli.migrate_domain_models] delete_product_to_block_relations={}
2022-10-27 11:45:10 [info] create_resource_types             [orchestrator.core.cli.migrate_domain_models] create_resource_types={'username', 'age', 'group_name', 'user_id', 'group_id'}
2022-10-27 11:45:10 [info] rename_resource_types             [orchestrator.core.cli.migrate_domain_models] rename_resource_types={}
2022-10-27 11:45:10 [info] delete_resource_types             [orchestrator.core.cli.migrate_domain_models] delete_resource_types=set()
2022-10-27 11:45:10 [info] create_resource_type_relations    [orchestrator.core.cli.migrate_domain_models] create_resource_type_relations={'group_name': {'UserGroupBlock'}, 'group_id': {'UserGroupBlock'}, 'username': {'UserBlock'}, 'age': {'UserBlock'}, 'user_id': {'UserBlock'}}
2022-10-27 11:45:10 [info] delete_resource_type_relations    [orchestrator.core.cli.migrate_domain_models] delete_resource_type_relations={}
2022-10-27 11:45:10 [info] create_product_blocks             [orchestrator.core.cli.migrate_domain_models] create_blocks={'UserGroupBlock': <class 'products.product_blocks.user_group.UserGroupBlock'>, 'UserBlock': <class 'products.product_blocks.user.UserBlock'>}
2022-10-27 11:45:10 [info] delete_product_blocks             [orchestrator.core.cli.migrate_domain_models] delete_blocks=set()
2022-10-27 11:45:10 [info] create_product_block_relations    [orchestrator.core.cli.migrate_domain_models] create_product_block_relations={'UserGroupBlock': {'UserBlock'}}
2022-10-27 11:45:10 [info] delete_product_block_relations    [orchestrator.core.cli.migrate_domain_models] delete_product_block_relations={}

You will be asked to confirm the actions in order to continue:

WARNING: Deleting products will also delete its subscriptions.
Confirm the above actions [y/N]:

After confirming, it will start generating the SQL, logging the SQL on debug level and prompt the user for new resources:

  • new product example:

    Create new products
    Product: UserGroup User group
    Supply the production description: User group product
    Supply the product tag: GROUP

    2022-10-27 11:45:10 [debug] generated SQL [orchestrator.core.cli.domain_gen_helpers.helpers] sql_string=INSERT INTO products (name, description, product_type, tag, status) VALUES ('User group', 'User group product', 'UserGroup', 'GROUP', 'active') RETURNING products.product_id
    

  • new fixed input (the type isn’t checked, so typing an incorrect value will insert in db):

    Create fixed inputs
    Supply fixed input value for product User internal and fixed input affiliation: internal
    Supply fixed input value for product User external and fixed input affiliation: external

    2022-10-27 11:45:10 [debug] generated SQL [orchestrator.core.cli.domain_gen_helpers.helpers] sql_string=INSERT INTO fixed_inputs (name, value, product_id) VALUES ('affiliation', 'internal', (SELECT products.product_id FROM products WHERE products.name IN ('User internal'))), ('affiliation', 'external', (SELECT products.product_id FROM products WHERE products.name IN ('User external')))
    

  • new product block:

    Create product blocks
    Product block: UserGroupBlock
    Supply the product block description: User group settings
    Supply the product block tag: UGS

    2022-10-27 11:45:10 [debug] generated SQL [orchestrator.core.cli.domain_gen_helpers.helpers] sql_string=`#!sql INSERT INTO product_blocks (name, description, tag, status) VALUES ('UserGroupBlock', 'User group settings', 'UGS', 'active') RETURNING product_blocks.product_block_id`
    

  • new resource type:

    Create resource types
    Supply description for new resource type group_name: Unique name of user group

    2022-10-27 11:45:10 [debug] generated SQL [orchestrator.core.cli.domain_gen_helpers.helpers] sql_string=INSERT INTO resource_types (resource_type, description) VALUES ('group_name', 'Unique name of user group') RETURNING resource_types.resource_type_id
    

  • default value for resource type per product block (necessary for adding a default value to existing instances):

    Create subscription instance values
    Supply default subscription instance value for resource type group_name and product block UserGroupBlock: group

    2022-10-27 11:45:10 [debug] generated SQL [orchestrator.core.cli.domain_gen_helpers.resource_type_helpers] sql_string=
                    WITH subscription_instance_ids AS (
                        SELECT subscription_instances.subscription_instance_id
                        FROM   subscription_instances
                        WHERE  subscription_instances.product_block_id IN (
                            SELECT product_blocks.product_block_id
                            FROM   product_blocks
                            WHERE  product_blocks.name = 'UserGroupBlock'
                        )
                    )
    
                    INSERT INTO
                        subscription_instance_values (subscription_instance_id, resource_type_id, value)
                    SELECT
                        subscription_instance_ids.subscription_instance_id,
                        resource_types.resource_type_id,
                        'group'
                    FROM resource_types
                    CROSS JOIN subscription_instance_ids
                    WHERE resource_types.resource_type = 'group_name'
    

Last part generates the migration with the generated SQL:

Generating migration file
2022-10-27 11:45:10 [info] Version Locations [orchestrator.core.cli.database] locations=/home/tjeerddie/projects_surf/example-orchestrator/migrations/versions/schema /home/tjeerddie/projects_surf/example-orchestrator/.venv/lib/python3.10/site-packages/orchestrator/migrations/versions/schema
  Generating /home/tjeerddie/projects_surf/example-orchestrator/migrations/versions/schema/2022-10-27_a8946b2d1647_test.py ...  done
Migration generated. Don't forget to create a database backup before migrating!

If you are running with --test, the SQL file will not be generated.

generate

Generate products, workflows and other artifacts.

Products can be described in a YAML configuration file which makes it easy to generate product and product block domain models, and skeleton workflows and unit tests. Note that this is a one time thing, the generate commands do not support updating existing products, product-blocks, workflows and migrations, in this case have a look at the db migrate-domain-models and db migrate-workflows commands. But it does however help in defining new products with stakeholders, will generate code that conforms to current workfloworchestrator coding BCP, and will actually run (although limited in functionality of course).

After describing a new product in a configuration file, the following commands are typically run:

python main.py generate product-blocks
python main.py generate products
python main.py generate workflows
python main.py generate migration

The generate command should be called from the top level folder of your orchestrator implementation, this is the folder that contains the products sub folder, among others, except when the --prefix is used to point to that folder. In case there are product blocks defined that use other generated product blocks, the order in which generate product-blocks is run is important, the code for the blocks used in other blocks should be generated first.

config file

See the Generate Config File guide for full documentation on the YAML product configuration format used by the generate commands.

migration

The python main.py generate migration command creates a migration from a configuration file.

Options

–config-file - The configuration file [default: None] –python-version - Python version for generated code [default: 3.11] –skip-existing-blocks - If set, the migration will not contain product blocks for which a python implementation exists [default: False]

product

The python main.py generate product command creates a product domain model from a configuration file.

Options

–config-file - The configuration file [default: None] –dryrun | –no-dryrun - Dry run [default: dryrun] –force - Force overwrite of existing files –python-version - Python version for generated code [default: 3.11] –folder-prefix - Folder prefix, e.g. /workflows [default: None]

product-blocks

The python main.py generate product-blocks command creates product block domain models from a configuration file.

Options

–config-file - The configuration file [default: None] –dryrun | –no-dryrun - Dry run [default: dryrun] –force - Force overwrite of existing files –python-version - Python version for generated code [default: 3.11] –folder-prefix - Folder prefix, e.g. /workflows [default: None]

unit-tests

The python main.py generate unit-tests command creates unit tests from a configuration file.

Options

–config-file - The configuration file [default: None] –dryrun | –no-dryrun - Dry run [default: dryrun] –force - Force overwrite of existing files –python-version - Python version for generated code [default: 3.11] –tdd - Force test driven development with failing asserts [default: True]

workflows

The python main.py generate workflows command creates create, modify, terminate and validate workflows from a configuration file. The --custom-templates option can be used to specify a folder with custom templates to add additional import statements, input form fields and workflow steps to the create, modify and terminate workflows.

Options

–config-file - The configuration file [default: None] –dryrun | –no-dryrun - Dry run [default: dryrun] –force - Force overwrite of existing files –python-version - Python version for generated code [default: 3.11] –folder-prefix - Folder prefix, e.g. /workflows [default: None] –custom-templates - Custom templates folder [default: None]

Note

The workflows/__init__.py will only be extended with the needed LazyWorkflowInstance declarations when --force is used.

index

(Re-)index the search tables used by AI / Hybrid Search. Run the commands below for the initial build of the index and after bulk changes to subscriptions, products, processes, workflows, product blocks or resource types.

Indexing can take a long time

Indexing a large database can take a long time (up to hours on production deployments). The db upgrade and db downgrade commands therefore do not update the search index by default. Run the indexing commands below as a separate post-deployment step, or pass --index to the migration command to opt in.

The syntax of an index command is:

python main.py index <sub_command>

Some examples:

python main.py index all

python main.py index subscriptions --subscription-id <uuid>

python main.py index rebuild-paths

subscriptions

Index the subscription search index.

CLI Options
Options:
    --subscription-id <str>               UUID (default = all)
    --dry-run / --no-dry-run              No DB writes  [default: no-dry-run]
    --force-index / --no-force-index      Force re-index (ignore hash cache)  [default: no-force-index]
    --show-progress / --no-show-progress  Show per-entity progress  [default: no-show-progress]
    --help                                Show this message and exit.
Source code in orchestrator-core/orchestrator/core/cli/search/index_llm.py
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@app.command("subscriptions")
def subscriptions_command(
    subscription_id: EntityId = None,
    dry_run: DryRun = False,
    force_index: ForceIndex = False,
    show_progress: ShowProgress = False,
) -> None:
    """Index the subscription search index.

    Args:
        subscription_id: UUID of a single subscription to index, `None` indexes all.
        dry_run: Make no database writes and no embedding calls.
        force_index: Re-index every field, ignoring the content hashes.
        show_progress: Show per-entity progress.

    CLI Options:
        ```shell
        Options:
            --subscription-id <str>               UUID (default = all)
            --dry-run / --no-dry-run              No DB writes  [default: no-dry-run]
            --force-index / --no-force-index      Force re-index (ignore hash cache)  [default: no-force-index]
            --show-progress / --no-show-progress  Show per-entity progress  [default: no-show-progress]
            --help                                Show this message and exit.
        ```
    """
    run_indexing_for_entity(
        entity_kind=EntityType.SUBSCRIPTION,
        entity_id=subscription_id,
        dry_run=dry_run,
        force_index=force_index,
        show_progress=show_progress,
    )

products

Index the product search index.

CLI Options
Options:
    --product-id <str>                    UUID (default = all)
    --dry-run / --no-dry-run              No DB writes  [default: no-dry-run]
    --force-index / --no-force-index      Force re-index (ignore hash cache)  [default: no-force-index]
    --show-progress / --no-show-progress  Show per-entity progress  [default: no-show-progress]
    --help                                Show this message and exit.
Source code in orchestrator-core/orchestrator/core/cli/search/index_llm.py
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@app.command("products")
def products_command(
    product_id: EntityId = None,
    dry_run: DryRun = False,
    force_index: ForceIndex = False,
    show_progress: ShowProgress = False,
) -> None:
    """Index the product search index.

    Args:
        product_id: UUID of a single product to index, `None` indexes all.
        dry_run: Make no database writes and no embedding calls.
        force_index: Re-index every field, ignoring the content hashes.
        show_progress: Show per-entity progress.

    CLI Options:
        ```shell
        Options:
            --product-id <str>                    UUID (default = all)
            --dry-run / --no-dry-run              No DB writes  [default: no-dry-run]
            --force-index / --no-force-index      Force re-index (ignore hash cache)  [default: no-force-index]
            --show-progress / --no-show-progress  Show per-entity progress  [default: no-show-progress]
            --help                                Show this message and exit.
        ```
    """
    run_indexing_for_entity(
        entity_kind=EntityType.PRODUCT,
        entity_id=product_id,
        dry_run=dry_run,
        force_index=force_index,
        show_progress=show_progress,
    )

processes

Index the process search index.

CLI Options
Options:
    --process-id <str>                    UUID (default = all)
    --dry-run / --no-dry-run              No DB writes  [default: no-dry-run]
    --force-index / --no-force-index      Force re-index (ignore hash cache)  [default: no-force-index]
    --show-progress / --no-show-progress  Show per-entity progress  [default: no-show-progress]
    --help                                Show this message and exit.
Source code in orchestrator-core/orchestrator/core/cli/search/index_llm.py
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
@app.command("processes")
def processes_command(
    process_id: EntityId = None,
    dry_run: DryRun = False,
    force_index: ForceIndex = False,
    show_progress: ShowProgress = False,
) -> None:
    """Index the process search index.

    Args:
        process_id: UUID of a single process to index, `None` indexes all.
        dry_run: Make no database writes and no embedding calls.
        force_index: Re-index every field, ignoring the content hashes.
        show_progress: Show per-entity progress.

    CLI Options:
        ```shell
        Options:
            --process-id <str>                    UUID (default = all)
            --dry-run / --no-dry-run              No DB writes  [default: no-dry-run]
            --force-index / --no-force-index      Force re-index (ignore hash cache)  [default: no-force-index]
            --show-progress / --no-show-progress  Show per-entity progress  [default: no-show-progress]
            --help                                Show this message and exit.
        ```
    """
    run_indexing_for_entity(
        entity_kind=EntityType.PROCESS,
        entity_id=process_id,
        dry_run=dry_run,
        force_index=force_index,
        show_progress=show_progress,
    )

workflows

Index the workflow search index.

CLI Options
Options:
    --workflow-id <str>                   UUID (default = all)
    --dry-run / --no-dry-run              No DB writes  [default: no-dry-run]
    --force-index / --no-force-index      Force re-index (ignore hash cache)  [default: no-force-index]
    --show-progress / --no-show-progress  Show per-entity progress  [default: no-show-progress]
    --help                                Show this message and exit.
Source code in orchestrator-core/orchestrator/core/cli/search/index_llm.py
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
@app.command("workflows")
def workflows_command(
    workflow_id: EntityId = None,
    dry_run: DryRun = False,
    force_index: ForceIndex = False,
    show_progress: ShowProgress = False,
) -> None:
    """Index the workflow search index.

    Args:
        workflow_id: UUID of a single workflow to index, `None` indexes all.
        dry_run: Make no database writes and no embedding calls.
        force_index: Re-index every field, ignoring the content hashes.
        show_progress: Show per-entity progress.

    CLI Options:
        ```shell
        Options:
            --workflow-id <str>                   UUID (default = all)
            --dry-run / --no-dry-run              No DB writes  [default: no-dry-run]
            --force-index / --no-force-index      Force re-index (ignore hash cache)  [default: no-force-index]
            --show-progress / --no-show-progress  Show per-entity progress  [default: no-show-progress]
            --help                                Show this message and exit.
        ```
    """
    run_indexing_for_entity(
        entity_kind=EntityType.WORKFLOW,
        entity_id=workflow_id,
        dry_run=dry_run,
        force_index=force_index,
        show_progress=show_progress,
    )

product-blocks

Index the product block search index.

CLI Options
Options:
    --product-block-id <str>              UUID (default = all)
    --dry-run / --no-dry-run              No DB writes  [default: no-dry-run]
    --force-index / --no-force-index      Force re-index (ignore hash cache)  [default: no-force-index]
    --show-progress / --no-show-progress  Show per-entity progress  [default: no-show-progress]
    --help                                Show this message and exit.
Source code in orchestrator-core/orchestrator/core/cli/search/index_llm.py
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
@app.command("product-blocks")
def product_blocks_command(
    product_block_id: EntityId = None,
    dry_run: DryRun = False,
    force_index: ForceIndex = False,
    show_progress: ShowProgress = False,
) -> None:
    """Index the product block search index.

    Args:
        product_block_id: UUID of a single product block to index, `None` indexes all.
        dry_run: Make no database writes and no embedding calls.
        force_index: Re-index every field, ignoring the content hashes.
        show_progress: Show per-entity progress.

    CLI Options:
        ```shell
        Options:
            --product-block-id <str>              UUID (default = all)
            --dry-run / --no-dry-run              No DB writes  [default: no-dry-run]
            --force-index / --no-force-index      Force re-index (ignore hash cache)  [default: no-force-index]
            --show-progress / --no-show-progress  Show per-entity progress  [default: no-show-progress]
            --help                                Show this message and exit.
        ```
    """
    run_indexing_for_entity(
        entity_kind=EntityType.PRODUCT_BLOCK,
        entity_id=product_block_id,
        dry_run=dry_run,
        force_index=force_index,
        show_progress=show_progress,
    )

resource-types

Index the resource type search index.

CLI Options
Options:
    --resource-type-id <str>              UUID (default = all)
    --dry-run / --no-dry-run              No DB writes  [default: no-dry-run]
    --force-index / --no-force-index      Force re-index (ignore hash cache)  [default: no-force-index]
    --show-progress / --no-show-progress  Show per-entity progress  [default: no-show-progress]
    --help                                Show this message and exit.
Source code in orchestrator-core/orchestrator/core/cli/search/index_llm.py
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
@app.command("resource-types")
def resource_types_command(
    resource_type_id: EntityId = None,
    dry_run: DryRun = False,
    force_index: ForceIndex = False,
    show_progress: ShowProgress = False,
) -> None:
    """Index the resource type search index.

    Args:
        resource_type_id: UUID of a single resource type to index, `None` indexes all.
        dry_run: Make no database writes and no embedding calls.
        force_index: Re-index every field, ignoring the content hashes.
        show_progress: Show per-entity progress.

    CLI Options:
        ```shell
        Options:
            --resource-type-id <str>              UUID (default = all)
            --dry-run / --no-dry-run              No DB writes  [default: no-dry-run]
            --force-index / --no-force-index      Force re-index (ignore hash cache)  [default: no-force-index]
            --show-progress / --no-show-progress  Show per-entity progress  [default: no-show-progress]
            --help                                Show this message and exit.
        ```
    """
    run_indexing_for_entity(
        entity_kind=EntityType.RESOURCE_TYPE,
        entity_id=resource_type_id,
        dry_run=dry_run,
        force_index=force_index,
        show_progress=show_progress,
    )

all

Index all entity types and rebuild the ai_search_paths table.

CLI Options
Options:
    --force-index / --no-force-index  Force re-index (ignore hash cache)  [default: no-force-index]
    --help                            Show this message and exit.
Source code in orchestrator-core/orchestrator/core/cli/search/index_llm.py
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
@app.command("all")
def all_command(force_index: ForceIndex = False) -> None:
    """Index all entity types and rebuild the ai_search_paths table.

    Args:
        force_index: Re-index every field, ignoring the content hashes.

    CLI Options:
        ```shell
        Options:
            --force-index / --no-force-index  Force re-index (ignore hash cache)  [default: no-force-index]
            --help                            Show this message and exit.
        ```
    """
    run_indexing_for_all_entities(force_index=force_index)

rebuild-paths

Recompute the ai_search_paths distinct-paths table from ai_search_index.

CLI Options
Options:
    --help  Show this message and exit.
Source code in orchestrator-core/orchestrator/core/cli/search/index_llm.py
257
258
259
260
261
262
263
264
265
266
267
@app.command("rebuild-paths")
def rebuild_paths_command() -> None:
    """Recompute the ai_search_paths distinct-paths table from ai_search_index.

    CLI Options:
        ```shell
        Options:
            --help  Show this message and exit.
        ```
    """
    rebuild_search_paths()

scheduler

Commands to interact with the scheduler and scheduled jobs.

run

Starts the scheduler in the foreground.

While running, this process will:

  • Periodically wake up when the next schedule is due for execution, and run it
  • Process schedule changes made through the schedule API
Source code in orchestrator-core/orchestrator/core/cli/scheduler.py
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@app.command()
def run() -> None:
    """Starts the scheduler in the foreground.

    While running, this process will:

      * Periodically wake up when the next schedule is due for execution, and run it
      * Process schedule changes made through the schedule API
    """

    def _get_scheduled_task_item_from_queue(redis_conn: Redis) -> tuple[str, bytes] | None:
        """Get an item from the Redis Queue for scheduler tasks."""
        try:
            return redis_conn.brpop(SCHEDULER_QUEUE, timeout=1)
        except ConnectionError as e:
            typer.echo(f"There was a connection error with Redis. Retrying in 3 seconds... {e}")
            time.sleep(3)
        except Exception as e:
            typer.echo(f"There was an unexpected error with Redis. Retrying in 1 second... {e}")
            time.sleep(1)

        return None

    with get_scheduler() as scheduler_connection:
        redis_connection = create_redis_client(app_settings.CACHE_URI.get_secret_value())
        while True:
            item = _get_scheduled_task_item_from_queue(redis_connection)
            if not item:
                continue

            with db.database_scope():
                workflow_scheduler_queue(item, scheduler_connection)

force

Force the execution of (a) scheduler(s) based on a schedule ID.

Use the show-schedule command to determine the ID of the schedule to execute.

CLI Arguments
Arguments:
    SCHEDULE_ID  ID of the schedule to execute
Source code in orchestrator-core/orchestrator/core/cli/scheduler.py
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
@app.command()
def force(task_id: str) -> None:
    """Force the execution of (a) scheduler(s) based on a schedule ID.

    Use the `show-schedule` command to determine the ID of the schedule to execute.

    CLI Arguments:
        ```shell
        Arguments:
            SCHEDULE_ID  ID of the schedule to execute
        ```
    """
    task = get_scheduler_task(task_id)

    if not task:
        typer.echo(f"Task '{task_id}' not found.")
        raise typer.Exit(code=1)

    typer.echo(f"Running Task [{task.id}] now...")
    try:
        task.func(*task.args or (), **task.kwargs or {})
        typer.echo("Task executed successfully.")
    except Exception as e:
        typer.echo(f"Task execution failed: {e}")
        raise typer.Exit(code=1)

show_schedule

The show-schedule command shows an overview of the scheduled jobs.

Source code in orchestrator-core/orchestrator/core/cli/scheduler.py
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@app.command()
def show_schedule(
    verbose: bool = typer.Option(False, "-v", help="Show additional details, such as job kwargs"),
) -> None:
    """The `show-schedule` command shows an overview of the scheduled jobs."""
    from rich.console import Console
    from rich.table import Table

    from orchestrator.core.schedules.service import get_linker_entries_by_schedule_ids

    console = Console()

    table = Table(title="Scheduled Tasks", show_lines=verbose)
    table.add_column("id", no_wrap=True)
    table.add_column("name")
    table.add_column("source")
    table.add_column("next run time")
    table.add_column("trigger")
    if verbose:
        table.add_column("kwargs")

    scheduled_tasks = get_all_scheduler_tasks()
    _schedule_ids = [task.id for task in scheduled_tasks]
    api_managed = {str(i.schedule_id) for i in get_linker_entries_by_schedule_ids(_schedule_ids)}

    # Sort by next run time, then by name
    scheduled_tasks.sort(key=lambda x: (x.next_run_time, x.name))

    for task in scheduled_tasks:
        table.add_row(*_to_schedule_row(task, api_managed_ids=api_managed, verbose=verbose))

    console.print(table)

load_initial_schedule

The load-initial-schedule command loads the initial schedule using the scheduler API.

The initial schedules are
  • Task Resume Workflows
  • Task Clean Up Tasks
  • Task Validate Subscriptions
  • Task Validate Products
  • Task Validate Awaiting Callbacks

By default, this command is idempotent since v4.7.1 when the scheduler is running. The schedules are only created when they do not already exist in the database. This behavior can be altered through the –recreate option.

Source code in orchestrator-core/orchestrator/core/cli/scheduler.py
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
@app.command()
def load_initial_schedule(
    recreate: bool = typer.Option(False, help="Whether to delete any existing schedules before creating"),
) -> None:
    """The `load-initial-schedule` command loads the initial schedule using the scheduler API.

    The initial schedules are:
      - Task Resume Workflows
      - Task Clean Up Tasks
      - Task Validate Subscriptions
      - Task Validate Products
      - Task Validate Awaiting Callbacks

    By default, this command is idempotent since v4.7.1 when the scheduler is running.
    The schedules are only created when they do not already exist in the database.
    This behavior can be altered through the --recreate option.
    """
    for workflow_name in load_schedules(INITIAL_SCHEDULES, recreate=recreate):
        typer.echo(f"Workflow '{workflow_name}' not found. Skipping schedule.")

To register your own project’s schedules from code, see Registering your own schedules from code.

load_schedules

Register schedules declared in code, resolving each workflow by name.

This is the entry point core’s own scheduler load-initial-schedule uses, and the supported way for a project to register its schedules at deploy time, where the REST API is not reachable. See :func:add_unique_scheduled_task_to_queue for the idempotency caveat.

Parameters:

  • schedules (collections.abc.Sequence[dict[str, typing.Any]]) –

    Dicts of :class:APSchedulerJobCreate fields minus workflow_id, which is resolved from workflow_name.

  • recreate (bool, default: False ) –

    Whether to delete existing schedule(s) for each workflow first.

Returns:

  • list[str] –

    The workflow_name of each schedule that was skipped because no such workflow is

  • list[str] –

    registered. Raise on a non-empty list to fail a deploy on an unmigrated workflow.

Raises:

  • ValueError –

    A schedule’s workflow_name is missing or not a string, or two schedules name the same workflow — malformed declarations rather than unknown workflows. Uncaught, this exits the CLI non-zero with a traceback.

  • ValidationError –

    A schedule’s remaining fields do not build an :class:APSchedulerJobCreate, for example trigger_kwargs the trigger rejects.

Every schedule is validated before any of them is queued, so malformed input registers nothing.

Source code in orchestrator-core/orchestrator/core/schedules/service.py
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def load_schedules(schedules: Sequence[dict[str, Any]], *, recreate: bool = False) -> list[str]:
    """Register schedules declared in code, resolving each workflow by name.

    This is the entry point core's own ``scheduler load-initial-schedule`` uses, and the supported
    way for a project to register its schedules at deploy time, where the REST API is not reachable.
    See :func:`add_unique_scheduled_task_to_queue` for the idempotency caveat.

    Args:
        schedules: Dicts of :class:`APSchedulerJobCreate` fields minus ``workflow_id``, which is
            resolved from ``workflow_name``.
        recreate: Whether to delete existing schedule(s) for each workflow first.

    Returns:
        The ``workflow_name`` of each schedule that was skipped because no such workflow is
        registered. Raise on a non-empty list to fail a deploy on an unmigrated workflow.

    Raises:
        ValueError: A schedule's ``workflow_name`` is missing or not a string, or two schedules name
            the same workflow — malformed declarations rather than unknown workflows. Uncaught, this
            exits the CLI non-zero with a traceback.
        ValidationError: A schedule's remaining fields do not build an :class:`APSchedulerJobCreate`,
            for example ``trigger_kwargs`` the trigger rejects.

    Every schedule is validated before any of them is queued, so malformed input registers nothing.
    """

    def workflow_name_of(schedule: dict[str, Any]) -> str:
        """Return the schedule's workflow name, rejecting one that is missing or not a string."""
        workflow_name = schedule["workflow_name"]
        if not isinstance(workflow_name, str) or not workflow_name:
            raise ValueError(
                f"Schedule {schedule.get('name', schedule)!r} has no valid workflow_name: {workflow_name!r}"
            )
        return workflow_name

    def to_payload(schedule: dict[str, Any], workflow_name: str) -> APSchedulerJobCreate | None:
        """Build the payload for one schedule, or None when its workflow is unknown."""
        workflow = get_workflow_by_name(workflow_name)
        if not workflow:
            logger.warning("Skipping schedule for unknown workflow", workflow_name=workflow_name)
            return None
        return APSchedulerJobCreate(**(schedule | {"workflow_id": workflow.workflow_id}))

    workflow_names = [workflow_name_of(schedule) for schedule in schedules]
    if duplicates := [name for name, count in Counter(workflow_names).items() if count > 1]:
        raise ValueError(
            f"Multiple schedules declared for workflow(s): {duplicates}. "
            "load_schedules registers at most one schedule per workflow; use the API for more."
        )

    payloads = [to_payload(schedule, name) for schedule, name in zip(schedules, workflow_names)]
    for payload in filter(None, payloads):
        logger.info("Loading schedule", payload=payload)
        add_unique_scheduled_task_to_queue(payload, recreate=recreate)

    return [name for name, payload in zip(workflow_names, payloads) if payload is None]