Instruction file imported from OHDSI/Data2Evidence (.github/instructions/flows.instructions.md). Copyright stays with the author.

D2E Flows Development Instructions

Quick Plugin Detection

Auto-detect from file path: plugins/flows/[plugin_group]/[individual_plugin]/ Plugin groups: base, data_management, data_transformation, hades, loyalty_score, search_embedding, i2b2, cohort_survival

Key indicators:

• flow.py + __init__.py in individual plugin folders
• Dockerfile + package.json in plugin group folders
• Dependencies: pyproject.toml + pixi.lock per group (pixi: conda-forge for python/R/Java/toolchain, PyPI via [project.dependencies]); renv.lock for R packages, restored by the group's setup_r.sh

Essential Development Workflow

1. Code Development

Always use Prefect patterns with proper type hints and logging:

from prefect import flow, task
import logging

logger = logging.getLogger(__name__)

@task
def hello_task(name: str) -> str:
    """Process name input with error handling and logging."""
    logger.info("Processing task", extra={"name": name})
    if not name:
        logger.warning("Empty name provided, using default")
        return "Hello Guest!"
    return f"Hello {name}!"

@flow
def hello_flow(name: str = "World") -> dict:
    """Main flow with structured logging and error handling."""
    logger.info("Starting hello flow", extra={"input_name": name})
    try:
        message = hello_task(name)
        logger.info("Flow completed successfully")
        return {"status": "success", "message": message}
    except Exception as e:
        logger.error("Flow failed", extra={"error": str(e)})
        return {"status": "error", "message": "Processing failed"}

CRITICAL: Always check utility files used by flow.py:

• types.py - Parameter validation, type definitions, conditional field requirements
• README.md - Plugin documentation, parameter examples, usage instructions
• init.py - Required for Python package structure
• Shared utilities - Check _shared_flow_utils imports for APIs, DAOs, types

Before modifying any flow:

1. Read types.py first - Understand parameter structure, validation rules, conditional requirements
2. Check imports - Identify all dependencies and shared utilities being used
3. Review README.md - Understand the plugin's purpose and parameter examples
4. Test parameter validation - Ensure your changes don't break existing validation logic

2. Package.json Generation

cd plugins/flows
python flowinit.py [package_name] [entrypoint] [plugin_type] [-dm]

Required: package_name, entrypoint (e.g. path/to/flow.py:function), plugin_type Optional: -dm (comma-separated datamodels) Plugin types: Use 'datamodel' for datamodel plugins, or a descriptive type like 'phenotype', 'test', 'analysis', etc.

Note: If you encounter ModuleNotFoundError: No module named 'fastapi', install it with:

pip install fastapi uvicorn --force-reinstall

or for uv projects:

uv add fastapi uvicorn --force

3. Build & Test

cd plugins/flows/[plugin_group]
yarn build  # Builds Docker image locally with local tag

Integration testing with D2E:

1. Directly modify docker-compose-local.yml:

   • Uncomment: - ./plugins/flows/[plugin_group]/package.json:/usr/src/plugins/d2e-flows/package.json
   • Plugins are baked into the trex image; no env toggle is required to pick up new flow versions — rebuild the image.

2. Restart trex service:

   cd ../../..  # Navigate to D2E root folder from plugins/flows/[plugin_group]
   # Restart only trex using yarn (recommended)
   yarn local --services trex start

3. Check Admin Portal > Jobs for your deployment

Plugin Architecture Patterns

Directory Structure

plugins/flows/[plugin_group]/
├── Dockerfile                   # Uses requirements_[name].txt OR pyproject.toml
├── package.json                 # Generated by flowinit.py
├── requirements_[name].txt      # Legacy deps (most plugins)
├── pyproject.toml + uv.lock     # Modern deps (loyalty_score, search_embedding, i2b2)
└── [individual_plugin]/
    ├── __init__.py
    ├── flow.py                  # Main entry point
    ├── types.py                 # Type definitions
    └── README.md

Dependency Detection Patterns

All groups are pixi projects: pyproject.toml (+ committed pixi.lock). After editing dependencies run pixi lock in the group dir and commit both files — CI (_pixi-lock-check.yml) fails stale locks and any compiled pypi sdist. R packages stay in renv.lock (setup-r task); binary assets are fetched by the setup-assets task. Flow runs execute on the process worker in the group's env via /app/run-flow.sh <npm-short-name>; HANA support is a pixi hana environment provisioned at runtime (never shipped).

Testing

# test_flow.py
from prefect.testing.utilities import prefect_test_harness
from .flow import hello_flow

def test_flow():
    with prefect_test_harness():
        result = hello_flow("test")
        assert result["status"] == "success"

Run: pytest test_flow.py

Common Development Tasks

Creating new flow:

1. Create types.py with Pydantic models and validation
2. Create @flow function → Break into @task functions
3. Create README.md with parameter documentation
4. Run flowinit.py → yarn build

Modifying existing flow:

1. Read types.py first - Understand current parameter structure
2. Check all imports - Identify dependencies that might be affected
3. Update logic → Update types.py if parameters change → Only regenerate package.json for your specific plugin if entry points changed → Rebuild → Test locally

This instruction file prioritizes practical development workflows with accurate build processes and plugin detection patterns.