Custom Strategies

This document describes how AIKP authors can override the standard strategy behavior.

There are two supported approaches:

1. TaskTemplate method override (recommended)

2. Module override via RECIPE_DEFAULTS (legacy)

Vector strategy examples and the overall workflow are described in ocli/ai/vector_strategies/README.md.

Recommended: TaskTemplate method override

Implement the strategy methods directly on your AIKP’s TaskTemplate class (typically in template.py). When present, OCLI will call these methods instead of the built-in defaults.

Typical methods to override:

• makegeojson(cls, task, cos_key=None, friendly_name=None)

• upload(cls, task, dry_run=False, cos_key=None)

• publish(cls, task, dry_run=False, cos_key=None)

Minimal skeleton:

from ocli.classes.task_template import TaskTemplate

class Template(TaskTemplate):
    @classmethod
    def makegeojson(cls, task, cos_key=None, friendly_name=None):
        ...

    @classmethod
    def upload(cls, task, dry_run=False, cos_key=None):
        ...

    @classmethod
    def publish(cls, task, dry_run=False, cos_key=None):
        ...

Important: where CLI options come from

For TaskTemplate overrides, OCLI usually calls your method like:

• Template.makegeojson(task)

• Template.upload(task)

• Template.publish(task)

That means runtime flags (for example COS key override, friendly name override, dry-run, print-json) are typically passed via Click context meta, not as Python arguments.

Use:

import click

ctx = click.get_current_context()
meta = ctx.meta

Step-by-step implementation guide

The recommended mental model is a 3-step pipeline:

1. makegeojson: create a DB-facing metadata document

2. upload: upload payload + metadata to COS

3. publish: post the metadata document to docs DB

1) makegeojson: generate the metadata document

Typical responsibilities:

• Load and (optionally) validate the recipe: task.get_recipe().

• Compute ResultKey (COS key) and friendly_name.

  • Respect CLI overrides stored in ctx.meta.

  • Convention: if an override starts with +, treat it as a suffix (append to the recipe-derived default instead of replacing it).

• Write the DB-facing metadata document to disk (commonly something like *_db.geojson).

Implementation sketch:

import click
from pathlib import Path
from monitored_ai_schema import Document # some more constructors
class Template(TaskTemplate):
    @classmethod
    def makegeojson(cls, task, cos_key=None, friendly_name=None):
        ctx = click.get_current_context()
        meta = ctx.meta

        recipe = task.get_recipe()
        task_cfg = task.config

        # Prefer values from click context meta
        cos_key = meta.get("cos_key")
        friendly_name = meta.get("friendly_name")

        # Compute defaults from recipe/task config
        default_result_key = recipe.get("COS", {}).get("ResultKey")
        default_friendly_name = recipe.get("friendly_name")

        # Apply +suffix convention
        if cos_key and cos_key.startswith("+"):
            result_key = f"{default_result_key}{cos_key[1:]}"
        else:
            result_key = cos_key or default_result_key

        if friendly_name and friendly_name.startswith("+"):
            final_friendly_name = f"{default_friendly_name}{friendly_name[1:]}"
        else:
            final_friendly_name = friendly_name or default_friendly_name

        # Write metadata document
        out_dir = Path(task.get_ai_results_path(full=True))
        doc_path = out_dir / "output_db.geojson"  # choose a stable naming convention
        doc_path.write_text(Document(
            # ... pass props
        ).model_dump_json(indent=2, by_alias=True))

2) upload: upload payload + metadata to COS

Typical responsibilities:

• Create a COS client from recipe credentials.

• Decide which files to upload (payload + the metadata document you generated in makegeojson).

• Decide the destination key prefix:

  • Prefer explicit override (from Click meta) when provided.

  • Otherwise derive it from the metadata document’s properties.ResultKey.

• Respect dry-run mode.

Implementation sketch:

import click

class Template(TaskTemplate):
    @classmethod
    def upload(cls, task, dry_run=False, cos_key=None):
        ctx = click.get_current_context()
        meta = ctx.meta

        dry_run = bool(meta.get("dry_run"))
        cos_key = meta.get("cos_key")

        recipe = task.get_recipe()

        # 1) create COS client from recipe["COS"]
        cos = COS(recipe.COS.model_dump())
        if cos_key:
            # Apply +suffix convention
            if cos_key and cos_key.startswith("+"):
                s3_dir = f"{default_result_key}{cos_key[1:]}"
            else:
                s3_dir = cos_key or default_result_key
            output.info(f"Changing cos_key to {cos_key}")
            s3_dir = Path(cos_key)
        else:
            s3_dir = Path(geojson_doc.properties.ResultKey)
        # 2) choose files to upload
        glb_path = recipe.meta.ai_results / GLB_FILE
        # 3) resolve destination keys
        s3_path = s3_dir / GLB_FILE
        # 4) upload (or log actions if dry_run)
        filesize = os.stat(glb_path).st_size
        if dry_run:
            output.info(f"File {glb_path} would be uploaded")
        else:
            with tqdm(total=filesize, unit='B', unit_scale=True, desc=cos_key) as t:
                cos.upload_to_cos(str(glb_path), str(s3_path), hook(t))

3) publish: post metadata document to docs DB

Typical responsibilities:

• Choose the metadata document path written by makegeojson.

• Call publish_json() with the correct docs DB from the recipe.

• Respect --dry-run and --print (print-json) flags if you support them.

Implementation sketch:

import click
import ocli.core.constants as gc
from ocli.cli.publish import publish_json

class Template(TaskTemplate):
    @classmethod
    def publish(cls, task, dry_run=False, cos_key=None):
        ctx = click.get_current_context()
        meta = ctx.meta

        dry_run = bool(meta.get("dry_run"))
        print_json = bool(meta.get("print_json"))

        recipe = task.get_recipe()

        doc_json_path = "...path to your *_db.geojson..."
        publish_json(doc_json_path, docs_db=recipe.get(gc.APP_DOCS_DB), dry_run=dry_run, print_json=print_json)

Practical checklist

• makegeojson writes a publishable metadata document to disk.

• upload uploads the payload and that metadata document.

• publish posts that same metadata document with publish post semantics.

Legacy(DEPRECATED!!!): Module override via RECIPE_DEFAULTS

Older templates can override strategies by mapping command names to module names in RECIPE_DEFAULTS in the AIKP package, and providing those modules with an execute() function.

Example mapping:

RECIPE_DEFAULTS = {
    "makegeojson": "make_geo_json",  # corresponds to make_geo_json.py
    "upload": "upload",              # corresponds to upload.py
    "publish": "publish",            # corresponds to publish.py
}

Then create the modules in your AIKP package.

make_geo_json.py

from ocli.cli.state import pass_task

@pass_task
def execute(task):
    # Generate metadata document(s)
    ...

upload.py

from ocli.cli.state import pass_task

@pass_task
def execute(task, dry_run=False, cos_key=None):
    # Upload artifacts + metadata to COS
    ...

publish.py

from ocli.cli.state import pass_task

@pass_task
def execute(task, dry_run=False, print_json=False):
    # Publish metadata document to docs DB
    ...

Notes:

• This approach is considered legacy; prefer TaskTemplate overrides for new AIKPs.

• Keep behavior consistent with the standard pipeline: makegeojson produces the publishable document, upload transfers it to COS, and publish posts it to docs DB.